#!/usr/bin/env python3
"""
creds-matrix.py -- D-137 credential-matrix checker (ADOPTED 2026-07-26).

The matrix (`creds-matrix.tsv`) is the EXPECTED-state register of every credential this
cloud is supposed to have. This script reconciles it against what the repo actually
declares. It exists because `creds-audit` is DECLARATION-based -- the manifest IS the
declaration, so an undeclared secret is structurally invisible -- and because no discovery
sweep can EVER detect a credential that was never minted at all. Absence is invisible to
discovery; only a forward register makes it a FAIL.

THREE TIERS, increasing cost (plan: docs/D-137-implementation-plan.md):
  1. STATIC     offline; runs in the gauntlet                    <-- BUILT (default)
  2. EXISTENCE  local + remote reads; the preflight gate          <-- BUILT (--tier2)
  3. VALIDITY   provenance digests + behavioral probes           <-- NOT BUILT YET
A tier that did not run SELF-SKIPS with an explicit [ok] naming it. Nothing here ever
passes silently -- a silent pass is the false-green this whole design exists to prevent,
and the same applies to a location that could not be read (see the probe states below).

METADATA ONLY. This script reads the matrix, the manifests, and the notes file. It never
opens a credential, and it must never gain a content-reading verb (its harness greps the
source to enforce that, mirroring tests/creds-audit/run-tests.sh:71-73).

EXPECTED RED BY DESIGN. A clean run against today's tree would mean the checker is wrong.
See the plan's "EXPECTED RED BY DESIGN" section: the ruling-5 identity conflation, the 12
operator-terminal reproducibility-debt rows, and SEC-021's per-DC divergence are DEFECTS
the register surfaces, not bugs to silence. Do not delete a row to go green.

Usage:
  python3 scripts/creds-matrix.py                 # tier 1 (static) -- the gauntlet entry
  python3 scripts/creds-matrix.py --all           # every implemented tier + skip notices
  python3 scripts/creds-matrix.py --dc vr1-dc1    # narrow the per-DC checks to one DC
  python3 scripts/creds-matrix.py --render        # print manifests rendered from the matrix
  python3 scripts/creds-matrix.py --tier2 --remote --privileged \
      --pending-stage vr0-phase01 ...             # the full three-site existence sweep
Exit: 0 clean, 1 findings, 2 usage/IO error. ASCII-only output.
"""
import sys, os, re, glob, shlex, fnmatch, argparse, subprocess, socket

COLUMNS = ("id", "cardinality", "site-key", "host-role", "filename", "access-type",
           "principal", "custody", "mint-stage", "mint-ref", "sec-ref", "notes-ref")
NCOL = len(COLUMNS)

CARDINALITY = {"singleton", "per-site", "per-DC", "per-node", "per-tenant"}
# `region` added 2026-07-30 for the D-132-amendment per-DC MAAS region controller VM
# (one per DC, utility-band .6). It is NOT `headend` -- that role is the Office1 region --
# and NOT `rack`, which is the DC rack controller the region is deliberately separate from.
HOST_ROLE   = {"jumphost", "headend", "rack", "region", "edge", "netbox", "cloud", "unit",
               "client", "-"}
ACCESS_TYPE = {"gui", "api", "ssh", "cli-profile", "console", "none"}
PRINCIPAL   = {"human", "service", "-"}
CUSTODY     = {"consolidated", "source-of-record", "off-manifest-known",
               "not-consolidated-ruled", "verbatim-copy"}
# Custodies whose files live IN the site creds folder, and therefore appear in its
# (generated) manifest. `verbatim-copy` adds a byte-identity assertion for tier 3; it does
# not move the file out of the folder.
IN_FOLDER = {"consolidated", "verbatim-copy"}

# site-key discipline imported from scripts/lib-net.sh:160-162 / lib-hosts.sh:153-155:
# region-qualified only. A bare `dc0` is REJECTED -- it is the drifting form that has
# repeatedly meant two different things across the two DCs.
SITE_KEY_RE = re.compile(r"^(vr1-dc[01]|vr1-office1)$")
MINT_REF_RE = re.compile(r"^(script|runbook):([^:]+):(\d+)$")
DC_TOKEN_RE = re.compile(r"dc[01]")

# The manifest mode column is DERIVED, not a matrix field: a public key half is 0644,
# everything else 0600. That rule reproduces all three checked-in manifests exactly, so
# it stays a derivation rather than a 13th column.
# A filename may carry a `<token>` placeholder for a per-tenant / per-node instance name
# (`<client>-svc-cred.txt`). Without this the per-tenant rows could never match a real
# basename, so at the first tenant onboard they would have produced five false absences
# AND five false undeclared findings on the SAME five real files -- noise that trains
# operators to ignore the most sensitive rows in the register (committee finding).
TEMPLATE_RE = re.compile(r"<[a-z]+>")


def fn_matches(declared, actual):
    """Exact match, unless the declared name carries a `<token>` instance placeholder.

    Per-tenant and per-node rows name an instance they cannot know at authoring time.
    Without this the tenant rows could never match a real basename, so the first tenant
    onboard would have produced five false absences AND five false undeclared findings on
    the SAME five real files -- noise that trains operators to ignore the most sensitive
    rows in the register (committee finding)."""
    if "<" not in declared:
        return declared == actual
    return fnmatch.fnmatchcase(actual, TEMPLATE_RE.sub("*", declared))


def derived_mode(filename):
    return "644" if filename.endswith(".pub") else "600"


class Row(object):
    def __init__(self, fields, lineno):
        self.lineno = lineno
        for name, value in zip(COLUMNS, fields):
            setattr(self, name.replace("-", "_"), value)

    def key(self):
        # `id` is part of the key on purpose: two DIFFERENT identities may legitimately
        # share one file (the Vault init artifact holds both the unseal shares and the
        # root token). What is never legitimate is the SAME identity declared twice at
        # the same location.
        return (self.id, self.site_key, self.host_role, self.filename)

    def has_artifact(self):
        # `filename == "-"` means the credential has no host artifact (it lives in the
        # cloud, on a unit, or -- the juju-user case -- nowhere at all). Existence-type
        # checks must SKIP such rows, never fault them.
        return self.filename != "-"


def load_matrix(path, fails):
    rows = []
    try:
        with open(path) as fh:
            lines = fh.readlines()
    except IOError as exc:
        sys.stderr.write("ERROR: cannot read matrix %s: %s\n" % (path, exc)); sys.exit(2)
    for n, line in enumerate(lines, 1):
        text = line.split("#", 1)[0].strip()
        if not text:
            continue
        fields = text.split()
        if len(fields) != NCOL:
            fails.append("matrix:%d has %d field(s), want %d -- a field with whitespace "
                         "in it would also land here: %s" % (n, len(fields), NCOL, text[:60]))
            continue
        rows.append(Row(fields, n))
    return rows


