homelab/scripts/iam/zitadel-provision.py

#!/usr/bin/env python3
"""Provision ZITADEL end to end: machine key, Terraform, SOPS secrets, users.

One command replaces the manual sequence the Authentik script
(scripts/iam/provision-rbac.py) used to do in a single pass. Phases run in a
fixed order and every phase is safe to re-run:

  keys         extract the iam-admin machine key and PAT from namespace iam
  apply        create project, roles and OIDC apps through the ZITADEL API
  secrets      write generated client ids/secrets back into the SOPS files
  users        import one human user, converting the Django password hash
  verify       check discovery, client count and the groups-claim gate

`all` runs keys, apply, secrets, verify. It never runs users: the password hash
must be supplied by a human, one user at a time.

Nothing here talks to Authentik. Authentik keeps serving until a consumer is
moved deliberately.

Usage:
  scripts/iam/zitadel-provision.py keys
  scripts/iam/zitadel-provision.py apply --org-id <id>
  scripts/iam/zitadel-provision.py secrets --dry-run
  scripts/iam/zitadel-provision.py secrets --apply-config
  scripts/iam/zitadel-provision.py users --django-hash '<pbkdf2_sha256$...>'
  scripts/iam/zitadel-provision.py verify
"""

from __future__ import annotations

import argparse
import base64
import json
import os
import re
import signal
import stat
import subprocess
import sys
import tempfile
from pathlib import Path

import yaml

REPO = Path(__file__).resolve().parents[2]
SOPS_DIR = REPO / "k8s" / "argocd" / "secrets"
SOPS_CONFIG = REPO / ".sops.yaml"
CONF_DIR = Path(os.environ.get("ZITADEL_CONF_DIR", Path.home() / ".config" / "zitadel"))
STATE_FILE = CONF_DIR / "state.json"
NAMESPACE = os.environ.get("ZITADEL_NAMESPACE", "iam")
BASE_URL = os.environ.get("ZITADEL_BASE_URL", "https://zitadel.riotpiao.com")
# Notification delivery target: the in-cluster SMTP -> Gotify relay.
SMTP_HOST = os.environ.get("ZITADEL_SMTP_HOST", "smtp-gotify.iam.svc.cluster.local:2525")
SMTP_FROM = os.environ.get("ZITADEL_SMTP_FROM", "[email protected]")
SMTP_FROM_NAME = os.environ.get("ZITADEL_SMTP_FROM_NAME", "ZITADEL")

# ── desired ZITADEL state ──────────────────────────────────────────────────
# Declared here, applied through the API by the `apply` phase. This replaces the
# retired terraform/zitadel root module: same objects, no state file to drift and
# no second writer.
PROJECT_NAME = os.environ.get("ZITADEL_PROJECT_NAME", "homelab")
PROJECT_GROUP = "homelab"

# Role keys EQUAL the Authentik group names, so the `groups` claim keeps working
# for every consumer that maps on it (kube-apiserver RBAC, MinIO policies,
# Grafana roles).
ROLES = {
    "homelab-admins": "Cluster administrators with full access",
    "minio-admins": "MinIO administrators",
    "minio-photos": "Photos bucket (Immich) access",
    "minio-documents": "Documents bucket (Paperless) access",
    "minio-backups": "Backups bucket read-only access",
    "poimen-admins": "Poimen memory administrators",
    "poimen-devs": "Dev and staging projects access",
    "poimen-prod-readonly": "Production projects read-only access",
    "paperless-admins": "Paperless administrators",
    "paperless-finance": "Finance documents",
    "paperless-legal": "Legal documents",
    "paperless-hr": "HR documents",
    "grafana-admins": "Grafana administrators",
    "grafana-editors": "Grafana dashboard editors",
    "grafana-viewers": "Grafana dashboard viewers",
    "sqs-users": "SQS/Temporal queue read access",
    "sqs-writers": "SQS/Temporal queue read/write access",
    "s3-users": "S3 read access",
    "s3-writers": "S3 read/write access",
    # 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)",
    "llm-inference": "LLM inference access (dash alias)",
}

