home-network-k3s-impl/conformance/tests/l0-static/keycloak-passkey-reconciler/check.sh

#!/usr/bin/env bash
# =============================================================================
# L0 static check (offline, no cluster) — keycloak-passkey-reconciler
# Serves: R-IDP-03 ("Identity config (realms, clients, groups, roles) is declarative."
#                   — spec docs/15-identity-and-access.md)
#         R-IDP-04 ("IdP secrets SOPS-encrypted; MFA available."
#                   — spec docs/15-identity-and-access.md)
#
# WHY THIS EXISTS — THE REALM IMPORT IS A ONE-SHOT THAT FAILS OPEN
# ---------------------------------------------------------------
# platform/keycloak/keycloak-realm-import.yaml declares the WebAuthn PASSWORDLESS
# policy and a restructured `browser-mfa` browser flow. Measured on prod 2026-10-18,
# hours after that declaration merged:
#
#   kubectl -n keycloak get keycloakrealmimport homelab -o jsonpath='{.status}'
#   Done=True  Started=False  HasErrors=False
#
# A KeycloakRealmImport is a ONE-SHOT. Once it rests at Done=True the operator never
# re-runs it, so every realm-level field and every flow declared after the FIRST
# import is inert — in git, applied by Flux, reported Ready, reaching the realm
# never. The same shape as the 2026-09-06 client-secret incident (12 prod clients
# holding the literal `(env:...)` placeholder behind an entirely green cluster), and
# the same fix: a level-triggered reconciler. This check is the static half.
#
# THE RECONCILER HOLDS WRITE ACCESS TO THE IDENTITY PROVIDER. That is the reason this
# check is unusually thorough about its BLAST RADIUS rather than only its behaviour.
#
# GIVEN the repository as authored,
# THEN  (1) THE DESIRED STATE IS DERIVED, NOT DECLARED TWICE: the reconciler's input
#           (platform/keycloak/passkey-desired-state.yaml) is GENERATED from the realm
#           import by ci/render-keycloak-passkey-policy.py, and this check RUNS the
#           renderer with --check, so a hand edit that makes the two disagree is
#           unmergeable. The RP ID is the sharpest case: it is baked into every passkey
#           at enrolment and can NEVER be re-keyed, so two declarations that can skew
#           about the domain is an unrecoverable-credential bug, not a cosmetic one.
#       (2) BOTH FILES ARE RENDERED BY THE SAME FLUX KUSTOMIZATION.
#           postBuild.substituteFrom is per-Kustomization, so a split would let the
#           ${BASE_DOMAIN} token in the realm import and the one in the desired state
#           resolve to two different values — and an RP ID mismatch invalidates every
#           enrolled passkey with no error the user can act on.
#       (3) THE DESIRED STATE USES NO SUBSTITUTION TOKEN THE REALM IMPORT DOES NOT, and
#           declares no realm key outside the reconciler's write surface.
#       (4) THE TWO WRITE SURFACES AGREE: the renderer's exported policy keys and the
#           reconciler's POLICY_FIELDS are the SAME tuple. If they drifted, the renderer
#           could emit a key the reconciler refuses (fail-closed forever) or the
#           reconciler could accept a key the renderer never emits (dead surface).
#       (5) THE RECONCILER MANIFEST IS SAFE AND WIRED: reconciled from BOTH cluster
#           entrypoints, mounts exactly the ConfigMaps its code defaults to, carries NO
#           ServiceAccount token, and contains no dollar sign (it ships inside a
#           SUBSTITUTED Flux path, where an unescaped dollar-brace expansion is
#           rewritten before any container runs).
#       (6) THE CURRENTLY BOUND FLOW IS STRUCTURALLY UNWRITABLE — asserted against the
#           SOURCE, not the comments. The flow converges BLUE/GREEN (build under a new
#           alias, verify, flip browserFlow in one atomic PUT), so flow-create and
#           execution-add endpoints ARE on the allow-list and their presence is not the
#           hazard. What is asserted instead: nothing is DELETABLE (the superseded flow
#           is the rollback target); the execution- and config-id-scoped write endpoints
#           stay off the list because they carry no flow alias and so cannot be screened;
#           no required-action write reaches any alias but the one named literally; and
#           the guard the allow-list CANNOT express exists in the source and is wired
#           into the single request choke point, FAIL-CLOSED until the live binding has
#           been read. That last one is what the whole design rests on: a reconciler that
#           half-rewrites the BOUND browser flow is a total SSO outage on a loop, and
#           this repo has had one (2026-09-06).
#       (7) THE SIBLING RECONCILER IS UNTOUCHED: client-secret-reconciler.yaml's
#           allow-list is NOT widened by this work. Its --selftest asserts it refuses
#           /users, /groups, /roles, /authentication and the whole-realm endpoint, and
#           its guarantee that prod's real user and the media-users / platform-admins /
#           viewers groups survive every run must stay LITERALLY true.
#       (8) THE RECONCILER'S BEHAVIOUR IS PROVEN, NOT ASSERTED: this check EXTRACTS the
#           script from the ConfigMap and executes its `--selftest`, which drives the
#           real code paths against an in-process stub Keycloak that MODELS the three
#           Keycloak 26.0.6 behaviours that bite — the RpEntityName gate on the
#           passwordless policy, the all-or-nothing policy rebuild that defaults every
#           omitted key, and the required-action representation replace.
#      (8b) AND THAT PROOF IS FALSIFIABLE: mutations.py injects plausible regressions
#           one at a time and requires --selftest to CATCH every one. A passing suite
#           proves nothing if it cannot fail, and "reported success while the realm was
#           broken" is the exact failure this component exists to end.
#   AND (9) THIS CHECK IS LOAD-BEARING: it DECLARES
#           `conformance.home.k3s/scope: whole-repo` so ci/l0.sh runs it unconditionally
#           via --include-scope, ci/gate.sh blocks the merge when the l0 stage fails, and
#           ci/touched-requirements.sh maps platform/keycloak/ to these requirement IDs.
#           An unreferenced test is precisely the "looks applied and isn't" failure this
#           repo keeps hitting.
#
# NOT proven here (named honestly, in the style of the sibling L0 checks):
#   - that the LIVE realm's policy, enrolment action or browser flow match git. That is
#     cluster state. The reconciler's own `--audit` mode is the read-only probe for it
#     and is what an L2 test should run; no such L2 test is added by this change, so the
#     live realm is UNVERIFIED by CI.
#   - that the declared browser flow actually WORKS as a login path. Nothing offline can
#     prove that a passkey ceremony completes, that the ALTERNATIVE fork degrades to the
#     password form, or that a platform-admin is still asked for TOTP. Those need a real
#     browser against a real realm.
#   - that the reconciler's first real run against prod will succeed. The flow is built
#     through five admin-API endpoints whose request shapes were read from the 26.0.6
#     source but never exercised against a live Keycloak.
#   - that the admin credential decrypts to a usable password. Key PRESENCE is checkable
#     offline; the plaintext is not, and must not be.
#   - that the CronJob's image can be pulled or its schedule fires.
# =============================================================================
set -euo pipefail
DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
ROOT="$(cd "${DIR}/../../../.." && pwd)"        # repo root
PYTHON="${PYTHON:-python3}"

