#!/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 <-- IMPLEMENTED HERE
2. EXISTENCE local + remote reads; runs as a preflight gate <-- NOT BUILT YET
3. VALIDITY provenance digests + behavioral probes <-- NOT BUILT YET
Tiers 2 and 3 SELF-SKIP with an explicit [ok] naming them as unbuilt. They never pass
silently -- a silent pass is the false-green this whole design exists to prevent.
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
Exit: 0 clean, 1 findings, 2 usage/IO error. ASCII-only output.
"""
import sys, os, re, argparse
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 pending tier 2 -- they need the "
"declared path from creds-manifests/vm-secret-locations (ruling 3), "
"which is not built yet" % 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))
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("--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.all:
# SELF-SKIP WITH AN EXPLICIT [ok]: an unbuilt tier must be visible, never silent.
oks.append("tier 2 (EXISTENCE) NOT BUILT -- needs creds-manifests/vm-secret-locations "
"and creds-audit --remote; SEC-022/-023 and SEC-021's on-disk half cannot "
"be reproduced until it exists")
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())