_WEB = {"app_type_id": "OIDC_APP_TYPE_WEB", "auth_method_id": "OIDC_AUTH_METHOD_TYPE_BASIC"}
_NATIVE = {"app_type_id": "OIDC_APP_TYPE_NATIVE", "auth_method_id": "OIDC_AUTH_METHOD_TYPE_NONE"}
_PKCE = {"app_type_id": "OIDC_APP_TYPE_USER_AGENT", "auth_method_id": "OIDC_AUTH_METHOD_TYPE_NONE"}

OIDC_APPS = {
    "api-gw": {**_WEB, "redirect_uris": [
        "http://localhost:3000/callback", "https://api.riotpiao.com/callback"]},
    "argocd": {**_WEB, "redirect_uris": [
        "http://localhost:8080/auth/callback",
        "https://argocd.riotpiao.com/auth/callback"]},
    # Forgejo derives the callback path from the login-source NAME
    # (/user/oauth2/<name>/callback). Both spellings are registered so a
    # rename between authentik and zitadel never breaks sign-in.
    "forgejo": {**_WEB, "redirect_uris": [
        "http://localhost:3000/user/oauth2/zitadel/callback",
        "https://forgejo.riotpiao.com/user/oauth2/zitadel/callback",
        "http://localhost:3000/user/oauth2/authentik/callback",
        "https://forgejo.riotpiao.com/user/oauth2/authentik/callback"]},
    "grafana": {**_WEB, "redirect_uris": [
        "http://localhost:3000/login/generic_oauth",
        "https://grafana.riotpiao.com/login/generic_oauth"]},
    "immich": {**_WEB, "app_type_id": "OIDC_APP_TYPE_NATIVE", "redirect_uris": [
        "app.immich:///oauth-callback", "https://img.riotpiao.com/auth/login",
        "https://img.riotpiao.com/user/oauth2/callback"]},
    "kubernetes": {**_PKCE, "dev_mode": True, "redirect_uris": [
        "http://localhost:8000", "http://localhost:18000"]},
    "minio": {**_WEB, "redirect_uris": [
        "http://localhost:9000/auth/sso/oauth2/code",
        "https://minio.riotpiao.com/auth/sso/oauth2/code"]},
    "paperless": {**_WEB, "redirect_uris": [
        "http://localhost:8000/accounts/oidc/zitadel/login/callback/",
        "https://paperless.riotpiao.com/accounts/oidc/zitadel/login/callback/",
        "http://localhost:8000/accounts/oidc/authentik/login/callback/",
        "https://paperless.riotpiao.com/accounts/oidc/authentik/login/callback/"]},
    "poimen": {**_WEB, "redirect_uris": [
        "http://localhost:3000/callback", "https://poimen.riotpiao.com/callback"]},
    "queue-sqs": {**_WEB, "redirect_uris": [
        "http://localhost:8080/callback", "https://queue.riotpiao.com/callback"]},
    "vault": {**_WEB, "redirect_uris": [
        "http://localhost:8200/ui/vault/auth/oidc/oidc/callback",
        "https://vault.riotpiao.com/ui/vault/auth/oidc/oidc/callback"]},
}


_TEMP_FILES: set[str] = set()


def _cleanup_temps(*_a) -> None:
    for path in list(_TEMP_FILES):
        try:
            os.unlink(path)
        except OSError:
            pass
        _TEMP_FILES.discard(path)


for _sig in (signal.SIGTERM, signal.SIGINT, signal.SIGHUP):
    signal.signal(_sig, lambda s, f: (_cleanup_temps(), sys.exit(130)))


def die(msg: str) -> None:
    _cleanup_temps()
    print(f"FATAL: {msg}", file=sys.stderr)
    sys.exit(1)


def run(cmd: list[str], **kw) -> subprocess.CompletedProcess:
    return subprocess.run(cmd, check=False, text=True, capture_output=True, **kw)


def need(binary: str) -> None:
    if subprocess.run(["sh", "-c", f"command -v {binary}"], capture_output=True).returncode != 0:
        die(f"{binary} is not on PATH")


def write_private(path: Path, data: bytes) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    try:
        os.fchmod(fd, 0o600)  # open() mode applies at creation only
        with os.fdopen(fd, "wb") as fh:
            fh.write(data)
    finally:
        pass


def secret_value(name: str, key: str) -> str:
    cp = run(["kubectl", "-n", NAMESPACE, "get", "secret", name, "-o", f"jsonpath={{{key}}}"])
    if cp.returncode != 0:
        die(f"cannot read secret {NAMESPACE}/{name}: {cp.stderr.strip()}")
    raw = cp.stdout.strip()
    if not raw:
        die(f"secret {NAMESPACE}/{name} has no key {key}")
    return base64.b64decode(raw).decode()


def load_env_file(path: Path = REPO / ".env") -> None:
    """Load KEY=VALUE lines from .env without overriding the real environment.

    The file is git-ignored and holds the bootstrap token a human pastes in.
    Sourcing it in the shell would work too, but then every phase depends on the
    caller remembering to do it.
    """
    if not path.exists():
        return
    for line in path.read_text().splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _, value = line.partition("=")
        key = key.strip()
        value = value.strip().strip('"').strip("'")
        if key and key not in os.environ:
            os.environ[key] = value


def bootstrap_token() -> str | None:
    """Resolve the iam-admin PAT: environment, then cached file, then cluster.

    Order matters. ZITADEL_BOOTSTRAP_TOKEN in .env is filled by a human after the
    first sync and wins. The cached file under ~/.config/zitadel keeps a run
    working when the API server is unreachable. The cluster Secret is the last
    resort and the only source that always reflects a rotation.
    """
    token = os.environ.get("ZITADEL_BOOTSTRAP_TOKEN", "").strip()
    if token:
        return token
    cached = CONF_DIR / "iam-admin.pat"
    if cached.exists():
        return cached.read_text().strip()
    try:
        return secret_value("iam-admin-pat", ".data.pat").strip()
    except SystemExit:
        return None


def curl_json(method: str, url: str, token: str | None = None, body: dict | None = None,
              timeout: int = 20) -> tuple[int, str]:
    """Call the API without putting a secret in the process argv.

    `curl --config -` reads the options from stdin, so neither the bearer token
    nor the password hash of a user import shows up in `ps`.
    """
    def esc(value: str) -> str:
        return value.replace("\\", "\\\\").replace('"', '\\"')

    cfg = [f'url = "{esc(url)}"', f'request = "{method}"', "silent", "show-error",
           f"max-time = {timeout}", 'write-out = "\\n%{http_code}"',
           'header = "Content-Type: application/json"']
    # Reaching the instance by name needs either public DNS through the tunnel
    # or a pinned address. ZITADEL picks the instance from the Host header, so a
    # plain port-forward fails with "Instance not found". Set this to bypass DNS:
    #   ZITADEL_CURL_RESOLVE=zitadel.riotpiao.com:443:192.168.1.160
    pinned = os.environ.get("ZITADEL_CURL_RESOLVE", "").strip()
    if pinned:
        cfg.append(f'resolve = "{esc(pinned)}"')
    if token:
        cfg.append(f'header = "Authorization: Bearer {esc(token)}"')
    if body is not None:
        cfg.append(f'data = "{esc(json.dumps(body))}"')
    cp = run(["curl", "--config", "-"], input="\n".join(cfg) + "\n")
    if cp.returncode != 0:
        die(f"{method} {url} failed: {cp.stderr.strip()}")
    payload, _, code = cp.stdout.rpartition("\n")
    return int(code.strip() or 0), payload


# ── phase: keys ────────────────────────────────────────────────────────────

def phase_keys(_args: argparse.Namespace) -> int:
    write_private(CONF_DIR / "iam-admin.json", secret_value("iam-admin", ".data.iam-admin\\.json").encode())
    write_private(CONF_DIR / "iam-admin.pat", secret_value("iam-admin-pat", ".data.pat").strip().encode())
    print(f"machine key  -> {CONF_DIR / 'iam-admin.json'} (0600)")
    print(f"PAT          -> {CONF_DIR / 'iam-admin.pat'} (0600)")
    print("both files stay outside the repo; never commit them")
    return 0


# ── phase: apply ───────────────────────────────────────────────────────────

def _api(method: str, path: str, token: str, body: dict | None = None,
         timeout: int = 30) -> tuple[int, dict]:
    code, payload = curl_json(method, f"{BASE_URL}{path}", token=token, body=body, timeout=timeout)
    try:
        return code, (json.loads(payload) if payload.strip() else {})
    except json.JSONDecodeError:
        return code, {"_raw": payload}


def _ensure_project(token: str, name: str) -> str:
    code, d = _api("POST", "/management/v1/projects/_search", token, {})
    project_id = ""
    for row in d.get("result") or []:
        if row.get("name") == name:
            project_id = row["id"]
            break
    if not project_id:
        code, d = _api("POST", "/management/v1/projects", token, {"name": name})
        if code not in (200, 201) or not d.get("id"):
            die(f"create project {name} failed: {code} {d}")
        project_id = d["id"]

    # projectRoleAssertion puts the project roles into the ACCESS token as
    # `urn:zitadel:iam:org:project:roles`. Without it the api-gw sees no
    # llm:inference role and answers 403 for every inference call, even when
    # the caller is correctly authorised in ZITADEL.
    code, d = _api("PUT", f"/management/v1/projects/{project_id}",
                   token, {"name": name, "projectRoleAssertion": True})
    if code not in (200, 201):
        die(f"enable projectRoleAssertion on {name} failed: {code} {d}")
    return project_id


def _ensure_builtin_login(token: str) -> None:
    """Force the built-in login page instead of the hosted Login UI.

    The hosted UI authenticates as the machine user `login-client`, which the
    chart creates only from its one-shot setup job. When that user is missing,
    every session write fails with "Could not create session for user" and no
    one can sign in, at ZITADEL or at any relying party.

    Disabling the loginV2 feature makes the server serve its own login page,
    which needs no machine user. This is instance state in the database, not a
    values key, so it must be applied explicitly.
    """
    code, d = _api("PUT", "/v2/features/instance", token, {"loginV2": {"required": False}})
    if code not in (200, 201):
        die(f"disable loginV2 failed: {code} {d}")
    print("login page    : built-in (loginV2.required=false)")


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
    # this set empty, so every re-run tried to recreate existing roles and died
    # on the first 409.
    have = {(r.get("key") or r.get("roleKey")) for r in (d.get("result") or [])}
    for key, display in roles.items():
        if key in have:
            continue
        code, d = _api("POST", f"/management/v1/projects/{project_id}/roles", token,
                       {"roleKey": key, "displayName": display, "group": PROJECT_GROUP})
        if code not in (200, 201):
            die(f"create role {key} failed: {code} {d}")


def _ensure_oidc_apps(token: str, project_id: str, apps: dict[str, dict]) -> tuple[dict, dict]:
    code, d = _api("POST", f"/management/v1/projects/{project_id}/apps/_search", token, {})
    existing = {a.get("name"): a for a in (d.get("result") or [])}
    ids: dict[str, str] = {}
    secrets: dict[str, str] = {}
    for name, spec in apps.items():
        row = existing.get(name)
        if row:
            ids[name] = (row.get("oidcConfig") or {}).get("clientId", "")
            app_id = row.get("id")
        else:
            body = {
                "name": name,
                "redirectUris": spec["redirect_uris"],
                "responseTypes": ["OIDC_RESPONSE_TYPE_CODE"],
                "grantTypes": ["OIDC_GRANT_TYPE_AUTHORIZATION_CODE",
                               "OIDC_GRANT_TYPE_REFRESH_TOKEN"],
                "appType": spec.get("app_type_id", "OIDC_APP_TYPE_WEB"),
                "authMethodType": spec.get("auth_method_id", "OIDC_AUTH_METHOD_TYPE_BASIC"),
                "devMode": bool(spec.get("dev_mode")),
                "accessTokenType": "OIDC_TOKEN_TYPE_BEARER",
                "version": "OIDC_VERSION_1_0",
                "idTokenRoleAssertion": True,
            }
            code, d = _api("POST", f"/management/v1/projects/{project_id}/apps/oidc", token, body)
            if code not in (200, 201) or not d.get("appId"):
                die(f"create app {name} failed: {code} {d}")
            app_id = d["appId"]
            ids[name] = d.get("clientId", "")
        if spec.get("auth_method_id") == "OIDC_AUTH_METHOD_TYPE_NONE":
            continue
        code, d = _api("POST",
                       f"/management/v1/projects/{project_id}/apps/{app_id}"
                       "/oidc_config/_generate_client_secret", token, {})
        if code in (200, 201) and d.get("clientSecret"):
            secrets[name] = d["clientSecret"]
    return ids, secrets