IMPORT_FILE="platform/keycloak/keycloak-realm-import.yaml"
RECON_FILE="platform/keycloak/passkey-reconciler.yaml"
STATE_FILE="platform/keycloak/passkey-desired-state.yaml"
RENDERER="ci/render-keycloak-passkey-policy.py"
SIBLING_FILE="platform/keycloak/client-secret-reconciler.yaml"

# Honest SKIP (never green) if the thing under test is absent entirely.
for f in "$IMPORT_FILE" "$RECON_FILE" "$STATE_FILE" "$RENDERER"; do
  if [ ! -f "${ROOT}/${f}" ]; then
    echo "check SKIPPED: ${f} absent — nothing to audit" >&2
    exit 3
  fi
done

cd "${ROOT}"
"${PYTHON}" - "${ROOT}" <<'PY'
"""Static audit of the Keycloak passkey declaration, the renderer that derives the
reconciler's desired state from it, and the reconciler's blast radius. PyYAML parses the
manifests; lib/status.py is IMPORTED for the Flux graph walk rather than forked, so there
is one definition of "what a Flux path renders" in the harness."""
from __future__ import annotations

import ast
import json
import re
import subprocess
import sys
import tempfile
from pathlib import Path

import yaml

ROOT = Path(sys.argv[1])
# This body is fed to `python3 -` on stdin, so there is no __file__ to derive from; the
# test directory is named relative to the repo root the wrapper passes in.
DIR_PATH = (ROOT / "conformance" / "tests" / "l0-static"
            / "keycloak-passkey-reconciler")
sys.path.insert(0, str(ROOT / "conformance" / "lib"))
import status as st  # noqa: E402

IMPORT_FILE = "platform/keycloak/keycloak-realm-import.yaml"
RECON_FILE = "platform/keycloak/passkey-reconciler.yaml"
STATE_FILE = "platform/keycloak/passkey-desired-state.yaml"
RENDERER = "ci/render-keycloak-passkey-policy.py"
SIBLING_FILE = "platform/keycloak/client-secret-reconciler.yaml"
STATE_CM_NAME = "keycloak-passkey-desired-state"
STATE_DATA_KEY = "passkey-desired-state.json"
SCRIPT_KEY = "reconcile-keycloak-passkeys.py"
ENROL_ACTION = "webauthn-register-passwordless"
DOLLAR = chr(36)

failures: list[str] = []


def check(cond, msg):
    if not cond:
        failures.append(msg)


def load(rel):
    return list(yaml.safe_load_all((ROOT / rel).read_text()))


# ================================================== (1) the desired state is DERIVED
proc = subprocess.run([sys.executable, str(ROOT / RENDERER), "--check"],
                      capture_output=True, text=True, timeout=120)
check(proc.returncode == 0,
      f"{STATE_FILE} has DRIFTED from {IMPORT_FILE} (ci/render-keycloak-passkey-policy"
      f".py --check failed). The reconciler would converge the realm on a passkey policy "
      f"the realm import does not declare, and audit the login flow against a tree "
      f"nobody authored. The RP ID is the sharpest case: it is baked into every passkey "
      f"at enrolment and can NEVER be re-keyed, so two declarations that skew about the "
      f"domain is an unrecoverable-credential bug.\n        "
      + "\n        ".join((proc.stdout + proc.stderr).strip().splitlines()[-12:]))

# The realm import must actually declare the things the renderer derives from. If it
# stops, the renderer fails -- but say so HERE too, because a SKIP-shaped failure in the
# renderer would otherwise be the only signal.
imp_docs = [d for d in load(IMPORT_FILE) if d]
imp = next((d for d in imp_docs if d.get("kind") == "KeycloakRealmImport"), None)
check(imp is not None, f"{IMPORT_FILE}: no KeycloakRealmImport document")
realm = ((imp or {}).get("spec") or {}).get("realm") or {}
policy_keys_in_import = sorted(k for k in realm
                               if k.startswith("webAuthnPolicyPasswordless"))
check(policy_keys_in_import,
      f"{IMPORT_FILE} declares no webAuthnPolicyPasswordless* field, so there is no "
      f"passkey policy for the reconciler to converge")
check(realm.get("browserFlow"),
      f"{IMPORT_FILE} declares no spec.realm.browserFlow — a perfect flow that is not "
      f"BOUND changes nothing about how anyone logs in")

# THE requiredActions TRAP, enforced rather than only commented. The realm import's own
# header explains it at length: Keycloak 26.0.6 treats requiredActions in a realm
# representation as REPLACE, not merge, so declaring ONE action imports exactly that
# action plus delete_account and skips DefaultRequiredActions.addActions entirely --
# deleting CONFIGURE_TOTP, UPDATE_PASSWORD, VERIFY_EMAIL and seven more. Losing
# CONFIGURE_TOTP is a hard lock-out of every platform-admin, because auth-otp-form in the
# browser flow still demands a TOTP they could then never enrol. The comment saying "DO
# NOT ADD ONE" is advice; this is the gate.
check("requiredActions" not in realm,
      f"{IMPORT_FILE} declares spec.realm.requiredActions. Keycloak 26.0.6 treats that "
      f"as REPLACE, not merge: declaring any action imports exactly those plus "
      f"delete_account and NEVER runs DefaultRequiredActions.addActions, so "
      f"CONFIGURE_TOTP, UPDATE_PASSWORD, VERIFY_EMAIL and seven more are DELETED. That "
      f"is a permanent lock-out of every platform-admin, because auth-otp-form in this "
      f"realm's browser flow still demands a TOTP they could no longer enrol. The "
      f"enrolment action is already registered by default (priority 80, enabled, "
      f"defaultAction false) and the passkey reconciler converges it through the "
      f"per-alias admin API, which has no such hazard")

# ============================== (2)(3) the two files share one substitution pass
state_docs = [d for d in load(STATE_FILE) if d]
state_cm = next((d for d in state_docs if d.get("kind") == "ConfigMap"), None)
check(state_cm is not None, f"{STATE_FILE}: no ConfigMap")
desired = {}
if state_cm is not None:
    check(state_cm["metadata"]["name"] == STATE_CM_NAME,
          f"{STATE_FILE}: the ConfigMap is named "
          f"{state_cm['metadata']['name']!r}, but the CronJob mounts {STATE_CM_NAME!r}")
    data = state_cm.get("data") or {}
    check(list(data) == [STATE_DATA_KEY],
          f"{STATE_FILE}: expected exactly the data key {STATE_DATA_KEY!r}, got "
          f"{sorted(data)} — the reconciler's PASSKEY_STATE_FILE default would not "
          f"resolve")
    raw = data.get(STATE_DATA_KEY, "")
    try:
        desired = json.loads(raw)
    except ValueError as e:
        check(False, f"{STATE_FILE}: the {STATE_DATA_KEY} payload is not valid JSON "
                     f"({e}) — the reconciler would fail closed on every run")

# Both files must be rendered by ONE Flux Kustomization. substituteFrom is
# per-Kustomization, so a split lets the same ${BASE_DOMAIN} token resolve to two
# different values -- and an RP ID mismatch invalidates every enrolled passkey silently.
import_owners = set()
state_owners = set()
recon_owners = set()
for env in ("dev", "prod"):
    entry = f"clusters/{env}"
    if not (ROOT / entry).is_dir():
        continue
    applied, _ = st.reconciled_files(ROOT, entry)
    check(RECON_FILE in applied,
          f"{RECON_FILE} is not reachable from the {entry} entrypoint — the passkey "
          f"healer would exist on one cluster only")
    check(STATE_FILE in applied,
          f"{STATE_FILE} is not reachable from the {entry} entrypoint — the reconciler "
          f"would fail closed on that cluster forever")
    check(IMPORT_FILE in applied, f"{IMPORT_FILE} is not reachable from {entry}")

# Which Kustomization DIRECTORY lists each file. Asserted structurally: both must be
# listed as resources of the same kustomization.yaml.
for kpath in ROOT.glob("*/kustomization.yaml"):
    try:
        k = yaml.safe_load(kpath.read_text()) or {}
    except Exception:
        continue
    res = [str(r) for r in (k.get("resources") or [])]
    base = kpath.parent.name
    if any(r.endswith("keycloak-realm-import.yaml") for r in res):
        import_owners.add(base)
    if any(r.endswith("passkey-desired-state.yaml") for r in res):
        state_owners.add(base)
    if any(r.endswith("passkey-reconciler.yaml") for r in res):
        recon_owners.add(base)
check(import_owners and import_owners == state_owners,
      f"{IMPORT_FILE} is rendered by {sorted(import_owners)} but {STATE_FILE} by "
      f"{sorted(state_owners)}. They MUST be the same Flux Kustomization: "
      f"postBuild.substituteFrom is per-Kustomization, so a split would let the "
      f"{DOLLAR}{{BASE_DOMAIN}} token in each resolve to a different value. An RP ID is "
      f"baked into every passkey at enrolment and can never be re-keyed, so a domain the "
      f"two files disagree about is an unrecoverable credential, not a cosmetic drift")