def load_manifest(path):
    """Return {filename: (mode, source)} for a checked-in manifest, or None if absent."""
    if not os.path.exists(path):
        return None
    out = {}
    with open(path) as fh:
        for line in fh:
            text = line.split("#", 1)[0].strip()
            if not text:
                continue
            parts = text.split()
            if len(parts) >= 3:
                out[parts[0]] = (parts[1], parts[2])
            elif len(parts) == 2:
                out[parts[0]] = (parts[1], "-")
            else:
                # Do NOT drop it: a bare filename with no mode used to vanish, so the
                # undeclared-file bound never fired for it -- exactly T8's defect class.
                out[parts[0]] = ("?", "?")
    return out


def load_notes(path):
    keys = set()
    if not os.path.exists(path):
        return None
    with open(path) as fh:
        for line in fh:
            if line.startswith("## "):
                keys.add(line[3:].strip())
    return keys


# --------------------------------------------------------------------------- tier 1
def s1_schema(rows, oks, fails):
    bad = 0
    if not rows:
        # A register with no rows asserts nothing, yet every check below reports clean
        # over an empty set. That is the both-bounds failure applied to the input itself.
        fails.append("S1 FLOOR: the matrix has ZERO rows -- an empty register cannot be "
                     "clean, it is unpopulated")
        return
    for r in rows:
        where = "matrix:%d %s" % (r.lineno, r.id)
        if r.cardinality not in CARDINALITY:
            fails.append("%s cardinality '%s' invalid" % (where, r.cardinality)); bad += 1
        if r.site_key != "-" and not SITE_KEY_RE.match(r.site_key):
            fails.append("%s site-key '%s' is not region-qualified (bare 'dc0' is REJECTED)"
                         % (where, r.site_key)); bad += 1
        if r.host_role not in HOST_ROLE:
            fails.append("%s host-role '%s' invalid (must be a role token, never a VM name)"
                         % (where, r.host_role)); bad += 1
        if r.access_type not in ACCESS_TYPE:
            fails.append("%s access-type '%s' invalid" % (where, r.access_type)); bad += 1
        if r.principal not in PRINCIPAL:
            fails.append("%s principal '%s' invalid" % (where, r.principal)); bad += 1
        if r.custody not in CUSTODY:
            fails.append("%s custody '%s' invalid" % (where, r.custody)); bad += 1
    seen = {}
    for r in rows:
        if r.has_artifact():
            if r.key() in seen:
                fails.append("matrix:%d duplicate (id,site,host-role,filename) row -- also "
                             "at matrix:%d" % (r.lineno, seen[r.key()])); bad += 1
            seen[r.key()] = r.lineno
    if not bad:
        oks.append("S1 schema: %d rows, all enums valid, site-keys region-qualified, "
                   "no duplicate (id,site,host-role,filename)" % len(rows))


def s2_manifest_coverage(rows, manifest_dir, oks, fails):
    """BOTH BOUNDS. Imported from netbox/sandbox-fidelity-check.py:131-143, which records
    the identical false-green in-repo: a subset-only check 'was an upper bound
    masquerading as an assertion. It must be BOTH bounds.' Here the same failure is one
    level up -- enumerate no expectations and nothing is ever undeclared, so the register
    reads CLEAN while declaring nothing."""
    # BOTH BOUNDS AT THE SITE LEVEL TOO. Enumerating only the sites the MATRIX mentions
    # is the same false-green one level up: an empty (or site-truncated) matrix would
    # examine no manifest, find nothing undeclared, and report CLEAN while every declared
    # credential sat unregistered. The site set is the UNION of both sides.
    sites = {r.site_key for r in rows if r.site_key != "-"}
    if os.path.isdir(manifest_dir):
        for entry in os.listdir(manifest_dir):
            if entry.endswith(".manifest"):
                sites.add(entry[:-len(".manifest")])
    sites = sorted(sites)
    total_missing = total_extra = total_drift = 0
    for site in sites:
        manifest = load_manifest(os.path.join(manifest_dir, "%s.manifest" % site))
        if manifest is None:
            fails.append("S2 %s: no manifest file -- every site in the matrix needs one"
                         % site)
            continue
        # A manifest declares exactly the CONSOLIDATED jumphost copies for its site.
        expected = {r.filename: r for r in rows
                    if r.site_key == site and r.custody in IN_FOLDER
                    and r.host_role == "jumphost" and r.has_artifact()}
        # bound 1: expected-but-absent
        for fname, r in sorted(expected.items()):
            if fname not in manifest:
                total_missing += 1
                fails.append("S2 %s EXPECTED-BUT-ABSENT: '%s' (id %s%s) is expected by the "
                             "matrix but NOT declared in the manifest -- the credential is "
                             "either missing or undeclared"
                             % (site, fname, r.id, "" if r.sec_ref == "-" else ", " + r.sec_ref))
        # bound 2: declared but unexpected
        for fname in sorted(manifest):
            if fname not in expected:
                total_extra += 1
                fails.append("S2 %s UNEXPECTED-EXTRA: manifest declares '%s' with no "
                             "matching matrix row -- add it to the register or remove it"
                             % (site, fname))
        # field drift on the rows present in both
        for fname, r in sorted(expected.items()):
            if fname not in manifest:
                continue
            want = derived_mode(fname)
            if manifest[fname][0] != want:
                total_drift += 1
                fails.append("S2 %s FIELD DRIFT: '%s' manifest mode %s, matrix derives %s"
                             % (site, fname, manifest[fname][0], want))
        # rows the matrix marks as deliberately NOT in the folder must not appear in it
        for r in rows:
            if (r.site_key == site and r.host_role == "jumphost" and r.has_artifact()
                    and r.custody not in IN_FOLDER and r.filename in manifest):
                total_drift += 1
                fails.append("S2 %s FIELD DRIFT: '%s' is custody=%s yet declared in the "
                             "manifest" % (site, r.filename, r.custody))
    if not (total_missing or total_extra or total_drift):
        oks.append("S2 manifest coverage: both bounds clean across %d site(s)" % len(sites))


def s3_render_drift(rows, manifest_dir, notes, oks, fails):
    """D-137 ruling 2: manifests are DERIVED; checked-in must equal rendered."""
    sites = sorted({r.site_key for r in rows if r.site_key != "-"})
    drift = pending = 0
    for site in sites:
        manifest = load_manifest(os.path.join(manifest_dir, "%s.manifest" % site))
        if manifest is None:
            continue
        for fname, mode, source, r in render_rows(rows, site):
            if fname not in manifest:
                continue                       # already reported by S2
            if manifest[fname][0] != mode:
                drift += 1
                fails.append("S3 %s render drift: '%s' mode checked-in=%s rendered=%s"
                             % (site, fname, manifest[fname][0], mode))
            if source == "PENDING-TIER2":
                # The source field for a VM-minted credential needs the declared PATH,
                # which ruling 3 puts in creds-manifests/vm-secret-locations -- a tier-2
                # artifact. Skip the field EXPLICITLY rather than let it read as matching.
                pending += 1
                continue
            if manifest[fname][1] != source:
                drift += 1
                fails.append("S3 %s render drift: '%s' source checked-in=%s rendered=%s"
                             % (site, fname, manifest[fname][1], source))
    if pending:
        oks.append("S3 render: %d source field(s) SKIPPED -- rendering them needs the "
                   "declared path from creds-manifests/vm-secret-locations (ruling 3); the "
                   "list now EXISTS but the source-field derivation is not wired" % pending)
    if not drift:
        # SCOPE, stated so this [ok] cannot be over-read: S3 compares
        # PARSED ROW FIELDS (mode, source) for rows present on both sides. It does NOT
        # compare manifest header prose or field-definition blocks, and render emits only
        # consolidated jumphost rows -- so ruling 2's "checked-in == rendered" obligation
        # is only PARTLY gated here. Full byte-equality becomes enforceable when the
        # manifests are actually replaced by rendered output (operator-gated; see the
        # 2026-07-26 changelog, item 3).
        oks.append("S3 render drift: rendered row fields (mode, source) match checked-in; "
                   "header prose and non-jumphost rows are OUT OF SCOPE of this compare")


def render_rows(rows, site):
    """Yield (filename, mode, source, row) for one site's rendered manifest."""
    src_of_record = {}
    for r in rows:
        if r.custody == "source-of-record":
            src_of_record.setdefault(r.id, r)
    out = []
    for r in rows:
        if (r.site_key == site and r.custody in IN_FOLDER
                and r.host_role == "jumphost" and r.has_artifact()):
            source = "PENDING-TIER2" if r.id in src_of_record else "local"
            out.append((r.filename, derived_mode(r.filename), source, r))
    # Sort on the rendered fields only -- Row is not orderable, and a duplicate row
    # (T5) would otherwise reach a Row-vs-Row comparison and crash before S1's finding
    # could be printed.
    return sorted(out, key=lambda t: t[:3])


def s4_mint_ref(rows, repo, oks, fails):
    bad = 0
    terminal = 0
    for r in rows:
        ref = r.mint_ref
        if ref == "-":
            continue
        if ref == "operator-terminal":
            terminal += 1
            continue
        m = MINT_REF_RE.match(ref)
        if not m:
            fails.append("matrix:%d %s mint-ref '%s' is not script:<path>:<line>, "
                         "runbook:<path>:<line>, or operator-terminal"
                         % (r.lineno, r.id, ref)); bad += 1
            continue
        _, path, line = m.groups()
        full = os.path.join(repo, path)
        if not os.path.exists(full):
            fails.append("matrix:%d %s mint-ref target does not exist: %s"
                         % (r.lineno, r.id, path)); bad += 1
            continue
        if not os.path.isfile(full):
            fails.append("matrix:%d %s mint-ref target is not a file: %s"
                         % (r.lineno, r.id, path)); bad += 1
            continue
        try:
            with open(full, errors="replace") as fh:
                n = sum(1 for _ in fh)
        except OSError as exc:
            fails.append("matrix:%d %s mint-ref target unreadable (%s): %s"
                         % (r.lineno, r.id, exc.__class__.__name__, path)); bad += 1
            continue
        if int(line) > n:
            fails.append("matrix:%d %s mint-ref %s points past EOF (file has %d lines)"
                         % (r.lineno, r.id, ref, n)); bad += 1
    if not bad:
        oks.append("S4 mint-ref: every script:/runbook: reference resolves to a real "
                   "location")
    # Research FINDING 1 is a DEBT REPORT, not a failure: `ssh-keygen` returns zero hits
    # repo-wide, so these rows genuinely have no mint command. Admitting them is the point.
    if terminal:
        oks.append("S4 provenance debt: %d row(s) are mint-ref=operator-terminal -- NOT "
                   "reproducible from the repo (research FINDING 1). Admitted by design; "
                   "converting them is remediation, not a checker fix." % terminal)


def s5_per_dc_symmetry(rows, dc_filter, oks, fails):
    """Per-DC rows must be SYMMETRIC. sec-ref and notes-ref are excluded from the
    comparison -- the two DCs legitimately carry different SEC rows for the same shape."""
    dcs = ["vr1-dc0", "vr1-dc1"]
    if dc_filter:
        oks.append("S5 per-DC symmetry: --dc %s narrows the run; symmetry needs BOTH DCs, "
                   "so the check is SKIPPED (explicitly, not passed)" % dc_filter)
        return
    shapes = {}
    for dc in dcs:
        shapes[dc] = set()
        for r in rows:
            if r.site_key != dc or r.cardinality != "per-DC":
                continue
            shapes[dc].add((
                DC_TOKEN_RE.sub("dcN", r.id),
                DC_TOKEN_RE.sub("dcN", r.filename),
                r.host_role, r.access_type, r.principal, r.custody, r.mint_stage,
                DC_TOKEN_RE.sub("dcN", r.mint_ref),
            ))
    only0 = shapes["vr1-dc0"] - shapes["vr1-dc1"]
    only1 = shapes["vr1-dc1"] - shapes["vr1-dc0"]
    for label, extra, missing_from in (("vr1-dc0", only0, "vr1-dc1"),
                                       ("vr1-dc1", only1, "vr1-dc0")):
        for shape in sorted(extra):
            fails.append("S5 ASYMMETRY: %s declares id=%s file=%s on %s (custody=%s) with "
                         "no counterpart in %s -- a per-DC credential must exist at BOTH "
                         "DCs in the same shape"
                         % (label, shape[0], shape[1], shape[2], shape[5], missing_from))
    if not (only0 or only1):
        oks.append("S5 per-DC symmetry: %d shared per-DC shape(s), no asymmetry"
                   % len(shapes["vr1-dc0"]))


def s6_principal_invariant(rows, oks, fails):
    """D-137 ruling 5: ONE IDENTITY SERVES ONE PRINCIPAL TYPE. Rows sharing an `id` are
    forms/copies of one identity, so two differing principals across them IS the
    conflation. Rows with principal '-' (non-credential artifacts) are skipped."""
    by_id = {}
    for r in rows:
        if r.principal == "-":
            continue
        by_id.setdefault(r.id, {}).setdefault(r.principal, []).append(r)
    bad = 0
    for ident in sorted(by_id):
        principals = by_id[ident]
        if len(principals) > 1:
            bad += 1
            detail = "; ".join(
                "%s via %s" % (p, ", ".join(sorted({x.access_type for x in rs})))
                for p, rs in sorted(principals.items()))
            fails.append("S6 IDENTITY CONFLATION: id '%s' serves %d principal types (%s) "
                         "-- ruling 5 requires one identity to serve one principal type"
                         % (ident, len(principals), detail))
    if not bad:
        oks.append("S6 principal invariant: every identity serves exactly one principal "
                   "type (%d identities checked)" % len(by_id))


def s7_notes_ref(rows, notes, notes_path, oks, fails):
    if notes is None:
        fails.append("S7 notes file missing: %s -- the 12-column amendment makes it the "
                     "home for every governance note ruling 2 forbids dropping" % notes_path)
        return
    bad = 0
    used = set()
    for r in rows:
        if r.notes_ref == "-":
            continue
        used.add(r.notes_ref)
        if r.notes_ref not in notes:
            fails.append("matrix:%d %s notes-ref '%s' has no '## %s' entry in the notes file"
                         % (r.lineno, r.id, r.notes_ref, r.notes_ref)); bad += 1
    orphans = sorted(k for k in notes if k.startswith("n-") and k not in used)
    for k in orphans:
        fails.append("S7 orphan note '%s' is referenced by no matrix row" % k)
    if not bad and not orphans:
        oks.append("S7 notes: %d note key(s) referenced, all resolve, none orphaned"
                   % len(used))


# --------------------------------------------------------------------------- tier 2
# A declared path is interpolated into a remote shell command, so its charset is a
# security boundary, not a style rule: ruling 3 calls the bound ABSOLUTE, and an
# unvalidated `;` or `$(...)` in this file would execute on the remote host -- under
# `sudo -n` when --privileged is passed. Shell metacharacters are rejected outright and
# the path is additionally shell-quoted at use.
SAFE_PATH_RE = re.compile(r"^[A-Za-z0-9_@%+=:,./~*-]+$")
SAFE_TARGET_RE = re.compile(r"^[A-Za-z0-9_.-]+$")       # no leading '-', no ssh options


def load_stages(path, rows, fails):
    """The declared reach of each mint stage (D-137 tier 2). Returns {token: bool} or None.

    EXHAUSTIVE BY CONSTRUCTION: every mint-stage token the matrix uses must be classified
    here. An unclassified token is a hard FAIL, never a default -- defaulting to `reached`
    floods the run with false absences, and defaulting to `pending` silently defers real
    ones, which is the false-green class this control exists to prevent."""
    if not os.path.exists(path):
        return None
    out = {}
    for n, line in enumerate(open(path), 1):
        text = line.split("#", 1)[0].strip()
        if not text:
            continue
        parts = text.split()
        if len(parts) != 2 or parts[1] not in ("reached", "pending"):
            fails.append("stages-reached:%d must be '<stage> reached|pending', got: %s"
                         % (n, text[:50])); continue
        out[parts[0]] = (parts[1] == "reached")
    used = {r.mint_stage for r in rows}
    for tok in sorted(used - set(out)):
        fails.append("stages-reached: mint-stage '%s' is used in the matrix but NOT "
                     "classified -- classify it reached|pending; the checker will not "
                     "default it in either direction" % tok)
    for tok in sorted(set(out) - used):
        fails.append("stages-reached: '%s' is classified but used by no matrix row -- "
                     "stale declaration" % tok)
    return out


def load_host_identity(path, locations, fails):
    """Which physical host each role's `local` locations live on. Returns {role: host} or None.

    An ssh-target of `local` means "this machine", which is only correct when the checker runs
    on the host the ROLE denotes. Measured 2026-07-29: running on the headend probed the
    jumphost's declared paths against the headend's own SEC-022 shadow stores and reported 26
    artefact findings as fact. Absence asserted over the wrong filesystem is worse than
    "could not look" -- so the binding is declared, and unbound roles REFUSE.

    EXHAUSTIVE BY CONSTRUCTION, like load_stages: every role using a `local` target must be
    declared, and a declared role never used locally is a stale declaration."""
    if not os.path.exists(path):
        return None
    out = {}
    for n, line in enumerate(open(path), 1):
        text = line.split("#", 1)[0].strip()
        if not text:
            continue
        parts = text.split()
        if len(parts) != 2:
            fails.append("host-identity:%d must be '<host-role> <hostname>', got: %s"
                         % (n, text[:50])); continue
        role, host = parts
        if role not in HOST_ROLE:
            fails.append("host-identity:%d host-role '%s' invalid (must be a role token, "
                         "never a VM name)" % (n, role)); continue
        if role in out:
            fails.append("host-identity:%d duplicate declaration for role '%s'"
                         % (n, role)); continue
        out[role] = host
    local_roles = {role for role, _s, target, _p in locations if target == "local"}
    for role in sorted(local_roles - set(out)):
        fails.append("host-identity: role '%s' uses a `local` target in vm-secret-locations "
                     "but is NOT bound to a host -- bind it; the checker will not assume "
                     "that the machine it runs on is that role's host" % role)
    for role in sorted(set(out) - local_roles):
        fails.append("host-identity: role '%s' is bound to a host but uses no `local` "
                     "target -- stale declaration" % role)
    return out


def load_locations(path, repo, fails):
    """The ruling-3 declared-location list.
    Returns [(role, site, ssh_target, path)], or None if the file is ABSENT."""
    if not os.path.exists(path):
        return None
    out = []
    for n, line in enumerate(open(path), 1):
        text = line.split("#", 1)[0].strip()
        if not text:
            continue
        parts = text.split()
        if len(parts) != 4:
            fails.append("vm-secret-locations:%d has %d field(s), want 4 "
                         "(<host-role> <site-key> <ssh-target> <path>)" % (n, len(parts)))
            continue
        role, site, target, p = parts
        if site != "-" and not SITE_KEY_RE.match(site):
            fails.append("vm-secret-locations:%d site-key '%s' is not region-qualified"
                         % (n, site)); continue
        if not SAFE_TARGET_RE.match(target):
            fails.append("vm-secret-locations:%d ssh-target '%s' has unsafe characters "
                         "-- it is passed to ssh argv" % (n, target)); continue
        # The repo has been renamed once (D-110); its name is never hardcoded.
        p = p.replace("<repo>", repo)
        if not SAFE_PATH_RE.match(p):
            fails.append("vm-secret-locations:%d path '%s' contains a shell metacharacter "
                         "-- rejected: this string reaches a remote shell" % (n, p))
            continue
        out.append((role, site, target, p))
    return out


# Probes return (state, {basename: mode}) where state is one of:
#   "ok"          the location was read
#   "absent"      the directory does not exist (a stage not yet reached, usually)
#   "unreadable"  it EXISTS but could not be listed -- typically root-owned
#   "unreachable" the host could not be contacted
#
# `unreadable` exists as its own state because of a false green found during the build:
# a shell glob over an unreadable directory does not expand, the loop finds no files, and
# the probe would have returned an EMPTY listing -- reporting "nothing undeclared here"
# about a directory it was never able to open. The region secrets dir (SEC-020) is exactly
# such a directory. "Could not look" must never be recorded as "nothing there".
def _dir_of(pattern):
    """The directory whose readability decides this location's state.

    For `dir/*` that is the directory. For a LITERAL FILE path it is the containing
    directory -- previously this returned None for literal paths, so they skipped the
    absent/unreadable determination entirely and an unreadable directory read as
    'fully probed, nothing there' (committee finding: the same false green T33 was
    written to kill, surviving in the other path form)."""
    # Both forms want the same thing: dirname("a/b/*") == "a/b" (the directory whose
    # readability we test) and dirname("a/b/file") == "a/b" (the file's parent). An
    # earlier rewrite rstripped "/*" first and climbed one level too far.
    return os.path.dirname(pattern)