def _ensure_smtp(token: str) -> str:
    """Make sure the instance has one active SMTP provider.

    ZITADEL sends verification codes and password resets by SMTP only. A
    freshly added config is created INACTIVE (state 3), and an inactive config
    makes GetSMTPConfig answer 404 "SMTP configuration not found", so nothing is
    ever delivered. Activating it is what enables sending: after the activate
    call the same GET returns the config and a resend reaches the relay.

    This is instance state in the database, not a values key, so it must be
    applied explicitly. The host points at the in-cluster smtp-gotify relay
    (k8s/infra/iam/smtp-gotify.yaml), which forwards to Gotify.
    """
    code, d = _api("GET", "/admin/v1/smtp", token)
    if code == 200 and (d.get("smtpConfig") or {}).get("id"):
        smtp_id = d["smtpConfig"]["id"]
        print(f"smtp      : already active ({smtp_id})")
        return smtp_id

    code, d = _api("POST", "/admin/v1/smtp", token, {
        "senderAddress": SMTP_FROM,
        "senderName": SMTP_FROM_NAME,
        "host": SMTP_HOST,
        "tls": False,
    })
    if code not in (200, 201) or not d.get("id"):
        die(f"create smtp config failed: {code} {d}")
    smtp_id = d["id"]

    # The activate call is mandatory: without it the config stays inactive and
    # every send is dropped while the API still reports success.
    code, d = _api("POST", f"/admin/v1/smtp/{smtp_id}/_activate", token, {})
    if code not in (200, 201):
        die(f"activate smtp config {smtp_id} failed: {code} {d}")

    code, d = _api("GET", "/admin/v1/smtp", token)
    if code != 200 or not (d.get("smtpConfig") or {}).get("id"):
        die(f"smtp config {smtp_id} still not visible after activate: {code} {d}")
    print(f"smtp      : active ({smtp_id}) -> {SMTP_HOST}")
    return smtp_id


def phase_apply(_args: argparse.Namespace) -> int:
    """Declare the ZITADEL objects through the API. No Terraform, no state drift."""
    token = bootstrap_token()
    if not token:
        die("no iam-admin PAT — run the keys phase first")

    _ensure_builtin_login(token)
    _ensure_smtp(token)
    project_id = _ensure_project(token, PROJECT_NAME)
    print(f"project   : {PROJECT_NAME} ({project_id})")
    _ensure_roles(token, project_id, ROLES)
    print(f"roles     : {len(ROLES)} ensured")
    ids, secrets = _ensure_oidc_apps(token, project_id, OIDC_APPS)
    print(f"oidc apps : {len(ids)} ensured, {len(secrets)} secrets issued")

    state = {
        "project_id": {"value": project_id},
        "client_ids": {"value": ids},
        "client_secrets": {"value": secrets},
        "machine_keys": {"value": {}},
        "groups_claim_enabled": {"value": False},
    }
    write_private(STATE_FILE, json.dumps(state, indent=2, sort_keys=True).encode())
    print(f"state     -> {STATE_FILE} (0600)")
    return 0


# ── phase: secrets ─────────────────────────────────────────────────────────

