feat(iam): a role and a machine user for each cross-service caller

Adds svc-caller, a capability that names a machine-to-machine caller, separate from
the human roles so a service account can be identified by its grant alone.

Adds MACHINE_CALLERS: one ZITADEL machine user per caller, created once with its own
key and granted its roles. poimen-llm is the first, granted llm:inference and
svc-caller. Its key is separate from every other caller's, so revoking one service
leaves the rest working and ZITADEL can overlap keys during a rotation.

The role is granted even though ZITADEL does not assert roles into a machine-user
access token on this instance. The grant is the record of what the account may do, and
it takes effect the moment that behaviour changes. Until then the gateway authorises
the caller from its allow-list entry.
This commit is contained in:
2026-10-27 23:40:54 -07:00
parent e599c12f91
commit bd81503e33
+58
View File
@@ -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 -> "<service>: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: