"""Personas: declared by ROLE in a scenario, resolved per runner.
THE ROLE CATALOGUE is the composition's file of pool roles → permissions — the same file the pool is seeded
from (`dodil appid roles set`), so what a scenario's clerk may do in process is what a real clerk may do:
# roles.yaml
roles:
clerk: [orders:order:write]
manager: [orders:order:write, orders:order:approve, orders:clock:set]
ticker: [orders:tick:run]
service_clients: [ticker] # machines: client_credentials, not a person's sign-in
A persona is a role, or "<role>#<n>" for another person holding the same role ("clerk#2" is not "clerk" — the
404 probes stand on that). "anonymous" is nobody.
IN PROCESS a persona is the gateway's headers for that role's permissions (seam.gateway_headers) — what the
Ignite gateway injects after it verified a token. LIVE it is a real token, obtained per persona from the
secrets directory (ERP_SCENARIO_SECRETS), whose `personas.json` names one of:
{"token_url": "https://appid.dodil.io/<org>/<pool>/token", # or "pool": "<org>/<pool>"
"personas": {"clerk": {"email": "...", "password": "..."}, # the appid password grant
"ticker": {"client_id": "...", "client_secret": "..."}}} # a service client
{"local_signing_key": "signing-key.pem", "issuer": "...", "audience": "..."} # a LOCAL issuer
# (scenarios.gateway) mints every persona
Every file read from that directory must be mode 600 (owner only) or it is refused; no value from it is
ever printed, logged or put in a report.
"""
import json
import os
import pathlib
import stat
import threading
import time
from dataclasses import dataclass, field
from typing import Optional
import yaml
ANONYMOUS = "anonymous"
EMAIL_DOMAIN = "scenario.test"
def split(persona: str) -> tuple:
role, _, n = persona.partition("#")
return role, int(n or 1)
@dataclass
class RoleCatalogue:
roles: dict # role → [permission, ...]
service_clients: frozenset = frozenset()
source: str = ""
# 0.9.2: personas a probe adds (a `*:*:*` grant, a malformed one) — in process only, never in `roles`
synthetic: dict = field(default_factory=dict)
def known(self, role: str) -> bool:
return role in self.roles or role in self.synthetic
@classmethod
def load(cls, path) -> "RoleCatalogue":
if isinstance(path, RoleCatalogue):
return path
if isinstance(path, dict):
spec, source = path, "(inline)"
else:
spec, source = yaml.safe_load(pathlib.Path(path).read_text()) or {}, str(path)
roles = {str(r): [str(p) for p in (perms or [])] for r, perms in (spec.get("roles") or {}).items()}
if not roles:
raise ValueError(f"{source}: a role catalogue needs roles: {{role: [permission, ...]}}")
if ANONYMOUS in roles:
raise ValueError(f"{source}: 'anonymous' is nobody, not a role")
services = frozenset(spec.get("service_clients") or [])
unknown = services - set(roles)
if unknown:
raise ValueError(f"{source}: service_clients {sorted(unknown)} are not roles")
return cls(roles, services, source)
def permissions(self, persona: str) -> list:
role, _ = split(persona)
return list(self.roles.get(role, self.synthetic.get(role, [])))
def holds(self, persona: str, permission: Optional[str]) -> bool:
if persona == ANONYMOUS:
return False
if permission is None:
return True
from erp_platform import auth # the runtime's rule, wildcards and machine permissions included
return auth.has_permission({"permissions": self.permissions(persona)}, permission)
def claims(self, persona: str) -> Optional[dict]:
"""The identity the gateway would mint for this persona in process; None for anonymous."""
if persona == ANONYMOUS:
return None
role, n = split(persona)
if not self.known(role):
raise KeyError(f"no role {role!r} in {self.source}")
email = f"{role}.{n}@{EMAIL_DOMAIN}"
return {"sub": f"{role}.{n}", "email": email, "connection": "service" if role in self.service_clients else "local",
"app_roles": [role], "permissions": self.permissions(persona)}
# ── live credentials ───────────────────────────────────────────────────────────────────────
class SecretsRefused(RuntimeError):
"""The secrets directory is missing, unreadable, or readable by someone other than its owner."""
def read_secret(path: pathlib.Path) -> str:
if not path.is_file():
raise SecretsRefused(f"{path}: no such file in the secrets directory")
mode = stat.S_IMODE(path.stat().st_mode)
if mode & 0o077:
raise SecretsRefused(f"{path}: mode {oct(mode)} lets others read it — chmod 600 {path.name}")
return path.read_text()
@dataclass
class LiveCredentials:
"""Tokens for personas, from ERP_SCENARIO_SECRETS. Values never leave this object except as a bearer."""
directory: pathlib.Path
roles: RoleCatalogue
_spec: dict = field(default_factory=dict, repr=False)
_cache: dict = field(default_factory=dict, repr=False)
_issuer: object = field(default=None, repr=False)
_lock: threading.Lock = field(default_factory=threading.Lock, repr=False)
@classmethod
def from_env(cls, roles: RoleCatalogue, directory: Optional[str] = None) -> "LiveCredentials":
raw = directory or os.environ.get("ERP_SCENARIO_SECRETS", "")
if not raw:
raise SecretsRefused("set ERP_SCENARIO_SECRETS to the directory holding personas.json (mode 600)")
d = pathlib.Path(raw).expanduser()
spec = json.loads(read_secret(d / "personas.json"))
creds = cls(d, roles, spec)
if spec.get("local_signing_key"):
from erp_platform.scenarios.gateway import LocalIssuer
creds._issuer = LocalIssuer.from_pem(read_secret(d / spec["local_signing_key"]).encode(),
issuer=spec["issuer"], audience=spec["audience"])
return creds
@property
def token_url(self) -> str:
if self._spec.get("token_url"):
return self._spec["token_url"]
pool = self._spec.get("pool", "")
from erp_platform.bricks import APPID_PUBLIC_URL
return f"{APPID_PUBLIC_URL}/{pool}/token"
def missing(self, personas) -> list:
"""The personas this directory cannot sign in."""
if self._issuer is not None:
return [p for p in personas if p != ANONYMOUS and split(p)[0] not in self.roles.roles]
known = self._spec.get("personas") or {}
return [p for p in personas if p != ANONYMOUS and p not in known]
def email(self, persona: str) -> str:
if self._issuer is not None:
return self.roles.claims(persona)["email"]
entry = (self._spec.get("personas") or {}).get(persona, {})
return entry.get("email") or entry.get("client_id", "")
def token(self, persona: str) -> str:
with self._lock:
hit = self._cache.get(persona)
if hit and hit[1] > time.time():
return hit[0]
value, ttl = self._fetch(persona)
self._cache[persona] = (value, time.time() + max(ttl - 60, 30))
return value
def _fetch(self, persona: str) -> tuple:
if self._issuer is not None:
return self._issuer.mint(self.roles.claims(persona)), 3600
entry = (self._spec.get("personas") or {}).get(persona)
if not entry:
raise SecretsRefused(f"no credentials for persona {persona!r} in {self.directory}/personas.json")
import httpx
if "password" in entry:
data = {"grant_type": "password", "email": entry["email"], "password": entry["password"]}
elif "client_secret" in entry:
data = {"grant_type": "client_credentials", "client_id": entry["client_id"],
"client_secret": entry["client_secret"]}
else:
raise SecretsRefused(f"persona {persona!r}: give email+password or client_id+client_secret")
r = httpx.post(self.token_url, data=data, timeout=30)
if r.status_code != 200:
# The status only: the body of a refused grant can echo what was sent.
raise SecretsRefused(f"persona {persona!r}: the token endpoint answered {r.status_code}")
body = r.json()
return body["access_token"], int(body.get("expires_in", 300))