# client -> [(sops file, Secret name inside the file, key, kind)]
# Only secrets with a git-tracked SOPS file appear here. Every other client is
# printed as manual: guessing a path writes a credential into the wrong Secret.
SECRET_MAP = {
    "grafana": [("grafana-oidc-secrets.enc.yaml", "grafana-oidc",
                 "GF_AUTH_GENERIC_OAUTH_CLIENT_SECRET", "secret")],
    "minio": [("minio-secrets.enc.yaml", "minio-oidc",
               "MINIO_IDENTITY_OPENID_CLIENT_SECRET", "secret")],
    "paperless": [("paperless-ai-secrets.enc.yaml", "paperless-ai-config",
                   "LLM_AUTH_CLIENT_SECRET", "secret")],
}

# Clients whose client_id lives in a plain (unencrypted) manifest. They need the
# id update as much as the secret: ZITADEL generates a new client_id for every
# application, so a secret-only re-point yields invalid_client.
CLIENT_ID_MAP = {
    "grafana": ("k8s/infra/logging/grafana-values.yaml",
                r"(?m)^(\s*client_id:\s*)\S+\s*$", r"\g<1>{value}"),
    "minio": ("k8s/infra/minio/minio-tenant.yaml",
              r'(?m)^(\s*value:\s*)"(?:minio|\d+)"\s*$', r'\g<1>"{value}"'),
    "paperless": ("k8s/apps/paperless/paperless-ai.yaml",
                  r"client_id=(?:paperless-ai-agent|\d+)", "client_id={value}"),
}

# Secrets that a live workload mounts but no SOPS file in git defines. The phase
# cannot re-point them safely, so it names them.
MANUAL_ONLY: dict[str, str] = {}


def state_outputs() -> dict:
    """Read the object map written by the apply phase."""
    if not STATE_FILE.exists():
        die(f"{STATE_FILE} missing — run the apply phase first")
    return json.loads(STATE_FILE.read_text())


def sops_env() -> dict:
    env = dict(os.environ)
    env.setdefault("SOPS_AGE_KEY_FILE", str(Path.home() / ".sops" / "key.txt"))
    return env


def write_secret(rel_path: str, secret_name: str, key: str, value: str,
                 dry_run: bool) -> None:
    path = SOPS_DIR / rel_path
    if not path.exists():
        die(f"{path} does not exist — the mapping and the repo disagree")
    env = sops_env()
    dec = run(["sops", "--config", str(SOPS_CONFIG), "decrypt", str(path)], env=env)
    if dec.returncode != 0:
        die(f"cannot decrypt {rel_path}: {dec.stderr.strip()}")
    docs = [d for d in yaml.safe_load_all(dec.stdout) if d]
    matching = [d for d in docs if (d.get("metadata") or {}).get("name") == secret_name]
    if len(matching) != 1:
        die(f"{rel_path}: expected exactly one document named {secret_name}, found {len(matching)}")

    current = (matching[0].get("stringData") or {}).get(key)
    if current == value:
        print(f"  unchanged {rel_path}:{secret_name}:{key}")
        return
    if dry_run:
        print(f"  would set {rel_path}:{secret_name}:{key} (len {len(value)})")
        return

    if len(docs) == 1:
        cp = run(["sops", "--config", str(SOPS_CONFIG), "set", "--value-stdin", "--idempotent",
                  str(path), f'["stringData"]["{key}"]'], env=env, input=json.dumps(value))
        if cp.returncode != 0:
            die(f"sops set failed on {rel_path}: {cp.stderr.strip()}")
        print(f"  wrote {rel_path}:{secret_name}:{key}")
        return

    # Multi-document file. `sops set` cannot address one document, so the whole
    # file is rewritten. sops encrypt reads the creation rule from the FILENAME,
    # so the temp path is overridden with the repo-relative path; without that
    # override sops finds no matching rule and refuses.
    matching[0].setdefault("stringData", {})[key] = value
    plain = "---\n".join(yaml.safe_dump(d, sort_keys=False) for d in docs)
    fd, tmp = tempfile.mkstemp(suffix=".yaml", dir=tempfile.gettempdir())
    _TEMP_FILES.add(tmp)
    try:
        with os.fdopen(fd, "w") as fh:
            fh.write(plain)
        enc = run(["sops", "--config", str(SOPS_CONFIG), "encrypt",
                   "--filename-override", str(path), tmp], env=env)
        if enc.returncode != 0:
            die(f"sops encrypt failed for {rel_path}: {enc.stderr.strip()}")
        mode = stat.S_IMODE(path.stat().st_mode)
        fd2, staged = tempfile.mkstemp(suffix=".enc.yaml", dir=str(path.parent))
        try:
            with os.fdopen(fd2, "w") as fh:
                fh.write(enc.stdout)
            os.chmod(staged, mode)
            os.replace(staged, path)  # same directory: atomic, never EXDEV
        finally:
            if os.path.exists(staged):
                os.unlink(staged)
        print(f"  wrote {rel_path}:{secret_name}:{key} (multi-document rewrite)")
    finally:
        if os.path.exists(tmp):
            os.unlink(tmp)
        _TEMP_FILES.discard(tmp)


