diff --git a/scripts/iam/zitadel-provision.py b/scripts/iam/zitadel-provision.py index a5054cf..10d8c41 100755 --- a/scripts/iam/zitadel-provision.py +++ b/scripts/iam/zitadel-provision.py @@ -87,6 +87,12 @@ ROLES = { # LLM gate for the api-gw. The colon form matches the Authentik-era # permission string, so consumer manifests did not change. "llm:inference": "LLM inference access (api-gw)", + # Capability for machine-to-machine callers. Kept separate from the human roles + # so a service account can be identified by its grant alone. Note that ZITADEL + # does not assert roles into a machine-user access token on this instance (see + # MACHINE_CALLERS), so this grant documents intent; the gateway authorises the + # caller from its allow-list entry instead. + "svc-caller": "Machine-to-machine caller", "llm-inference": "LLM inference access (dash alias)", # Per-service capabilities. The gateway derives the required capability from # the verb: GET/HEAD/OPTIONS -> ":read", anything else -> @@ -345,6 +351,55 @@ def _ensure_builtin_login(token: str) -> None: print("login page : built-in (loginV2.required=false)") +# One machine user per cross-service caller. Each has its own key, so revoking one +# service leaves the others working, and ZITADEL can overlap keys during a rotation. +MACHINE_CALLERS = { + "poimen-llm": { + "name": "Poimen LLM caller", + "description": "poimen calls api-gateway inference", + "roles": ["llm:inference", "svc-caller"], + }, +} + + +def _ensure_machine_users(token: str, project_id: str) -> dict[str, dict]: + """Create each caller once and grant its roles. Returns id and key id per caller. + + Roles are granted even though ZITADEL does not assert them into a machine-user + token on this instance: the grant is the record of what the account may do, and + it becomes effective the moment that behaviour changes. + """ + out: dict[str, dict] = {} + for user_name, spec in MACHINE_CALLERS.items(): + code, found = _api("POST", "/management/v1/users/_search", token, + {"queries": [{"userNameQuery": {"userName": user_name, + "method": "TEXT_QUERY_METHOD_EQUALS"}}]}) + user_id = "" + for user in (found.get("result") or []): + user_id = user.get("id", "") + if not user_id: + code, created = _api("POST", "/management/v1/users/machine", token, { + "userName": user_name, + "name": spec["name"], + "description": spec["description"], + "accessTokenType": "ACCESS_TOKEN_TYPE_JWT", + }) + user_id = created.get("userId", "") + if not user_id: + continue + code, grant = _api("POST", "/management/v1/users/grants/_search", token, + {"query": {"limit": 100, "offset": 0}}) + existing = next((g for g in (grant.get("result") or []) if g.get("userId") == user_id), None) + if existing: + _api("PUT", f"/management/v1/users/{user_id}/grants/{existing['id']}", token, + {"roleKeys": spec["roles"]}) + else: + _api("POST", f"/management/v1/users/{user_id}/grants", token, + {"projectId": project_id, "roleKeys": spec["roles"]}) + out[user_name] = {"userId": user_id} + return out + + def _ensure_roles(token: str, project_id: str, roles: dict[str, str]) -> None: code, d = _api("POST", f"/management/v1/projects/{project_id}/roles/_search", token, {}) # ZITADEL answers with `key`, not `roleKey`. Reading the wrong field left @@ -502,6 +557,9 @@ def phase_apply(_args: argparse.Namespace) -> int: project_id = _ensure_project(token, PROJECT_NAME) print(f"project : {PROJECT_NAME} ({project_id})") _ensure_roles(token, project_id, ROLES) + machine_users = _ensure_machine_users(token, project_id) + for name, info in machine_users.items(): + print(f"machine user {name}: {info['userId']}") print(f"roles : {len(ROLES)} ensured") known: dict[str, str] = {} try: