home-network-k3s-impl/conformance/tests/l0-static/keycloak-client-secret-reconciler/mutations.py

#!/usr/bin/env python3
# =============================================================================
# mutations.py -- prove the reconciler's --selftest IS FALSIFIABLE.
#
# WHY THIS EXISTS
# ---------------
# A selftest that passes proves nothing on its own: a selftest asserting `True ==
# True` also passes. The reconciler exists because a one-shot import REPORTED
# SUCCESS while the realm was broken, and the same trap applies one level up -- a
# green test suite that cannot fail is the same lie in a different place.
#
# So this injects a series of PLAUSIBLE regressions into the shipped script, one at
# a time, and requires that --selftest FAILS for every one. A mutation that
# SURVIVES means the suite does not actually check that property, and the check
# fails, naming it.
#
# Each mutation is a mistake a reasonable person could make while editing this
# code -- not a nonsense edit. Several were live design questions while it was
# written (order-sensitivity, additive-vs-exact URI sets, whether the two passes
# should share a failure path); the mutation is the record that the answer is
# enforced and not merely commented.
#
# Invoked by check.sh with the script EXTRACTED from the shipped ConfigMap, so what
# is proven falsifiable is byte-for-byte what runs on the cluster.
#
#   python3 mutations.py <path-to-extracted-reconciler.py>
#
# Runtime is ~20s (one subprocess per mutation, each running the full selftest
# against an in-process stub Keycloak; no network, no cluster).
# =============================================================================
from __future__ import annotations

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


def dedent4(text: str) -> str:
    """Anchors are written with the YAML file's indentation. The script extracted from
    the ConfigMap block scalar has 4 fewer leading spaces on every line, so anchors are
    authored to match the YAML (easy to diff against the manifest) and dedented here."""
    return "\n".join(ln[4:] if ln.startswith("    ") else ln for ln in text.split("\n"))


# (name, the mistake being modelled, anchor, replacement)
MUTATIONS: list[tuple[str, str, str, str]] = [
    ("order-sensitive URL comparison",
     "compare redirectUris in the order Keycloak returns them -- Keycloak does not "
     "preserve the authored order, so a converged realm is rewritten on every run "
     "forever while every run still reports success",
     "if sorted(stored) != sorted(declared_value):",
     "if stored != declared_value:"),

    ("fat round-tripped URL PUT",
     "PUT the whole client representation back instead of only the drifted fields -- "
     "the live server rejects it with 400 'Cannot parse the JSON', and even if it did "
     "not, the blast radius would cover every field of the client",
     "            rep = {\"clientId\": client.get(\"clientId\")}\n"
     "            for field in sorted(drift):\n"
     "                rep[field] = want[cid][field]",
     "            rep = dict(client)\n"
     "            for field in sorted(drift):\n"
     "                rep[field] = want[cid][field]"),

    ("write unconditionally (compare-then-write removed)",
     "drop the converged short-circuit -- every client is rewritten on every run, so "
     "the CronJob becomes a permanent write load against the IdP database",
     "            drift = url_drift(client, want[cid])\n            if not drift:",
     "            drift = url_drift(client, want[cid]) or dict(\n"
     "                (f, (\"\", want[cid][f])) for f in want[cid])\n            if False:"),

    ("additive-only URI reconciliation",
     "union the declared URIs with the stored ones -- a redirect URI REMOVED from git "
     "is never removed from the realm, so a stale wildcard keeps working and quietly "
     "widens what the realm will redirect to",
     "                rep[field] = want[cid][field]",
     "                rep[field] = (sorted(set(want[cid][field]) | set(client.get(field) or []))\n"
     "                              if field in URL_LIST_FIELDS else want[cid][field])"),

    ("URL pass fails OPEN on an unreadable input",
     "treat a missing ConfigMap as 'nothing declared' -- the run reports success while "
     "reconciling nothing, which is precisely the fail-open of the realm import",
     "        except OSError as e:\n"
     "            url_input_fail = (\"cannot read %s (%s) -- the keycloak-client-urls ConfigMap \"\n"
     "                              \"is not mounted\" % (URLS_FILE, e.__class__.__name__))",
     "        except OSError:\n            want_urls = {}"),

    ("URL pass skips public clients",
     "copy the secret pass's publicClient skip into the URL pass -- the `kubernetes` "
     "client is PKCE-public, so kubectl's redirect URIs would never converge",
     "            drift = url_drift(client, want[cid])",
     "            if client.get(\"publicClient\"):\n                continue\n"
     "            drift = url_drift(client, want[cid])"),

    ("the two passes are coupled",
     "abort the whole run when either input is bad -- a typo in a hostname ConfigMap "
     "would stop client-SECRET repair, making credential healing depend on a file that "
     "has nothing to do with credentials",
     "        if want_secrets is None and want_urls is None:",
     "        if want_secrets is None or want_urls is None:"),

    ("unmanaged fields forced to empty",
     "iterate every writable field instead of only the declared ones -- a rootUrl or "
     "adminUrl the realm holds and git does not declare would be WIPED, turning "
     "'not declared' into 'must be empty'",
     "        for field, declared_value in sorted(wanted.items()):",
     "        for field in URL_FIELDS:\n"
     "            declared_value = wanted.get(\n"
     "                field, [] if field in URL_LIST_FIELDS else \"\")"),

    ("no readback verification",
     "trust the 204 -- a server that accepts the PUT and does not apply it is reported "
     "as REPAIRED, which is 'reports success while the realm never converged' again",
     "            still = url_drift(after[0], want[cid])",
     "            still = {}"),

    ("allow-list widened to the whole realm",
     "add a realm-level endpoint 'just in case' -- /users, /groups, /roles and "
     "/authentication stop being structurally unreachable",
     "        r\"/admin/realms/[^/]+/clients\",",
     "        r\"/admin/realms/[^/]+\",\n        r\"/admin/realms/[^/]+/clients\","),

    ("desired-state field validation removed",
     "accept any field name from the ConfigMap -- an arbitrary client field becomes "
     "writable through a file edit, bypassing the five-field write surface entirely",
     "                if field not in URL_FIELDS:",
     "                if False:"),

    ("URL drift detected but never repaired",
     "leave the audit-mode early-continue in place on the write path -- drift is "
     "reported in the log and silently not fixed",
     "            if audit:\n                continue\n            # MINIMAL representation",
     "            if True:\n                continue\n            # MINIMAL representation"),
]