check(recon_owners == state_owners,
      f"{RECON_FILE} is rendered by {sorted(recon_owners)} but its desired state by "
      f"{sorted(state_owners)} — the reconciler could be applied without its input")

# (3) no substitution token the realm definition does not also use.
import_text = (ROOT / IMPORT_FILE).read_text()
TOKEN = re.compile(re.escape(DOLLAR) + r"\{([A-Za-z_][A-Za-z0-9_]*)")
state_tokens = set(TOKEN.findall((ROOT / STATE_FILE).read_text()))
import_tokens = set(TOKEN.findall(import_text))
# The renderer's own HEADER mentions ${BASE_DOMAIN} in prose; that is the same token the
# policy uses, so a subset check is the right assertion rather than an exact match.
check(state_tokens <= import_tokens,
      f"{STATE_FILE} uses substitution token(s) {sorted(state_tokens - import_tokens)} "
      f"that {IMPORT_FILE} does not. A token the realm definition never uses is either a "
      f"hand edit or a renderer bug, and an UNDEFINED one renders as the EMPTY STRING "
      f"silently — which for an RP ID means every passkey ceremony fails")

# ======================================= (4) the two write surfaces are the SAME tuple
recon_docs = [d for d in load(RECON_FILE) if d]
cm = next((d for d in recon_docs if d.get("kind") == "ConfigMap"), None)
cj = next((d for d in recon_docs if d.get("kind") == "CronJob"), None)
check(cm is not None, f"{RECON_FILE}: no ConfigMap carrying the reconciler script")
check(cj is not None, f"{RECON_FILE}: no CronJob running the reconciler")

script = ""
if cm is not None:
    data = cm.get("data") or {}
    check(len(data) == 1,
          f"{RECON_FILE}: the reconciler ConfigMap must carry exactly one script key "
          f"(found {sorted(data)}) — a second copy would be untested by this gate")
    check(SCRIPT_KEY in data,
          f"{RECON_FILE}: expected the script under {SCRIPT_KEY!r}, got {sorted(data)}")
    script = next(iter(data.values()), "")
    check(len(script.splitlines()) > 200,
          f"{RECON_FILE}: the extracted script is implausibly short "
          f"({len(script.splitlines())} lines) — the ConfigMap shape changed and the "
          f"selftest below would prove nothing")
    # THE nextcloud TRAP. This file is rendered by a Flux Kustomization with
    # postBuild.substituteFrom, so envsubst rewrites it before any container runs.
    check(DOLLAR not in script,
          f"{RECON_FILE}: the reconciler script contains a dollar sign. It ships inside "
          f"a SUBSTITUTED Flux path, where a dollar-brace expansion is rewritten at "
          f"reconcile time (see apps/wave-1/nextcloud.yaml). Rewrite without it — Python "
          f"never needs one; chr(36) spells it where a literal is genuinely required")


def tuple_literal(src, name):
    """The value of a module-level tuple assignment, via ast. Parsing the SOURCE rather
    than importing it keeps this check offline and side-effect free, and means the thing
    asserted is the text that ships."""
    try:
        tree = ast.parse(src)
    except SyntaxError:
        return None
    for node in tree.body:
        if isinstance(node, ast.Assign):
            for t in node.targets:
                if isinstance(t, ast.Name) and t.id == name:
                    try:
                        return ast.literal_eval(node.value)
                    except ValueError:
                        return None
    return None


renderer_src = (ROOT / RENDERER).read_text()


# POLICY_FIELDS is a CONCATENATION of other tuples in both files, so literal_eval of the
# assignment itself cannot evaluate it. Resolve it from its parts instead. Both files name
# their parts from the same vocabulary; a part present in one and not the other simply
# contributes nothing, which is why the comparison below is of the resolved SETS.
def resolve_policy(src):
    parts = ("POLICY_STR_FIELDS", "POLICY_ORDERED_LIST_FIELDS",
             "POLICY_UNORDERED_LIST_FIELDS", "POLICY_LIST_FIELDS", "POLICY_INT_FIELDS",
             "POLICY_BOOL_FIELDS")
    out = []
    for p in parts:
        v = tuple_literal(src, p)
        if v:
            out.extend(v)
    return sorted(set(out))


recon_keys = resolve_policy(script)
rend_keys = resolve_policy(renderer_src)
check(recon_keys and recon_keys == rend_keys,
      f"the reconciler's writable policy keys {recon_keys} and the renderer's exported "
      f"keys {rend_keys} DISAGREE. If the renderer emits a key the reconciler refuses, "
      f"every run fails closed forever; if the reconciler accepts a key the renderer "
      f"never emits, that is dead write surface on the identity provider")
# Every exported key must actually be a passwordless key. The two-factor
# webAuthnPolicy* family is a silent no-op for passkeys (the passwordless authenticator
# reads getWebAuthnPolicyPasswordless() only), so exporting one would be a write surface
# with no effect.
check(all(k.startswith("webAuthnPolicyPasswordless") for k in recon_keys),
      f"the reconciler's writable keys include a non-passwordless field: "
      f"{[k for k in recon_keys if not k.startswith('webAuthnPolicyPasswordless')]}. "
      f"The two-factor webAuthnPolicy* family is never consulted by a passkey login, so "
      f"writing it is a silent no-op")