def update_client_id(client: str, value: str, dry_run: bool, apply_config: bool) -> None:
    spec = CLIENT_ID_MAP.get(client)
    if not spec:
        return
    rel_path, pattern, repl = spec
    path = REPO / rel_path
    if not path.exists():
        die(f"{path} does not exist — the client-id mapping is stale")
    text = path.read_text()
    matches = list(re.finditer(pattern, text))
    if len(matches) != 1:
        die(f"{rel_path}: expected 1 match for the {client} client id, found {len(matches)}")
    new_line = matches[0].expand(repl.format(value=value))
    if matches[0].group(0) == new_line:
        print(f"  unchanged {rel_path}: {client} client id")
        return
    if dry_run or not apply_config:
        print(f"  would change {rel_path}: {matches[0].group(0).strip()} -> {new_line.strip()}")
        if not apply_config:
            print("    re-run with --apply-config to write this plaintext manifest edit")
        return
    path.write_text(text[:matches[0].start()] + new_line + text[matches[0].end():])
    print(f"  wrote {rel_path}: {client} client id")


def phase_secrets(args: argparse.Namespace) -> int:
    need("sops")
    out = state_outputs()
    ids = out.get("client_ids", {}).get("value", {})
    secrets = out.get("client_secrets", {}).get("value", {})
    machine_keys = out.get("machine_keys", {}).get("value", {})

    if not ids and not machine_keys:
        die("no ZITADEL objects recorded yet — run the apply phase first")

    print("[1/3] client secrets into SOPS")
    for client, entries in SECRET_MAP.items():
        for rel_path, secret_name, key, kind in entries:
            value = ids.get(client) if kind == "id" else secrets.get(client)
            if value:
                write_secret(rel_path, secret_name, key, value, args.dry_run)

    print("[2/3] client ids in plain manifests")
    for client in list(CLIENT_ID_MAP) + list(SECRET_MAP):
        if ids.get(client):
            update_client_id(client, ids[client], args.dry_run, args.apply_config)

    print("[3/3] machine keys")
    key_dir = CONF_DIR / "machine-keys"
    for name, key_json in machine_keys.items():
        if args.dry_run:
            print(f"  would write {key_dir / (name + '.json')}")
        else:
            write_private(key_dir / f"{name}.json", key_json.encode())
            print(f"  wrote {key_dir / (name + '.json')} (0600)")

    unhandled = sorted(set(ids) - set(SECRET_MAP) - set(CLIENT_ID_MAP) - set(MANUAL_ONLY))
    if MANUAL_ONLY or unhandled:
        print("\n  MANUAL — no safe write path exists for these clients:")
        for client, why in MANUAL_ONLY.items():
            print(f"    {client}: {why}")
        for client in unhandled:
            print(f"    {client}: set its client_id and secret in its own manifest or Secret file")
    return 0


# ── phase: users ───────────────────────────────────────────────────────────

HASH_RE = re.compile(r"^pbkdf2_sha256\$(?P<rounds>\d+)\$(?P<salt>[^$]+)\$(?P<hash>[^$]+)$")