def probe_local(pattern):
    """METADATA ONLY -- stat, never file content."""
    d = _dir_of(pattern)
    if d:
        dx = os.path.expanduser(d)
        # A glob (~/tenant-*/*) may legitimately match nothing; only a literal
        # directory path can be judged absent or unreadable.
        if not glob.has_magic(dx):
            if not os.path.isdir(dx):
                # CANNOT conclude "absent" if the PARENT is not traversable: to an
                # unprivileged reader a root-owned directory is indistinguishable from a
                # missing one. Reporting it absent would be a benign-looking skip over a
                # location that exists and holds secrets -- measured on the region
                # secrets dir (SEC-020), which read "does not exist" on the first sweep.
                parent = os.path.dirname(dx.rstrip("/")) or "/"
                if not os.path.isdir(parent):
                    return ("absent", {})          # the whole tree is absent
                if not os.access(parent, os.R_OK | os.X_OK):
                    return ("unreadable", {})
                return ("absent", {})
            if not os.access(dx, os.R_OK | os.X_OK):
                return ("unreadable", {})
    found = {}
    for f in glob.glob(os.path.expanduser(pattern)):
        if os.path.isfile(f):
            found[os.path.basename(f)] = oct(os.stat(f).st_mode & 0o777)[2:]
    return ("ok", found)


def probe_remote(target, pattern, privileged=False):
    """METADATA ONLY over ssh -- stat, never file content.

    `privileged` prefixes `sudo -n` and is used ONLY as an escalation retry after an
    unprivileged probe returned `unreadable`. Escalating only where it is actually needed
    keeps the privileged surface as small as the ruling-3 bound already keeps the search
    surface. It still reads metadata only: mode and name, never content."""
    d = _dir_of(pattern)
    probe = ""
    if d and not glob.has_magic(d):
        # See probe_local: "not a directory" is only ABSENT when the parent is
        # traversable; otherwise the truth is that we could not look. Guarded by
        # has_magic because a multi-word glob expansion makes `[ -d ... ]` a syntax
        # error, which silently disabled this whole state machine (committee finding).
        q = d   # validated by SAFE_PATH_RE; NOT quoted -- see note above
        probe = ('p=$(dirname %s); '
                 'if [ ! -d %s ]; then '
                 '  if [ ! -d "$p" ]; then echo "__STATE__ absent"; '
                 '  elif [ -r "$p" ] && [ -x "$p" ]; then echo "__STATE__ absent"; '
                 '  else echo "__STATE__ unreadable"; fi; '
                 'elif [ ! -r %s ] || [ ! -x %s ]; then echo "__STATE__ unreadable"; fi; '
                 % (q, q, q, q))
    # __DONE__ is the liveness marker. Without it, a loop that simply matched nothing
    # exits non-zero with empty output and was reported as UNREACHABLE -- which gated the
    # whole role and converted every absence FAIL at that role into an [ok]. Reaching
    # __DONE__ proves the command ran on the host, so "found nothing" is a real
    # observation and only its ABSENCE means we could not reach the host.
    cmd = probe + ('for f in %s; do [ -f "$f" ] && stat -c "%%n %%a" "$f"; done; '
                   'echo __DONE__' % pattern)   # pattern validated, deliberately unquoted
    if privileged:
        cmd = "sudo -n sh -c " + shlex.quote(cmd)
    try:
        p = subprocess.run(["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=10",
                            target, cmd], capture_output=True, text=True, timeout=60)
    except (OSError, subprocess.SubprocessError):
        return ("unreachable", {})
    if p.returncode == 255:                      # ssh transport failure
        return ("unreachable", {})
    found = {}
    done = False
    for line in p.stdout.splitlines():
        if line.startswith("__STATE__ "):
            return (line.split(None, 1)[1].strip(), {})
        if line.strip() == "__DONE__":
            done = True
            continue
        bits = line.rsplit(None, 1)
        if len(bits) == 2:
            found[os.path.basename(bits[0])] = bits[1]
    if not done:
        return ("unreachable", {})
    return ("ok", found)


def tier2_existence(rows, locations, use_remote, use_privileged, stages,
                    host_identity, this_host, oks, fails):
    """EXISTENCE: everything expected is present at its expected role, and nothing
    undeclared sits at any declared location -- including the remote ones (ruling 3)."""
    # SCOPE = (host-role, site-key), NOT host-role alone. Keying on role alone merged all
    # three jumphost creds folders into one basename namespace, so a credential present at
    # ANY site satisfied the row for EVERY site -- which is how SEC-021's dc0
    # opnsense-api.txt absence passed a green sweep while tier 1 flagged it. Four committee
    # lenses converged on this. A location with site `-` is not site-scoped and serves rows
    # that are likewise `-`.
    by_scope = {}
    for role, site, target, pat in locations:
        by_scope.setdefault((role, site), []).append((target, pat))

    observed = {}          # (role, site) -> {basename: mode}
    # A role is COMPLETE only if EVERY declared location for it was probed. Absence can
    # only be asserted over a fully-probed role: with one location skipped, the other
    # locations' listings would "cover" the role and manufacture false EXPECTED-BUT-ABSENT
    # findings for credentials that live in the location nobody looked at. Measured on the
    # first live sweep, where skipping the region secrets dir made every credential in it
    # report as missing from the headend.
    incomplete = set()
    for role, site, _t, _p in locations:
        observed.setdefault((role, site), {})
    skipped_remote = 0
    for role, site, target, pat in locations:
        scope = (role, site)
        if target != "local":
            if not use_remote:
                skipped_remote += 1
                incomplete.add(scope)
                continue
            state, seen = probe_remote(target, pat)
            if state == "unreadable" and use_privileged:
                pstate, pseen = probe_remote(target, pat, privileged=True)
                if pstate == "ok":
                    oks.append("E0 %s location '%s' read via ESCALATION (sudo -n, metadata "
                               "only) after an unprivileged probe could not open it"
                               % (role, pat))
                    state, seen = pstate, pseen
                else:
                    fails.append("E0 ESCALATION FAILED: %s location '%s' is still "
                                 "unreadable with sudo -n (%s) -- still no conclusion "
                                 "about its contents" % (role, pat, pstate))
        else:
            # `local` is only meaningful ON the host the ROLE denotes. Probing it anywhere
            # else compares this role's expected set against a DIFFERENT machine's
            # filesystem and reports the mismatches as findings -- measured 2026-07-29 as
            # 26 artefact findings when P5 ran on the headend instead of the jumphost.
            # An unbound role is NOT probed either: the checker never assumes the machine
            # it happens to run on is the right one.
            declared = (host_identity or {}).get(role)
            if declared is None:
                incomplete.add(scope)
                oks.append("E0 %s location '%s' NOT PROBED -- role '%s' has no host binding "
                           "in creds-manifests/host-identity, so `local` cannot be resolved; "
                           "absence is not asserted over it" % (role, pat, role))
                continue
            if declared.split(".")[0].lower() != this_host.split(".")[0].lower():
                incomplete.add(scope)
                oks.append("E0 %s location '%s' NOT PROBED -- role '%s' lives on '%s' and "
                           "this run is on '%s'; probing it here would measure a different "
                           "host's filesystem, so absence is not asserted over it"
                           % (role, pat, role, declared, this_host))
                continue
            state, seen = probe_local(pat)
        # "absent" is a CONCLUSIVE observation -- we looked and the location is not there,
        # so it hides nothing. Only "could not look" states gate the role.
        if state in ("unreadable", "unreachable"):
            incomplete.add(scope)
        if state == "absent":
            oks.append("E0 %s location '%s' does not exist -- SKIPPED (typically a mint "
                       "stage this deployment has not reached)" % (role, pat))
            continue
        if state == "unreadable":
            # NOT a pass and NOT an empty listing: the location exists and could not be
            # opened, so nothing may be concluded about what is or is not inside it.
            fails.append("E0 UNREADABLE: %s location '%s' EXISTS but could not be listed "
                         "-- no conclusion may be drawn about its contents; re-run with "
                         "sufficient privilege" % (role, pat))
            continue
        if state == "unreachable":
            oks.append("E0 %s (%s) UNREACHABLE -- location '%s' SKIPPED explicitly; "
                       "an unreachable host is never a pass" % (role, target, pat))
            continue
        observed.setdefault(scope, {}).update(seen)

    if skipped_remote:
        oks.append("E0 %d remote location(s) SKIPPED -- rerun with --remote to include "
                   "the headend shadow stores (SEC-022) and the region secrets dir "
                   "(SEC-020)" % skipped_remote)

    # bound 1: every expected artifact present at its expected role
    absent = 0
    deferred = 0
    gated = 0
    undeclarable = []
    for r in rows:
        if not r.has_artifact():
            continue
        scope = (r.host_role, r.site_key)
        if scope not in by_scope and (r.host_role, "-") not in by_scope:
            # NOT silently skipped: a row whose scope has no declared location can never
            # be checked, and the module's contract is that nothing passes quietly.
            undeclarable.append("%s '%s' (%s/%s)"
                                % (r.id, r.filename, r.host_role, r.site_key))
            continue
        pools = [observed.get(scope, {})]
        if r.site_key != "-":
            pools.append(observed.get((r.host_role, "-"), {}))   # shared location
        hit = got = None
        for pool in pools:
            hit = next((f for f in pool if fn_matches(r.filename, f)), None)
            if hit:
                got = pool[hit]
                break
        if hit:
            want = derived_mode(r.filename)
            if r.custody in IN_FOLDER and got != want:
                fails.append("E2 MODE: %s '%s' at %s/%s is %s, want %s"
                             % (r.id, r.filename, r.host_role, r.site_key, got, want))
            elif r.principal != "-" and want == "600" and int(got, 8) & 0o007:
                # World-readable SECRET, at ANY custody (a `.pub` half derives 644 and is
                # public by design, so it is excluded). The exact-mode check above only
                # covers `consolidated` rows, which left 43 of 77 rows -- including the
                # SEC-020 region dir and the SEC-022 shadow stores, the two that motivated
                # remote probing -- with no mode check at all.
                fails.append("E2 WORLD-READABLE: %s '%s' at %s/%s is mode %s"
                             % (r.id, r.filename, r.host_role, r.site_key, got))
            continue
        if scope in incomplete:
            gated += 1
            continue
        if stages.get(r.mint_stage, True) is False:
            # NOT-YET-EXPECTED per creds-manifests/stages-reached, which is a derived
            # projection of CURRENT-STATE coupled to it by repo-lint L12. The script still
            # carries no status claim of its own (GA-R1).
            deferred += 1
            continue
        absent += 1
        fails.append("E1 EXPECTED-BUT-ABSENT: %s '%s' expected at %s, not found "
                     "(mint-stage %s%s)"
                     % (r.id, r.filename, "%s/%s" % (r.host_role, r.site_key),
                        r.mint_stage,
                        "" if r.sec_ref == "-" else ", " + r.sec_ref))

    # bound 2: nothing undeclared at a declared location
    declared = {}
    for r in rows:
        if r.has_artifact():
            declared.setdefault((r.host_role, r.site_key), set()).add(r.filename)
    undeclared = 0
    for scope in sorted(observed):
        for name in sorted(observed[scope]):
            names = set(declared.get(scope, set()))
            if scope[1] == "-":
                # a shared location is satisfied by ANY site's row at that role
                for (role2, _s2), decl in declared.items():
                    if role2 == scope[0]:
                        names |= decl
            if not any(fn_matches(d, name) for d in names):
                undeclared += 1
                fails.append("E3 UNDECLARED: '%s' present at %s/%s but in NO matrix row "
                             "for that site -- an undeclared secret is the exact class "
                             "creds-audit cannot see" % (name, scope[0], scope[1]))

    if deferred:
        oks.append("E1 %d expected artifact(s) deferred as not-yet-minted per "
                   "creds-manifests/stages-reached"
                   % deferred)
    if gated:
        oks.append("E1 %d expected artifact(s) NOT JUDGED -- their role (%s) has at least "
                   "one location that could not be probed, so absence cannot be asserted "
                   "over it" % (gated, ", ".join("%s/%s" % c
                                                 for c in sorted(incomplete))))
    if undeclarable:
        fails.append("E4 UNCHECKABLE: %d row(s) have no declared location for their "
                     "(role, site) and can never be verified -- add a location row or "
                     "correct the matrix: %s" % (len(undeclarable), "; ".join(undeclarable)))
    if not (absent or undeclared):
        oks.append("E1/E3 existence: every expected artifact present and nothing "
                   "undeclared, across %d fully-probed role(s)"
                   % len([c for c in observed if c not in incomplete]))
    return observed



# --------------------------------------------------------------------------- tier 3
# VALIDITY. Motivated by a measured gap: `creds-audit` PARSES declared provenance and
# NEVER verifies it, so SEC-020's stale-trap warning -- "any future rotation MUST update
# both copies in one operation or the region copy becomes a stale trap" -- has been
# unchecked prose since the day it was written. The sha256 equality it depends on was a
# ONE-TIME manual act on 2026-07-25 and nothing has re-verified it since.
#
# DIGESTS ONLY, NEVER CONTENT. `sha256sum` is invoked as an external command on BOTH sides
# so that only a 64-hex digest ever enters this process -- the plan's hard constraint is
# "compare sha256sum digests over ssh -- never transfer content", and reading bytes into
# Python to hash them locally would satisfy the letter and not the spirit.
SHA_RE = re.compile(r"^([0-9a-f]{64})\s")


def digest_local(path):
    try:
        p = subprocess.run(["sha256sum", path], capture_output=True, text=True, timeout=60)
    except (OSError, subprocess.SubprocessError):
        return None
    m = SHA_RE.match(p.stdout)
    return m.group(1) if m else None


def digest_remote(target, path, privileged=False):
    cmd = "sha256sum %s" % path                      # path validated by SAFE_PATH_RE
    if privileged:
        cmd = "sudo -n sh -c " + shlex.quote(cmd)
    try:
        p = subprocess.run(["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=10",
                            target, cmd], capture_output=True, text=True, timeout=60)
    except (OSError, subprocess.SubprocessError):
        return None
    m = SHA_RE.match(p.stdout)
    return m.group(1) if m else None


# Declared-state vocabulary in docs/security-ledger.md. Keyword matching on prose is
# FRAGILE and deliberately conservative: a small explicit set, and every hit REPORTS the
# matched word so a human can see the basis rather than trusting the parse.
REVOKED_WORDS = ("BURNED", "REVOKED", "COMPROMISED")