# And the desired state must declare only those.
if isinstance(desired.get("webAuthnPolicyPasswordless"), dict):
    extra = sorted(set(desired["webAuthnPolicyPasswordless"]) - set(recon_keys))
    check(not extra,
          f"{STATE_FILE} declares policy key(s) {extra} outside the reconciler's write "
          f"surface — it would fail closed on every run")
    # The 26.0.6 gate: the policy is applied ONLY when RpEntityName is non-empty.
    check(desired["webAuthnPolicyPasswordless"].get(
              "webAuthnPolicyPasswordlessRpEntityName"),
          f"{STATE_FILE} carries no non-empty "
          f"webAuthnPolicyPasswordlessRpEntityName. Keycloak 26.0.6 applies the "
          f"passwordless policy only when that field is set "
          f"(DefaultExportImportManager.updateRealm), so every realm PUT would be "
          f"accepted with 204 and change nothing")

# THE ENROLMENT POSTURE. defaultAction true would force every user to enrol a passkey at
# next login -- a platform-wide lock-out risk, and the exact opposite of the realm
# import's stated "additive and optional" decision. It is one boolean, so it is gated.
ra = desired.get("requiredAction") or {}
check(ra.get("alias") == ENROL_ACTION,
      f"{STATE_FILE}: requiredAction.alias is {ra.get('alias')!r}, expected "
      f"{ENROL_ACTION!r} — the reconciler's allow-list names that alias literally, so "
      f"any other value is unreachable and the pass would fail closed")
check(ra.get("enabled") is True,
      f"{STATE_FILE}: requiredAction.enabled must be true, else passkey ENROLMENT is "
      f"impossible while the login flow offers passkeys")
check(ra.get("defaultAction") is False,
      f"{STATE_FILE}: requiredAction.defaultAction is {ra.get('defaultAction')!r}, which "
      f"must be false. True forces EVERY user to enrol a passkey at next login — this "
      f"IdP gates Grafana, Gitea, Nextcloud, Headlamp, Semaphore, PiHole, Crafty, the "
      f"*arr stack AND kubectl, so a misfiring mandatory enrolment is a total platform "
      f"lock-out. The realm import's whole stated posture is additive and OPTIONAL")

# ========================================= (5) the reconciler manifest is safe + wired
if cj is not None:
    pod = cj["spec"]["jobTemplate"]["spec"]["template"]["spec"]
    ctr = pod["containers"][0]
    vols = {v["name"]: v for v in pod.get("volumes", [])}
    mounts = {m["name"]: m for m in ctr.get("volumeMounts", [])}
    env = {e["name"]: e for e in ctr.get("env", [])}

    # NO KUBERNETES API ACCESS. This is what makes the mounted-ConfigMap design strictly
    # narrower than reading the KeycloakRealmImport CR through the API: no Role, no
    # RoleBinding, no token. It is also why this change adds no **/rbac.yaml.
    check(pod.get("automountServiceAccountToken") is False,
          f"{RECON_FILE}: the reconciler must set automountServiceAccountToken: false. "
          f"It talks only to the Keycloak admin API and reads a mounted ConfigMap, so a "
          f"ServiceAccount token is pure blast radius (R-SEC-02)")
    check("serviceAccountName" not in pod,
          f"{RECON_FILE}: the reconciler must not name a ServiceAccount")

    cm_vols = {n: v for n, v in vols.items() if "configMap" in v}
    names = {(v["configMap"] or {}).get("name") for v in cm_vols.values()}
    check(STATE_CM_NAME in names,
          f"{RECON_FILE}: the CronJob does not mount the {STATE_CM_NAME} ConfigMap, so "
          f"the reconciler would fail closed on every run. Mounted: {sorted(names)}")
    check(cm["metadata"]["name"] in names,
          f"{RECON_FILE}: the CronJob does not mount its own script ConfigMap")
    # NO SECRET IS MOUNTED AS A VOLUME. The admin credential is consumed via
    # secretKeyRef only; projecting it as a file would widen what a readable mount leaks.
    check(not [v for v in vols.values() if "secret" in v],
          f"{RECON_FILE}: the reconciler mounts a Secret as a volume. The admin "
          f"credential is referenced by secretKeyRef only; it must never be projected "
          f"as a file here")

    # The mount paths must be the ones the code DEFAULTS to, or the container starts and
    # fails closed forever while every manifest looks correct.
    state_vol = next((n for n, v in cm_vols.items()
                      if (v["configMap"] or {}).get("name") == STATE_CM_NAME), None)
    if state_vol:
        mp = (mounts.get(state_vol) or {}).get("mountPath")
        check(mp == "/etc/keycloak-passkey",
              f"{RECON_FILE}: the desired-state ConfigMap is mounted at {mp!r}, but the "
              f"script's PASSKEY_STATE_FILE default is "
              f"/etc/keycloak-passkey/{STATE_DATA_KEY}")
        check(f'"/etc/keycloak-passkey/{STATE_DATA_KEY}"' in script,
              f"{RECON_FILE}: the script's PASSKEY_STATE_FILE default no longer matches "
              f"the mount path — they are a pair and must move together")
        check((mounts.get(state_vol) or {}).get("readOnly") is True,
              f"{RECON_FILE}: the desired-state mount must be readOnly")

    # The admin credential, referenced BY NAME. Never decrypted, never a literal.
    for key in ("KEYCLOAK_ADMIN_USER", "KEYCLOAK_ADMIN_PASSWORD"):
        e = env.get(key) or {}
        ref = ((e.get("valueFrom") or {}).get("secretKeyRef") or {})
        check(ref.get("name") == "keycloak-admin-credentials",
              f"{RECON_FILE}: {key} must come from the keycloak-admin-credentials "
              f"Secret by secretKeyRef, got {e!r}")
        check("value" not in e,
              f"{RECON_FILE}: {key} carries a literal `value`. A credential in a "
              f"manifest is a plaintext secret in git")
    check((env.get("KEYCLOAK_REALM") or {}).get("value") == desired.get("realm"),
          f"{RECON_FILE}: KEYCLOAK_REALM is "
          f"{(env.get('KEYCLOAK_REALM') or {}).get('value')!r} but the desired state "
          f"declares realm {desired.get('realm')!r}. The reconciler would converge the "
          f"wrong realm, or none")

    check(ctr["securityContext"]["readOnlyRootFilesystem"] is True,
          f"{RECON_FILE}: readOnlyRootFilesystem must be true")
    check(ctr["securityContext"]["allowPrivilegeEscalation"] is False,
          f"{RECON_FILE}: allowPrivilegeEscalation must be false")
    check(pod["securityContext"].get("runAsNonRoot") is True,
          f"{RECON_FILE}: runAsNonRoot must be true")
    check(cj["spec"].get("concurrencyPolicy") == "Forbid",
          f"{RECON_FILE}: concurrencyPolicy must be Forbid — two concurrent reconcilers "
          f"would double the write attempts against the IdP during an outage recovery")
    check(re.fullmatch(r"\S+ \S+ \S+ \S+ \S+", cj["spec"].get("schedule", "")),
          f"{RECON_FILE}: schedule {cj['spec'].get('schedule')!r} is not a 5-field cron")

