"""End-user identity from the **Ignite gateway** — the app ships no auth code.
The app is deployed with a dodil-appid user pool attached:
dodil ignite app update <app> --user-pool <org>/<pool>
From that moment the **per-cluster Ignite gateway** does ALL end-user authentication at the
edge (Ignite v1.0.0, 2026-09-04):
· it verifies the pool's EdDSA JWT (`iss`-selected trust anchors resolved from the control
plane — no appid client or secret at the edge),
· it runs the **full browser login** itself — PKCE S256 against the pool issuer, AEAD-sealed
host-only session cookies, single-flight refresh — on the reserved `/.dodil/auth/*` paths,
· and it injects the verified identity into every request it forwards to this pod:
X-Dodil-User {"sub","email","connection","app_roles"} (plain JSON)
X-Dodil-User-Jwt the raw verified pool JWT (also carries `permissions`)
X-Dodil-Auth-Source "pool" (an app end-user) | "platform" (an operator/SA invoke)
· and it **always strips any inbound copy** of those headers first, on every mode and every
principal, so a caller can never forge them. The pod is only reachable through that ingress.
So this module is a *header-trust* reader, not a verifier. **We do NOT re-verify the JWT
signature** — the gateway already did, against trust anchors we do not hold. Re-doing it here
would mean shipping a JWKS client, an issuer/audience config and a crypto dependency for a
check that has already passed at the edge. What remains for the app is ROLE-BASED gating:
`require_permission(...)` on the routes that move money or override a forecast.
TWO-PLANE RULE (unchanged): this only identifies the USER. The app still reaches DataK3 via
its own service account (see db.py); an app-user is never a bucket principal.
LOCAL DEV (no gateway in front of `uvicorn routes:app`): set `DEV_ALLOW_ANON=1` and every
request is attributed to a stub developer identity. It is **opt-in only** — with the env unset
(the deployed default) a request with no `X-Dodil-User` is rejected 401, so the fallback can
never quietly open the deployed app.
"""
import base64
import json
import logging
import os
import re
from typing import Optional
from fastapi import Depends, HTTPException, Request
# ── what the gateway injects (lower-case: FastAPI/Starlette headers are case-insensitive) ──
USER_HEADER = "x-dodil-user" # JSON: sub / email / connection / app_roles
ACTING_FOR_HEADER = "x-dodil-acting-for" # a service reading for a person (erp_platform.seam), tiers 2–3
ACT_FOR = "erp:seam:act_for"
USER_JWT_HEADER = "x-dodil-user-jwt" # the raw pool JWT — carries the expanded `permissions`
AUTH_SOURCE_HEADER = "x-dodil-auth-source" # "pool" | "platform"
# ── local-dev fallback (opt-in; never set this in a deployed env) ───────────────────────────
_TRUE = {"1", "true", "yes", "on"}
DEV_ALLOW_ANON = os.environ.get("DEV_ALLOW_ANON", "").strip().lower() in _TRUE
DEV_USER_SUB = os.environ.get("DEV_USER_SUB", "dev-user")
DEV_USER_EMAIL = os.environ.get("DEV_USER_EMAIL", "dev@localhost")
# Comma-separated. Both default EMPTY: the stub is an ordinary signed-in user with no elevated
# rights, so the permission-gated routes stay gated even locally. To exercise them on a laptop:
# DEV_USER_PERMISSIONS=<module>:<object>:<verb>,... e.g. crm:quote:approve,gl:period:close
DEV_USER_ROLES = [r for r in os.environ.get("DEV_USER_ROLES", "").split(",") if r.strip()]
DEV_USER_PERMISSIONS = [p for p in os.environ.get("DEV_USER_PERMISSIONS", "").split(",") if p.strip()]
def _b64url_json(segment: str) -> dict:
"""Decode one base64url JWT segment to a dict. Never raises."""
try:
pad = "=" * (-len(segment) % 4)
return json.loads(base64.urlsafe_b64decode(segment + pad))
except Exception: # noqa: BLE001 — a malformed segment simply yields no extra claims
return {}
def _jwt_claims(token: Optional[str]) -> dict:
"""Read the payload of the gateway-verified pool JWT — **decode only, no verification**.
The signature was already checked at the edge (EdDSA, `iss`-selected trust anchors) and the
header cannot be forged by a caller (the gateway strips inbound copies). We only want the
claims the convenience header omits: `permissions` (expanded from the pool's role catalog at
mint time), plus `tenant`/`tenants` if a pool uses them.
"""
if not token:
return {}
parts = token.split(".")
return _b64url_json(parts[1]) if len(parts) == 3 else {}
def _identity(sub: str, email: str, roles: list, permissions: list, source: str,
connection: Optional[str] = None, tenant: Optional[str] = None) -> dict:
return {"sub": sub, "email": email or "", "roles": list(roles or []),
"permissions": list(permissions or []), "connection": connection,
"tenant": tenant, "auth_source": source}
def current_user(request: Request) -> dict:
"""FastAPI dependency: `user = Depends(current_user)`.
Yields the business user's identity for row-level / action authz, built from the
gateway-injected headers. Same name and same shape as the old JWKS-verify version —
routes are unchanged.
"""
return identity_from_headers(request.headers)
def identity_from_headers(headers) -> dict:
"""The caller's identity from the gateway's headers — ONE parser for every front door.
An HTTP route (via `current_user`) and an MCP tool call both come through here, so the two
cannot disagree about who the caller is. A brick that parsed identity twice would eventually
parse it two ways, and an agent would see rows its person cannot.
Trusting headers is safe ONLY because the gateway strips inbound copies and mints its own
from a verified token. The MCP SDK warns, correctly, that headers are client-supplied input;
here they are not, provided nothing serves these endpoints without the gateway in front.
"""
headers = {str(k).lower(): v for k, v in headers.items()}
raw = headers.get(USER_HEADER)
if raw:
try:
claims = json.loads(raw)
except Exception: # noqa: BLE001 — only the gateway writes this; malformed = not trusted
raise HTTPException(401, "malformed X-Dodil-User header")
sub = claims.get("sub")
if not sub:
raise HTTPException(401, "X-Dodil-User carries no sub")
# `permissions` is not in the convenience header — read it off the verified JWT.
jwt_claims = _jwt_claims(headers.get(USER_JWT_HEADER))
who = _identity(
sub=sub,
email=claims.get("email") or jwt_claims.get("email") or "",
roles=claims.get("app_roles") or jwt_claims.get("app_roles") or [],
permissions=jwt_claims.get("permissions") or [],
source=headers.get(AUTH_SOURCE_HEADER, "pool"),
connection=claims.get("connection") or jwt_claims.get("connection"),
tenant=jwt_claims.get("tenant"),
)
# 0.9.7, minted only by a seam in this process (the gateway strips inbound copies of both headers): which
# brick a service-client call comes from, and the calling brick's grant an onward read runs on.
if jwt_claims.get("service"):
who["service"] = str(jwt_claims["service"])
if jwt_claims.get("service_permissions"):
who["service_permissions"] = [str(p) for p in jwt_claims["service_permissions"]]
who["acting_service"] = str(jwt_claims.get("acting_service") or "")
return _acting_for(who, headers.get(ACTING_FOR_HEADER))
# No identity header. Deployed behind the gateway this means either an unauthenticated
# caller (the gateway 401s / 303s to `/.dodil/auth/login` before we ever see it) or a
# PLATFORM invoke (`x-dodil-auth-source: platform` — an operator or service account, which
# carries no end-user). Either way there is no business user to attribute the action to.
if DEV_ALLOW_ANON:
return _identity(DEV_USER_SUB, DEV_USER_EMAIL, DEV_USER_ROLES,
DEV_USER_PERMISSIONS, source="dev")
raise HTTPException(401, "end-user login required — no X-Dodil-User from the gateway "
"(deploy with `--user-pool <org>/<pool>`, or set DEV_ALLOW_ANON=1 "
"for local development)")
def _acting_for(who: dict, raw: Optional[str]) -> dict:
"""A service reading FOR a person, across apps (erp_platform.seam, `caller=`): honoured only when the
authenticated principal holds erp:seam:act_for — a permission granted to service clients and no person
role — so a person cannot borrow someone else's eyes by sending the header. The result is the PERSON's
sub, email and roles (their row scope) with NO permissions: what it is for is a read, and a permission
is not the service's to pass on. Otherwise the header is ignored and the caller is who they are."""
if not raw or ACT_FOR not in who["permissions"]:
return who
try:
person = json.loads(raw)
except Exception: # noqa: BLE001 — a malformed header from a trusted service: answer as the service
return who
if not isinstance(person, dict) or not person.get("sub"):
return who
out = _identity(sub=str(person["sub"]), email=str(person.get("email") or ""),
roles=[str(r) for r in (person.get("app_roles") or [])], permissions=[],
source="seam", connection="seam")
out["acting_service"] = who["sub"]
# 0.9.7: the service's own grant runs the door (may_run), never the row scope: the read is the service's to make,
# the rows are the person's to see.
out["service_permissions"] = [p for p in who["permissions"] if p != ACT_FOR]
return out
# ── role-scoped visibility (multi-user ownership) ───────────────────────────────────────
# Pool roles: `manager` (and `analyst`) see the whole book; `sales` sees only rows whose
# `owner` equals their identity email. A signed-in user with NO recognised role is treated
# as sales scoped to their own email — the safe default (never "no role = see everything").
SEE_ALL_ROLES = {"manager", "analyst"}
def _roles(user: dict) -> set:
return {str(r).strip().lower() for r in user.get("roles", []) if str(r).strip()}
def is_manager(user: dict) -> bool:
"""True for the `manager` app-role only — the reassignment gate. Deliberately an
explicit ROLE check, not a permission check: `deals:write`-style permissions say what
a rep may do to THEIR rows; changing who owns a row is a management act, so the role
is the gate (a pool that hands sales `deals:write` must not hand them reassignment)."""
return "manager" in _roles(user)
def owner_scope(user: dict, mine: bool = False, *, see_all: Optional[str] = None) -> Optional[str]:
"""Row-visibility scope for `owner`-carrying reads.
Returns None when the caller may see every row (manager/analyst), else the identity
email their reads must be filtered to. `mine=True` narrows ANY caller to their own
rows (the manager's "Mine" toggle rides on the same server-side filter the sales
scope uses — never a second code path)."""
email = (user.get("email") or "").strip() or None
if mine:
return email
if see_all is not None:
# 0.9.2: the way a brick decides — by PERMISSION (`<module>:<root>:read_all`). Roles are the customer's
# names; a brick never compares them. Given `see_all`, the roles are not read at all.
require_concrete(see_all, "owner_scope(see_all=)")
return None if has_permission(user, see_all) else email
if _roles(user) & SEE_ALL_ROLES: # pre-0.9.2 role names: erp-compose check warns on a brick relying on it
return None
return email
def resolve_owner(user: dict, requested: Optional[str]) -> Optional[str]:
"""The owner an API-created record lands with. Managers may assign any owner
explicitly; everyone else is FORCED-OWN — whatever `owner` the request carries, the
row belongs to the caller (documented choice: forced-own instead of 403, so seeded
UI forms and scripted imports keep working without a rejection path)."""
email = (user.get("email") or "").strip() or None
if is_manager(user):
return (requested or "").strip() or email
return email
def require_manager():
"""Dependency: gate a route on the `manager` app-role (owner reassignment)."""
def _dep(user: dict = Depends(current_user)) -> dict:
if not is_manager(user):
raise HTTPException(403, "manager role required")
return user
return _dep
def require_permission(perm: str):
"""Dependency factory: gate a route on a pool permission, e.g.
`Depends(require_permission("<module>:<object>:<verb>"))` — namespaced, so one
customer pool carries every module's roles without collision. The rule is `has_permission`'s;
a required permission is concrete, so one with a `*` is refused here, at import."""
require_concrete(perm, "require_permission")
def _dep(user: dict = Depends(current_user)) -> dict:
if not has_permission(user, perm):
raise HTTPException(403, f"missing permission: {perm}")
return user
return _dep
# ── permissions: `module:entity:action`, segment-wise wildcards, machine permissions ─────────
#
# A GRANT (what a pool role expands to in the JWT's `permissions`) is three segments, each `[a-z_]+` or `*`:
# `pj:item:read`, `pj:item:*`, `pj:*:read`, `pj:*:*`, `*:*:*`. No partial glob (`it*`), no other count, no
# digits or capitals: anything else is MALFORMED — ignored and logged once, never a crash, never a pass.
# A REQUIRED permission (what a use case or route names) is concrete: a `*` there is refused at registration.
# A MACHINE permission (MACHINE_PERMISSIONS + every Brick(machine_permissions=…)) is what only a service
# client holds — whoever holds it can act as something no person may be. It passes on an EXACT grant only:
# `*:*:*` reaches every person command and no machine one.
SEGMENT = re.compile(r"[a-z_]+")
WILDCARD = "*"
MACHINE_PERMISSIONS = frozenset({ACT_FOR}) # the platform's own; bricks add theirs (declare_machine)
_machine: set = set(MACHINE_PERMISSIONS)
_logged: set = set()
_log = logging.getLogger("erp_platform.auth")
def is_concrete(perm) -> bool:
"""`module:entity:action`, three `[a-z_]+` segments — what a use case may require."""
parts = perm.split(":") if isinstance(perm, str) else []
return len(parts) == 3 and all(SEGMENT.fullmatch(x) for x in parts)
def require_concrete(perm, where: str) -> None:
"""Refuse a required permission with a wildcard in it: it would name no one permission."""
if isinstance(perm, str) and WILDCARD in perm:
raise ValueError(f"{where}: a required permission is concrete — {perm!r} has a '*' (wildcards are for "
"grants: a role holds `pj:item:*`, a use case requires `pj:item:read`)")
def parse_grant(grant) -> Optional[tuple]:
"""A grant's three segments, or None if it is malformed (logged once per distinct grant)."""
parts = grant.split(":") if isinstance(grant, str) else []
if len(parts) == 3 and all(x == WILDCARD or SEGMENT.fullmatch(x) for x in parts):
return tuple(parts)
key = grant if isinstance(grant, str) else repr(type(grant))
if key not in _logged and len(_logged) < 1024: # once per distinct grant; a bounded memory
_logged.add(key)
_log.warning(json.dumps({"event": "malformed_permission_grant", "grant": key[:128],
"rule": "module:entity:action, each segment [a-z_]+ or *; ignored"}))
return None
def declare_machine(perms, where: str = "") -> None:
"""Mark permissions machine-only (Brick(machine_permissions=…) calls this): no wildcard reaches them."""
for p in perms:
if not is_concrete(p):
raise ValueError(f"{where or 'machine_permissions'}: {p!r} is not a concrete module:entity:action")
_machine.add(p)
def machine_permissions() -> frozenset:
"""Every machine permission known in this process: the platform's and every loaded brick's."""
return frozenset(_machine)
def is_machine(perm: str) -> bool:
return perm in _machine
def grants(user: dict) -> list:
"""The caller's grants as a list of strings — a `permissions` claim that is not a list is no grant at all
(a string would otherwise answer `in` by substring)."""
raw = (user or {}).get("permissions")
return [g for g in raw if isinstance(g, str)] if isinstance(raw, (list, tuple, set, frozenset)) else []
def grant_covers(grant, perm: str) -> bool:
"""Does one grant cover one concrete, non-machine permission? Segment by segment: equal, or `*`."""
parsed = parse_grant(grant)
return parsed is not None and all(g == WILDCARD or g == p for g, p in zip(parsed, perm.split(":")))
def has_permission(user: dict, perm: str) -> bool:
"""The one permission rule, shared by routes, MCP tools and the probes.
A required permission that is not concrete (a scaffold's `TODO(permission):…`) passes no one. A machine
permission passes only on an exact grant. Any other passes on a grant that covers it segment-wise.
The pre-0.9.2 fallback where a ROLE named like the permission passed is gone: no brick or composition
relied on it (searched 2026-10-18), and a pool hands permissions, not role names, in `permissions`."""
if not is_concrete(perm):
return False
held = grants(user)
if perm in _machine:
return perm in held
return any(grant_covers(g, perm) for g in held)
def may_run(user: dict, perm: str) -> bool:
"""May this caller run a use case needing `perm` — their own grant, or (0.9.7) the grant of the brick reading on
their behalf (`service_permissions`: an onward read through a seam with caller=). Row scope is never decided
here: owner_scope and has_permission read the person's own permissions only."""
if has_permission(user, perm):
return True
svc = user.get("service_permissions")
return bool(svc) and has_permission({"permissions": svc}, perm)
def effective(granted, catalog) -> list:
"""The concrete permissions of `catalog` that `granted` (a role's grants) reaches — what a role can run."""
held = {"permissions": list(granted)}
return sorted(p for p in set(catalog) if has_permission(held, p))