def main() -> int:
    if len(sys.argv) != 2:
        print("usage: mutations.py <extracted-reconciler.py>", file=sys.stderr)
        return 2
    src = Path(sys.argv[1]).read_text()

    problems: list[str] = []
    killed = 0
    for name, why, old, new in MUTATIONS:
        old, new = dedent4(old), dedent4(new)
        # An anchor that no longer matches is a FAILURE, not a skip: it means the code
        # moved and this mutation silently stopped proving anything.
        n = src.count(old)
        if n != 1:
            problems.append("%s: anchor matches %d times (expected exactly 1) -- the "
                            "code moved and this mutation no longer proves anything. "
                            "Re-anchor it against the current source." % (name, n))
            continue
        with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False) as fh:
            fh.write(src.replace(old, new, 1))
            path = fh.name
        try:
            proc = subprocess.run([sys.executable, path, "--selftest"],
                                  capture_output=True, text=True, timeout=240)
        finally:
            Path(path).unlink(missing_ok=True)
        if proc.returncode == 0:
            problems.append("%s SURVIVED -- --selftest still passed with this regression "
                            "applied, so nothing in the suite actually checks it. The "
                            "mistake it models: %s" % (name, why))
            continue
        killed += 1
        out = proc.stdout + proc.stderr
        m = re.search(r"SELFTEST FAIL: (\d+) of \d+", out)
        how = ("%s assertion(s) caught it" % m.group(1) if m
               else "the selftest aborted (caught, less cleanly)")
        print("  killed  %-42s %s" % (name, how))

    print()
    if problems:
        print("MUTATION CHECK FAIL: %d of %d mutation(s) not proven"
              % (len(problems), len(MUTATIONS)), file=sys.stderr)
        for p in problems:
            print("  FAIL %s" % p, file=sys.stderr)
        return 1
    print("MUTATION CHECK OK: all %d injected regressions were caught by --selftest "
          "(the suite is falsifiable, not decorative)" % killed)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())