def convert_hash(django_hash: str) -> str:
    """Django pbkdf2_sha256 -> passwap MCF $pbkdf2-sha256$...

    ZITADEL imports one MCF string via hashedPassword.hash (max 200 chars) and
    pins passwap, whose pbkdf2 format is $pbkdf2-sha256$<rounds>$<salt>$<hash>
    with alt-base64 (standard base64, + -> ., no padding). Rounds, salt and hash
    bytes are unchanged, so the password still verifies.
    """
    m = HASH_RE.match(django_hash.strip())
    if not m:
        die("unsupported hash format — only pbkdf2_sha256 is convertible")
    alt = m.group("hash").rstrip("=").replace("+", ".")
    mcf = f"$pbkdf2-sha256${m.group('rounds')}${m.group('salt')}${alt}"
    if len(mcf) > 200:
        die(f"converted hash is {len(mcf)} chars, ZITADEL accepts 200")
    return mcf


def phase_users(args: argparse.Namespace) -> int:
    token = bootstrap_token()
    if not token:
        die("no bootstrap token — set ZITADEL_BOOTSTRAP_TOKEN in .env, or run the keys phase")
    if not args.django_hash:
        die("pass --django-hash with the pbkdf2_sha256 value exported from Authentik")
    body = {
        "username": args.username,
        "profile": {"givenName": args.given_name, "familyName": args.family_name},
        "email": {"email": args.email, "isVerified": True},
        "hashedPassword": {"hash": convert_hash(args.django_hash)},
        "passwordChangeRequired": False,
    }
    code, payload = curl_json("POST", f"{BASE_URL}/v2/users/human", token=token, body=body)
    if code in (200, 201):
        print(f"  {args.username}: created")
        return 0
    if code == 409:
        print(f"  {args.username}: already exists — the password was NOT updated")
        return 1
    die(f"create {args.username} -> {code} {payload[:300]}")


# ── phase: verify ──────────────────────────────────────────────────────────

def phase_verify(_args: argparse.Namespace) -> int:
    code, _ = curl_json("GET", f"{BASE_URL}/.well-known/openid-configuration", timeout=10)
    print(f"discovery      : {code}")
    if code != 200:
        die("discovery endpoint is not 200 — the instance or the ingress is not ready")

    out = state_outputs()
    ids = out.get("client_ids", {}).get("value", {})
    print(f"client apps    : {len(ids)}")
    enabled = bool(out.get("groups_claim_enabled", {}).get("value"))
    print(f"groups action  : {enabled}")
    if not enabled:
        print("BLOCKED for cutover: the groups claim action is disabled. Every consumer")
        print("that reads `groups` (kube-apiserver, MinIO, Grafana) sees no groups.")
        print("Enable it only after the live gate passes, with the script path set.")
    return 0


PHASES = {"keys": phase_keys, "apply": phase_apply, "secrets": phase_secrets,
          "users": phase_users, "verify": phase_verify}


def main() -> int:
    parser = argparse.ArgumentParser(description="Provision ZITADEL end to end.")
    parser.add_argument("phase", choices=[*PHASES, "all"])
    parser.add_argument("--org-id", help="ZITADEL org id (required for apply)")
    parser.add_argument("--dry-run", action="store_true", help="print actions, write nothing")
    parser.add_argument("--apply-config", action="store_true",
                        help="also rewrite client ids inside plain manifests")
    parser.add_argument("--username", default="rock")
    parser.add_argument("--email", default="[email protected]")
    parser.add_argument("--given-name", default="Rock")
    parser.add_argument("--family-name", default="Liang")
    parser.add_argument("--django-hash", help="pbkdf2_sha256 value exported from Authentik")
    args = parser.parse_args()

    load_env_file()
    need("kubectl")
    if args.phase == "all":
        if not args.org_id:
            die("all requires --org-id")
        for name in ("keys", "apply"):
            rc = PHASES[name](args)
            if rc:
                return rc
        rc = phase_secrets(args)
        if rc:
            return rc
        return phase_verify(args)
    return PHASES[args.phase](args)


if __name__ == "__main__":
    sys.exit(main())