# ===== (6) THE BOUND FLOW IS STRUCTURALLY UNWRITABLE — asserted against the SOURCE
#
# This component converges the browser flow BLUE/GREEN: it builds the target structure
# under a NEW, unused alias, verifies it, then flips the realm's browserFlow in ONE
# atomic PUT. It therefore NEEDS flow-create and execution-add endpoints, so unlike the
# first design of this check the mere presence of a flow write is no longer the thing to
# forbid. What must hold instead is narrower and stronger:
#
#   (a) nothing may be DELETED, ever — the superseded flow is the rollback target;
#   (b) the single-execution and config-edit endpoints stay off the list — requirements
#       are set through the flow-scoped endpoint, which the guard in (d) screens;
#   (c) no required-action write may reach any alias but the one named LITERALLY;
#   (d) and the guard the allow-list CANNOT express must exist in the source: a
#       method-aware screen that refuses any flow write targeting the CURRENTLY BOUND
#       flow's subtree, FAIL-CLOSED until that binding has been read.
#
# (d) is the load-bearing one. Without it the reconciler could mutate the flow users are
# logging in through, and a run dying mid-restructure would leave a HALF-BUILT FLOW BOUND
# AS THE LOGIN PATH for Grafana, Gitea, Nextcloud, Headlamp, Semaphore, PiHole, Crafty,
# the *arr stack AND kubectl. This repo has already had a total SSO outage from a realm
# change (2026-09-06).
allow = tuple_literal(script, "_ALLOWED") or ()
check(bool(allow),
      f"{RECON_FILE}: could not parse the _ALLOWED endpoint allow-list out of the "
      f"script. This check's central assertions about the write surface cannot be made, "
      f"so it must FAIL rather than pass silently")
if allow:
    check(all(isinstance(e, tuple) and len(e) == 2 for e in allow),
          f"{RECON_FILE}: _ALLOWED is not a tuple of (method, path) pairs. A path-only "
          f"allow-list cannot distinguish reading a flow from writing one, which this "
          f"component's whole safety argument rests on")
    writes = [(m, p) for m, p in allow if m in ("PUT", "POST", "PATCH", "DELETE")]
    # (a) nothing is ever deleted.
    check(not [(m, p) for m, p in allow if m == "DELETE"],
          f"{RECON_FILE}: the allow-list contains a DELETE method. Nothing this program "
          f"does requires deleting anything: a superseded flow is left unbound as the "
          f"ROLLBACK TARGET, and DELETE /admin/realms/<realm> deletes the realm")
    # (b) the endpoints that would let it edit live objects in place.
    banned = [(m, p) for m, p in writes
              if re.fullmatch(r"/admin/realms/\[\^/\]\+/authentication/executions/"
                              r"\[\^/\]\+", p)
              or re.fullmatch(r"/admin/realms/\[\^/\]\+/authentication/config/\[\^/\]\+",
                              p)
              or p.endswith("/authentication/register-required-action")]
    check(not banned,
          f"{RECON_FILE}: the allow-list permits {banned}. The single-execution update "
          f"and config-edit endpoints must stay off it: requirements are set through the "
          f"FLOW-SCOPED endpoint, which carries the flow alias in its path and is "
          f"therefore screenable against the bound subtree. An execution- or "
          f"config-id-scoped write is not, so it would be an unscreened hole")
    # (c) no required-action write may reach any other alias.
    ra_writes = [(m, p) for m, p in writes if "required-actions" in p]
    check(all(p.endswith("required-actions/" + ENROL_ACTION) for _, p in ra_writes),
          f"{RECON_FILE}: the allow-list permits a required-action write to an alias "
          f"other than {ENROL_ACTION!r}: {ra_writes}. Disabling CONFIGURE_TOTP is a "
          f"permanent lock-out of every platform-admin, because auth-otp-form in the "
          f"browser flow still demands a TOTP that could no longer be enrolled")
    check(not [p for _, p in ra_writes if "[^/]+" in p.split("required-actions/")[-1]],
          f"{RECON_FILE}: a required-action write path uses a WILDCARD alias. The alias "
          f"must be named LITERALLY so no other action is writable by construction")
    # Nothing outside the realm + authentication surface at all.
    stray = [(m, p) for m, p in allow
             if "/users" in p or "/groups" in p or "/roles" in p or "/clients" in p]
    check(not stray,
          f"{RECON_FILE}: the allow-list reaches {stray}. Users, groups and roles must "
          f"be unreachable (prod's real user and the media-users / platform-admins / "
          f"viewers groups must survive every run); clients are the SIBLING "
          f"reconciler's surface and the two must not overlap")