# A declared-dead credential whose REMEDIATION the operator has explicitly deferred is a
# different state from one nobody has ruled on, and V2 must not report them identically:
# SEC-006 has carried "**DEFERRED by operator ruling 2026-07-13: revoke at completion of
# this deployment.**" since before this checker existed, so failing it every run reports the
# operator's own standing decision as a defect and trains readers to skip the finding.
#
# THE MATCH IS DELIBERATELY NARROW, and the narrowness is the whole safety property. It
# demands BOTH the emphasis-cased state word AND an explicit attribution to an operator
# ruling -- the GA-R5 bar. A looser pattern (any "defer") would let a credential be silenced
# by a sentence merely *discussing* deferral, which converts this check into the false green
# the register exists to prevent. Ruled-deferred is still reported, as an ACKNOWLEDGED
# DEFERRAL naming the ruling; it is never dropped and never silent.
DEFERRAL_RE = re.compile(r"DEFERRED\b[^|]{0,120}?operator ruling")


def load_sec_states(path):
    """{SEC-NNN: (state-word, snippet, deferral)} for rows the ledger declares dead.

    `deferral` is the matched ruled-deferral phrase, or None when no operator ruling has
    deferred the remediation.
    """
    out = {}
    if not os.path.exists(path):
        return out
    for line in open(path):
        m = re.match(r"\|\s*(SEC-\d+)\s*\|", line)
        if not m:
            continue
        for w in REVOKED_WORDS:
            if w in line:
                i = line.index(w)
                d = DEFERRAL_RE.search(line)
                out[m.group(1)] = (w, line[max(0, i - 60):i + 80].strip(),
                                   d.group(0).strip() if d else None)
                break
    return out


def v2_declared_state(rows, sec_states, oks, fails):
    """A credential the ledger declares DEAD must not sit in the register as healthy.

    Nothing reconciled these two surfaces before: SEC-006 has recorded the NetBox upstream
    token as BURNED -- "must be revoked + reissued, not merely rotated" -- since 2026-07-12,
    while the matrix carried it as an ordinary `consolidated` row. This is the durable half
    of validity: a liveness probe is a snapshot that can be stale a minute later, but a
    declared revocation does not un-declare itself."""
    bad = 0
    deferred = 0
    for r in rows:
        if r.sec_ref == "-" or r.sec_ref not in sec_states:
            continue
        word, snippet, deferral = sec_states[r.sec_ref]
        if r.custody not in IN_FOLDER and r.custody != "source-of-record":
            continue
        if deferral:
            # Reported, not dropped. The credential is still live and still exposed -- the
            # ruling defers the REMEDIATION, it does not change the state -- so the notice
            # says so rather than reading as an all-clear.
            deferred += 1
            oks.append("V2 ACKNOWLEDGED DEFERRAL: id '%s' (%s) is custody=%s and %s declares "
                       "it %s, but remediation is DEFERRED BY OPERATOR RULING -- the credential "
                       "is still live and still exposed until that ruling is discharged. "
                       "Ruling text: '%s'"
                       % (r.id, r.filename, r.custody, r.sec_ref, word, deferral))
            continue
        bad += 1
        fails.append("V2 DECLARED-DEAD BUT REGISTERED HEALTHY: id '%s' (%s) is custody=%s, "
                     "but %s declares it %s -- a credential that must be revoked and "
                     "reissued cannot also be a good consolidated copy. Ledger text: ...%s..."
                     % (r.id, r.filename, r.custody, r.sec_ref, word, snippet[:90]))
    if not bad:
        oks.append("V2 declared state: no credential is registered healthy while its SEC row "
                   "declares it %s" % "/".join(w.lower() for w in REVOKED_WORDS))


def tier3_validity(rows, locations, use_remote, use_privileged, observed, oks, fails):
    """Copies of one identity must be byte-identical, and the authority must be knowable."""
    # where each (role, site) lives, so a row can be resolved to a concrete path
    loc_by_scope = {}
    for role, site, target, pat in locations:
        loc_by_scope.setdefault((role, site), []).append((target, pat))

    # WHAT IS COMPARABLE. An identity's rows are not all byte-copies of one another:
    # `maas-region-admin` covers a password, an API key and a CLI profile -- three distinct
    # artifacts -- and `netbox-sandbox-token` is a raw token on the VM but an assembled ENV
    # FILE on the jumphost, same credential, different serialization. Comparing either pair
    # is a false positive, and the first draft of this check produced exactly those two.
    # Only two groupings are byte-comparable:
    #   (a) rows sharing (id, filename) at DIFFERENT scopes -- the same artifact, copied;
    #   (b) an id's `source-of-record` row against its `verbatim-copy` rows -- a declared
    #       byte-identical copy under a different name (SEC-020's admin.pass shape).
    # Anything else is skipped with a reason. A pair we cannot justify comparing is not a
    # pair we may report drift on.
    groups = {}
    for r in rows:
        if r.has_artifact():
            groups.setdefault((r.id, r.filename), []).append(r)
    comparable = [(k, v) for k, v in groups.items() if len(v) > 1]
    for ident in sorted({r.id for r in rows}):
        src = [r for r in rows if r.id == ident and r.custody == "source-of-record"
               and r.has_artifact()]
        cps = [r for r in rows if r.id == ident and r.custody == "verbatim-copy"
               and r.has_artifact()]
        if len(src) == 1 and cps:
            comparable.append((("%s (declared verbatim copies)" % ident, ""), src + cps))

    compared = matched = skipped = 0
    verified = []
    for (gid, _fn), copies in sorted(comparable, key=lambda t: str(t[0])):
        digests = {}
        for c in copies:
            scope = (c.host_role, c.site_key)
            paths = loc_by_scope.get(scope) or loc_by_scope.get((c.host_role, "-")) or []
            got = None
            for target, pat in paths:
                base = os.path.dirname(pat)
                if glob.has_magic(os.path.expanduser(base)) or "<" in c.filename:
                    continue                 # instance-templated: no single resolvable path
                full = base + "/" + c.filename
                if target == "local":
                    if os.path.isfile(os.path.expanduser(full)):
                        got = digest_local(os.path.expanduser(full))
                elif use_remote:
                    got = digest_remote(target, full)
                    if got is None and use_privileged:
                        got = digest_remote(target, full, privileged=True)
                if got:
                    break
            if got:
                digests["%s/%s:%s" % (c.host_role, c.site_key, c.filename)] = got
        if len(digests) < 2:
            skipped += 1
            continue
        compared += 1
        if len(set(digests.values())) == 1:
            matched += 1
            verified.append(str(gid))
            continue
        fails.append("V1 PROVENANCE DRIFT: '%s' -- %d copies that MUST be byte-identical "
                     "have %d different digests (%s). A rotation updated one copy and not "
                     "the other; the un-updated copy is a STALE TRAP (SEC-020)."
                     % (gid, len(digests), len(set(digests.values())),
                        ", ".join(sorted(digests))))
    if skipped:
        oks.append("V1 %d multi-copy identity(ies) SKIPPED -- fewer than two copies could "
                   "be digested (unresolvable path, instance template, or a location not "
                   "probed). A digest that could not be taken is never a match."
                   % skipped)
    if compared and matched == compared:
        # Name what was verified. "4 compared" is not an audit trail -- a reader must be
        # able to see that the SEC-020 pair is among them and not silently skipped.
        oks.append("V1 provenance: all copies byte-identical for %d group(s): %s"
                   % (compared, "; ".join(sorted(verified))))
    elif not compared:
        oks.append("V1 provenance: no identity had two digestible copies -- NOTHING was "
                   "verified here; this is a skip, not a pass")


