#!/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
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.
# 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 == "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
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_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, 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)."""
# 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:
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 == "consolidated" 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 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, "%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 (--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("%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]))
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)
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:
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())