# ---- (d) THE PROTECTED-SUBTREE GUARD EXISTS, IS WIRED, AND IS FAIL-CLOSED ----------
# Asserted against the SOURCE, because this is the property the blue/green design rests
# on and a comment claiming it is worth nothing. Three things must all be true.
check(re.search(r"^def _screen_write\(", script, re.M) is not None,
      f"{RECON_FILE}: the script has no _screen_write() function. That is the guard "
      f"that refuses any flow write targeting the CURRENTLY BOUND flow's subtree — the "
      f"one thing the allow-list cannot express, because its write entries are "
      f"alias-parameterised. Without it this component can mutate the live login path")
# It must be CALLED from the single choke point, not merely defined.
check(re.search(r"_screen_write\(method, path", script) is not None,
      f"{RECON_FILE}: _screen_write() is defined but never called from _http() with the "
      f"request's method and path. A guard that is not on the request path is decoration")
# FAIL-CLOSED: the protected sets must start undetermined, and the reset that runs at
# the top of every run must put them back there.
reset_fn = re.search(r"def _reset_protected\(\):(.*?)(?=\n\S)", script, re.S)
check(reset_fn is not None
      and "_PROTECTED_FLOW_ALIASES = None" in reset_fn.group(1)
      and "_PROTECTED_EXECUTION_IDS = None" in reset_fn.group(1),
      f"{RECON_FILE}: _reset_protected() does not set both protected sets to None. They "
      f"must start each run UNDETERMINED, because while they are, every flow write is "
      f"refused — that is the fail-closed half of the guard, and it is what stops a "
      f"build running before the live binding has been read")
check(re.search(r"_reset_protected\(\)", script.split("def _selftest")[0]) is not None,
      f"{RECON_FILE}: _reset_protected() is never called outside the selftest, so the "
      f"protected sets are not reset per run and a previous run's binding could be "
      f"trusted for this one")
# The create-flow endpoint carries NO alias in its path, so it must be screened on its
# BODY or it escapes the guard entirely. (Found by the selftest; recorded here so a
# future edit cannot silently undo it.)
check("_FLOW_CREATE_PATH" in script,
      f"{RECON_FILE}: there is no separate screen for POST .../authentication/flows. "
      f"That endpoint carries no alias in its path — the alias is in the request BODY — "
      f"so without a body-based screen it bypasses the protected-subtree guard "
      f"altogether")
# And the one function allowed to change the live binding must be distinct from the
# policy write, which must refuse to.
check("def bind_browser_flow(" in script,
      f"{RECON_FILE}: there is no bind_browser_flow() function. The atomic flip must be "
      f"one named, auditable operation, not an incidental field in some other write")
check(re.search(r"never rebinds the browser flow", script) is not None,
      f"{RECON_FILE}: the policy write no longer refuses to change browserFlow. "
      f"PUT /admin/realms/<realm> DOES read rep.getBrowserFlow(), so the policy pass "
      f"can rebind the login path by accident — to an UNVERIFIED flow")

# ======================================== (7) the SIBLING reconciler is NOT widened
if (ROOT / SIBLING_FILE).is_file():
    sib_docs = [d for d in load(SIBLING_FILE) if d]
    sib_cm = next((d for d in sib_docs if d.get("kind") == "ConfigMap"), None)
    sib_script = next(iter((sib_cm.get("data") or {}).values()), "") if sib_cm else ""
    sib_allow = tuple_literal(sib_script, "_ALLOWED") or ()
    check(bool(sib_allow),
          f"{SIBLING_FILE}: could not parse its _ALLOWED list, so this check cannot "
          f"prove the sibling was not widened")
    if sib_allow:
        # The sibling's list is PATH-ONLY by design. Its selftest asserts it refuses the
        # whole-realm endpoint and the /authentication tree; assert the same here so a
        # future edit that "just adds the realm endpoint there instead" is caught.
        bad = [p for p in sib_allow
               if isinstance(p, str)
               and (re.fullmatch(r"/admin/realms/\[\^/\]\+", p)
                    or "authentication" in p or "/users" in p or "/groups" in p
                    or "/roles" in p)]
        check(not bad,
              f"{SIBLING_FILE}: its allow-list was WIDENED to {bad}. That reconciler's "
              f"stated guarantee — prod's real user and the media-users / "
              f"platform-admins / viewers groups survive every run — is expressed as "
              f"endpoints it CANNOT reach, and its own --selftest asserts the refusal of "
              f"/users, /groups, /roles, /authentication and the whole-realm endpoint. "
              f"The passkey work is a SIBLING program with its own narrow list precisely "
              f"so that guarantee stays literally true. Revert the widening")

