#!/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, argparse, subprocess
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"}
HOST_ROLE = {"jumphost", "headend", "rack", "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"}
# 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.
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], "-")
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
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 == "consolidated"
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 != "consolidated" 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 == "consolidated"
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
with open(full) as fh:
n = sum(1 for _ in fh)
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
def load_locations(path, repo, fails):
"""The ruling-3 declared-location list. Returns [(role, ssh_target, path)] or None."""
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) != 3:
fails.append("vm-secret-locations:%d has %d field(s), want 3 "
"(<host-role> <ssh-target> <path>)" % (n, len(parts)))
continue
role, target, p = parts
# The repo has been renamed once (D-110); its name is never hardcoded.
out.append((role, target, p.replace("<repo>", repo)))
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):
return os.path.dirname(pattern) if pattern.endswith("/*") else None
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.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:
# See probe_local: "not a directory" is only ABSENT when the parent is
# traversable; otherwise the truth is that we could not look.
probe = ('p=$(dirname %s); '
'if [ ! -d %s ]; then '
' if [ -r "$p" ] && [ -x "$p" ]; then echo "__STATE__ absent"; '
' else echo "__STATE__ unreadable"; fi; '
'elif [ ! -r %s ] || [ ! -x %s ]; then echo "__STATE__ unreadable"; fi; '
% (d, d, d, d))
cmd = probe + 'for f in %s; do [ -f "$f" ] && stat -c "%%n %%a" "$f"; done' % pattern
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 = {}
for line in p.stdout.splitlines():
if line.startswith("__STATE__ "):
return (line.split(None, 1)[1].strip(), {})
bits = line.rsplit(None, 1)
if len(bits) == 2:
found[os.path.basename(bits[0])] = bits[1]
if p.returncode != 0 and not found:
return ("unreachable", {})
return ("ok", found)
def tier2_existence(rows, locations, use_remote, use_privileged, pending_stages,
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)."""
by_role = {}
for role, target, pat in locations:
by_role.setdefault(role, []).append((target, pat))
observed = {} # role -> {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, _t, _p in locations:
observed.setdefault(role, {})
skipped_remote = 0
for role, target, pat in locations:
if target != "local":
if not use_remote:
skipped_remote += 1
incomplete.add(role)
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:
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(role)
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(role, {}).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
for r in rows:
if not r.has_artifact() or r.host_role not in by_role:
continue
if r.host_role not in observed:
continue # that role was skipped/unreachable already
if r.filename in observed[r.host_role]:
want = derived_mode(r.filename)
got = observed[r.host_role][r.filename]
if r.custody == "consolidated" and got != want:
fails.append("E2 MODE: %s '%s' at %s is %s, want %s"
% (r.id, r.filename, r.host_role, got, want))
continue
if r.host_role in incomplete:
gated += 1
continue
if r.mint_stage in pending_stages:
# NOT-YET-EXPECTED, declared by the CALLER. The script carries no status
# claim of its own (GA-R1: status lives in CURRENT-STATE, nowhere else).
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, r.host_role, 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, set()).add(r.filename)
undeclared = 0
for role in sorted(observed):
for name in sorted(observed[role]):
if name not in declared.get(role, set()):
undeclared += 1
fails.append("E3 UNDECLARED: '%s' present at %s but in NO matrix row -- "
"an undeclared secret is the exact class creds-audit cannot "
"see" % (name, role))
if deferred:
oks.append("E1 %d expected artifact(s) deferred as not-yet-minted (--pending-stage)"
% 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(sorted(incomplete))))
if not (absent or undeclared):
oks.append("E1/E3 existence: every expected artifact present and nothing "
"undeclared, across %d fully-probed role(s)"
% len([r for r in observed if r not in incomplete]))
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("--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("--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("--pending-stage", action="append", default=[], metavar="STAGE",
help="a mint-stage this deployment has NOT reached; its expected "
"artifacts are deferred, not failed. Supply from CURRENT-STATE -- "
"this 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.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)
else:
tier2_existence(rows, locations, args.remote, args.privileged,
set(args.pending_stage), 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 args.all:
oks.append("tier 3 (VALIDITY) NOT BUILT -- needs cross-host sha256 provenance "
"comparison (never content transfer) and per-row behavioral probes")
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())