def main():
    ap = argparse.ArgumentParser(description="D-137 credential-matrix checker")
    here = os.path.dirname(os.path.abspath(__file__))
    repo = os.path.dirname(here)
    ap.add_argument("--matrix", default=os.path.join(repo, "creds-matrix.tsv"))
    ap.add_argument("--manifest-dir", default=os.path.join(repo, "creds-manifests"))
    ap.add_argument("--notes", default=os.path.join(repo, "creds-matrix-notes.md"))
    ap.add_argument("--repo", default=repo)
    ap.add_argument("--dc", default=None, help="narrow per-DC checks to one region-qualified DC")
    ap.add_argument("--all", action="store_true", help="run every implemented tier")
    ap.add_argument("--locations",
                    default=os.path.join(repo, "creds-manifests", "vm-secret-locations"))
    ap.add_argument("--host-identity",
                    default=os.path.join(repo, "creds-manifests", "host-identity"),
                    help="role -> hostname binding for `local` locations (D-137 tier 2)")
    ap.add_argument("--this-host", default=None,
                    help="override the detected hostname (harness fixtures only)")
    # V2 reads the SEC register. It was derived from --repo, which the harness must point at
    # the REAL repo for S4 mint-ref resolution -- so V2 was the one check with no fixture
    # path and therefore shipped with ZERO harness cases. This override is what makes it
    # testable; the default is unchanged.
    ap.add_argument("--ledger",
                    default=os.path.join(repo, "docs", "security-ledger.md"),
                    help="SEC register read by tier-3 V2 (default: <repo>/docs/security-ledger.md)")
    ap.add_argument("--tier2", action="store_true",
                    help="run tier 2 (EXISTENCE) -- local locations only unless --remote")
    ap.add_argument("--remote", action="store_true",
                    help="include ssh locations, bounded ABSOLUTELY by vm-secret-locations "
                         "(ruling 3). Metadata only; no file content is ever read.")
    ap.add_argument("--tier3", action="store_true",
                    help="run tier 3 (VALIDITY): cross-copy sha256 provenance. Implies "
                         "--tier2, which supplies the probe results.")
    ap.add_argument("--privileged", action="store_true",
                    help="allow a `sudo -n` ESCALATION RETRY on a remote location the "
                         "unprivileged probe could not open (the root-owned region and "
                         "netbox secret dirs). Retried only where needed; metadata only.")
    ap.add_argument("--stages",
                    default=os.path.join(repo, "creds-manifests", "stages-reached"),
                    help="declared reach of each mint stage. A derived projection of "
                         "CURRENT-STATE (repo-lint L12 couples them); the script carries "
                         "no status claim of its own (GA-R1).")
    ap.add_argument("--render", action="store_true",
                    help="print manifests rendered from the matrix (does not write)")
    args = ap.parse_args()

    if args.dc and not SITE_KEY_RE.match(args.dc):
        sys.stderr.write("ERROR: --dc must be region-qualified (vr1-dc0 | vr1-dc1)\n")
        return 2

    oks, fails = [], []
    rows = load_matrix(args.matrix, fails)
    notes = load_notes(args.notes)

    if args.render:
        for site in sorted({r.site_key for r in rows if r.site_key != "-"}):
            print("# %s.manifest -- GENERATED from creds-matrix.tsv (D-137 ruling 2)." % site)
            print("# Do not hand-edit: regenerate with scripts/creds-matrix.py --render.")
            for fname, mode, source, r in render_rows(rows, site):
                if r.notes_ref != "-":
                    print("# [%s] see creds-matrix-notes.md ## %s" % (r.id, r.notes_ref))
                print("%-32s %s  %s" % (fname, mode, source))
            print()
        return 0

    print("=== creds-matrix: tier 1 (STATIC) ===")
    s1_schema(rows, oks, fails)
    s2_manifest_coverage(rows, args.manifest_dir, oks, fails)
    s3_render_drift(rows, args.manifest_dir, notes, oks, fails)
    s4_mint_ref(rows, args.repo, oks, fails)
    s5_per_dc_symmetry(rows, args.dc, oks, fails)
    s6_principal_invariant(rows, oks, fails)
    s7_notes_ref(rows, notes, args.notes, oks, fails)

    if args.tier2 or args.tier3 or args.all:
        print("=== creds-matrix: tier 2 (EXISTENCE) ===")
        locations = load_locations(args.locations, args.repo, fails)
        if locations is None:
            fails.append("tier 2: no declared-location list at %s -- ruling 3 requires one "
                         "before any discovery may run" % args.locations)
        elif not locations:
            fails.append("tier 2: the declared-location list at %s is EMPTY -- ruling 3's "
                         "bound is absolute, so an unpopulated list REFUSES like a missing "
                         "one rather than affirming existence over zero locations"
                         % args.locations)
        else:
            stages = load_stages(args.stages, rows, fails)
            if stages is None:
                fails.append("tier 2: no stage declaration at %s -- without it the checker "
                             "cannot tell a MISSING credential from a NOT-YET-MINTED one, "
                             "and it will not guess" % args.stages)
                stages = {}
            host_identity = load_host_identity(args.host_identity, locations, fails)
            if host_identity is None:
                fails.append("tier 2: no host-identity declaration at %s -- without it the "
                             "checker cannot tell whether a `local` location belongs to the "
                             "machine it is running on, and it will not guess"
                             % args.host_identity)
                host_identity = {}
            this_host = args.this_host or socket.gethostname()
            print("  (host: %s)" % this_host)
            observed = tier2_existence(rows, locations, args.remote, args.privileged,
                                       stages, host_identity, this_host, oks, fails)
            if args.tier3 or args.all:
                print("=== creds-matrix: tier 3 (VALIDITY) ===")
                v2_declared_state(rows, load_sec_states(args.ledger), oks, fails)
                tier3_validity(rows, locations, args.remote, args.privileged,
                               observed, oks, fails)
    else:
        # SELF-SKIP WITH AN EXPLICIT [ok]: a tier that did not run must be visible.
        oks.append("tier 2 (EXISTENCE) NOT RUN -- pass --tier2 (add --remote for the "
                   "headend shadow stores and the region secrets dir)")

    if not (args.tier3 or args.all):
        oks.append("tier 3 (VALIDITY) NOT RUN -- pass --tier3 (with --tier2) to compare "
                   "cross-copy sha256 provenance")

    for o in oks:
        print("  [ok]   %s" % o)
    for f in fails:
        print("  [FAIL] %s" % f)
    print("\n%s: creds-matrix tier 1 -- %d row(s), %d check group(s) clean, %d finding(s)"
          % ("PASS" if not fails else "FAIL", len(rows), len(oks), len(fails)))
    return 1 if fails else 0


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