# ============================== (8) the reconciler's BEHAVIOUR, proven by its selftest
if script:
    with tempfile.TemporaryDirectory() as td:
        p = Path(td) / "reconcile.py"
        p.write_text(script)
        proc = subprocess.run([sys.executable, str(p), "--selftest"],
                              capture_output=True, text=True, timeout=300)
        tail = (proc.stdout + proc.stderr).strip().splitlines()
        check(proc.returncode == 0,
              "the passkey reconciler's --selftest FAILED (compare-then-write / "
              "idempotence / strict read-modify-write / the RpEntityName gate / "
              "flow-is-unwritable / the conditional-OTP regression / fail-closed):\n"
              "        " + "\n        ".join(tail[-14:]))
        if proc.returncode == 0:
            print("  passkey reconciler --selftest: " + (tail[-1] if tail else "(none)"))

        # ---- (8b) THE SELFTEST IS FALSIFIABLE, not decorative ----------------------
        # A passing suite proves nothing unless it can be shown to FAIL. mutations.py
        # injects plausible regressions one at a time -- a partial policy PUT, a
        # rebind of browserFlow, a widened allow-list that makes the flow writable, a
        # fail-open on an unreadable input -- and requires --selftest to catch every
        # one. This matters more than usual here: the whole incident class is a
        # component that REPORTED SUCCESS while the realm was broken, and a green test
        # that cannot fail is the same lie one level up.
        mut = DIR_PATH / "mutations.py"
        if mut.is_file():
            mproc = subprocess.run([sys.executable, str(mut), str(p)],
                                   capture_output=True, text=True, timeout=900)
            mtail = (mproc.stdout + mproc.stderr).strip().splitlines()
            check(mproc.returncode == 0,
                  "the passkey reconciler's --selftest is NOT FALSIFIABLE — at least "
                  "one injected regression survived it, so the suite does not prove "
                  "what it claims:\n        " + "\n        ".join(mtail[-16:]))
            if mproc.returncode == 0:
                print("  passkey reconciler mutations: "
                      + (mtail[-1] if mtail else "(no output)"))
        else:
            check(False,
                  "conformance/tests/l0-static/keycloak-passkey-reconciler/mutations.py "
                  "is missing — nothing proves the --selftest can fail, and an "
                  "unfalsifiable proof is the failure mode this component exists to end")

# ======================================================= (9) this check is load-bearing
L0 = (ROOT / "ci" / "l0.sh").read_text() if (ROOT / "ci" / "l0.sh").is_file() else ""
GATE = (ROOT / "ci" / "gate.sh").read_text() if (ROOT / "ci" / "gate.sh").is_file() else ""
TOUCHED = ((ROOT / "ci" / "touched-requirements.sh").read_text()
           if (ROOT / "ci" / "touched-requirements.sh").is_file() else "")
SELF_YAML = DIR_PATH / "chainsaw-test.yaml"
SELF = SELF_YAML.read_text() if SELF_YAML.is_file() else ""
check(re.search(r'^\s*conformance\.home\.k3s/scope:\s*["\']?whole-repo', SELF, re.M)
      is not None,
      "keycloak-passkey-reconciler/chainsaw-test.yaml does not declare "
      'conformance.home.k3s/scope: "whole-repo", so the merge gate selects this check '
      "only when a change happens to attribute one of the requirements it is labelled "
      "with. This invariant spans platform/, platform-config/, ci/ and clusters/, so a "
      "change that breaks it need not attribute R-IDP-03/04 — which is how 12 unusable "
      "prod OIDC clients sat behind an entirely green cluster on 2026-09-06")
check("--include-scope" in L0,
      "ci/l0.sh no longer passes --include-scope to the conformance runner, so the L0 "
      "stage is a purely requirement-scoped run again and the scope annotation is inert "
      "decoration — this check would then run only by coincidence")
check(re.search(r"^for s in .*\bl0\b.*; do", GATE, re.M)
      and re.search(r"fail\).*BLOCK=1", GATE),
      "ci/gate.sh does not set the merge block when the l0 stage fails")
check(re.search(r"^\s*platform/keycloak/\*\)", TOUCHED, re.M),
      "ci/touched-requirements.sh has no platform/keycloak/* arm, so editing the realm "
      "import or the reconciler would attribute NO requirement and the keycloak tests "
      "would be filtered out of the merge gate's touched subset")

# =============================================================================
if failures:
    print("CHECK FAIL: %d assertion(s) failed" % len(failures), file=sys.stderr)
    for f in failures:
        print("  FAIL %s" % f, file=sys.stderr)
    raise SystemExit(1)
print("keycloak-passkey-reconciler: all assertions passed "
      "(desired state derived from the realm import, one substitution pass, write "
      "surfaces agree, the BOUND flow is unwritable and fail-closed, nothing is "
      "deletable, sibling allow-list unwidened, selftest green and falsifiable)")
PY