admin/felhom.eu
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

R-315: wire-contract gate checks mirrored roots field-by-field and prints each root's check kind

Browse Source 
The escrow/retained and escrow-ACK wires are now compared path-by-path against the receiver's named
mirror type, so the measured mutation (agent renames superseded_at, the old name still a local-API map
key) convicts. Decoy pair in test_gate_decoys.py, seen failing with the mirror removed.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0159rPz1ZhFKsS53msqPYxtS
This commit is contained in:
authentik Default Admin
2026-10-05 21:31:35 +02:00
parent daeed175ca
commit 900e6d86e2
2 changed files with 84 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/test_gate_decoys.py
+40 -1
View File
@@ -55,7 +55,9 @@ COVERS = {
"decoy-coverage": "a gate registered in a runner with no decoy and no exemption (its red-proof)",
"wire-contract": ("R-555: an emitted tag whose name the receiver carries ONLY in a // and a /* */ comment "
"must convict (it passed for months as `language` did); the genuine article — the same "
"name in a struct tag beside a `//` inside a string literal — must pass"),
"name in a struct tag beside a `//` inside a string literal — must pass; and R-315: the "
"agent RENAMING its mirror's `superseded_at` tag (the old name still occurs as a local-API "
"map key) must convict on the field-by-field mirror check"),
"stands": ("R-819: a stand citing a register id that exists in NEITHER register, and a stand "
"marked 'walked' whose only source is a register row (a green dot from a label); "
"plus the genuine article — a stand citing only a CLOSED row, which must PASS"),
@@ -525,6 +527,43 @@ for _mode, _want in (("comment", 10), ("genuine", 0)):
print(" ok %-20s %s" % ("wire-contract/" + _mode,
"decoy rejected" if _want else "genuine accepted"))
# ── wire-contract mirror check (R-315) ───────────────────────────────────────────────────────────
#
# The measured R-315 mutation: the AGENT renames its mirror's `superseded_at` tag. The name check passed
# it, because `superseded_at` also occurs in the agent as a local-API map key. The rename is applied to
# the parsed mirror body only (no copy of the agent tree), so the token set still holds the old name —
# exactly the shape that fooled the gate. The genuine article (no rename) must pass.
_WCM = r"""
import os, sys
sys.path.insert(0, "scripts")
import wire_contract_gate as g
mode = sys.argv[1]
orig, agent = g.build_index, os.path.abspath(g.REPOS["agent"])
def patched(root):
by_dir, by_name = orig(root)
if mode == "renamed" and os.path.abspath(root) == agent:
k = ("internal/hub", "RetainedEscrowPackage")
assert 'json:"superseded_at"' in by_dir[k], "mutation target missing"
by_dir[k] = by_dir[k].replace('json:"superseded_at"', 'json:"superseded_at_RENAMED"')
by_name["RetainedEscrowPackage"] = [(d, by_dir[k] if d == k[0] else b)
for d, b in by_name["RetainedEscrowPackage"]]
return by_dir, by_name
g.build_index = patched
rc, conv = g.run(quiet=True)
hit = any(d == "packages.superseded_at" for _, _, m in conv for _, d in m)
print("rc=%d convicted=%s" % (rc, hit))
sys.exit(0 if rc == 0 else (10 if hit else 11))
"""
for _mode, _want in (("renamed", 10), ("genuine", 0)):
ran += 1
_p = subprocess.run([sys.executable, "-c", _WCM, _mode], cwd=ROOT, capture_output=True, text=True)
if _p.returncode != _want:
fails.append("wire-contract-mirror/%s: rc=%d, want %d (10 = the renamed mirror tag convicted, 0 = passed)\n%s"
% (_mode, _p.returncode, _want, (_p.stdout + _p.stderr)[-500:]))
else:
print(" ok %-20s %s" % ("wire-mirror/" + _mode,
"decoy rejected" if _want else "genuine accepted"))
# ── stands (R-819) ───────────────────────────────────────────────────────────────────────────────
#
# check_stands.py was red and in no runner. Its decoys are stand files written as a session would write
scripts/wire_contract_gate.py
+44
View File
@@ -47,6 +47,9 @@ gate must publish its holes.
This makes the gate conservative: it under-reports and does not over-report.
2. IT PROVES REACHABILITY OF A NAME, NOT THAT ANYTHING ACTS ON THE VALUE. A tag mentioned once in
a struct nobody consults passes. It answers "can this be decoded at all", not "is it used".
R-315: a root listed in MIRRORS gets a FIELD-BY-FIELD check against the receiver's named mirror
type instead (a renamed receiver tag convicts even when the old name occurs elsewhere); every
run prints which check each root got. The other roots are still name-checked only.
3. ROOTS ARE DECLARED, NOT DISCOVERED. Only the wires in ROOTS are covered. The hub's
desired-state is served as raw stored JSON (`host.DesiredJSON`) with no typed emitter to walk,
and the agent's local API has no single root type — NEITHER IS COVERED.
@@ -89,6 +92,19 @@ ROOTS = [
"hub", "internal/api", "RetainedEscrowResponse", "agent"),
]
# R-315: a root whose receiver decodes it into ONE NAMED MIRROR TYPE gets the stronger check —
# field-by-field: every emitted dotted path must be a json path of the mirror type (generic names
# included, since a path comparison is exact). The name-reachability check below could not see a
# renamed receiver tag when the old name occurred anywhere else in the receiving repo (measured: the
# agent's `superseded_at` renamed still passed, because the local API used the same string as a map
# key). Roots NOT listed here keep the weaker name check, and the run prints which check each root
# got — a green on a name-checked root is not decodability. Keyed by root label:
# (receiver package dir, receiver type).
MIRRORS = {
"hub -> agent (GET /hosts/<id>/escrow/retained)": ("internal/hub", "RetainedEscrowResponse"),
"hub -> controller (report ACK, `escrow` object)": ("internal/report", "EscrowStatus"),
}
# Tag names whose literal string carries no information in a repo-wide search. NOT CHECKED.
# Listed rather than silently skipped: each one is a hole.
GENERIC = {
@@ -574,6 +590,7 @@ def run(root_override=None, quiet=False):
# one pass per receiving repo, not one subprocess per tag
rtokens = {k: receiver_tokens(v) for k, v in repos.items()}
convictions = []
kinds = [] # R-315: (root label, which kind of check it got) — printed on every run
checked = skipped = 0
for label, emitter, pkgdir, rootname, receiver in ROOTS:
@@ -588,6 +605,31 @@ def run(root_override=None, quiet=False):
seen_tags.setdefault(tag, dotted)
opaque_roots = [p for (lbl, p) in OPAQUE_BELOW if lbl == label]
missing = []
if label in MIRRORS:
mdir, mtype = MIRRORS[label]
rby_dir, rby_name = indexes[receiver]
if (mdir, mtype) not in rby_dir:
die("wire-contract gate INCONCLUSIVE: declared mirror %s.%s not found in %s/%s\n"
" A mirror that cannot be resolved is not a pass — fix MIRRORS or the type."
% (receiver, mtype, receiver, mdir))
mirror_paths = {d for _, d in walk(rby_dir, rby_name, mdir, mtype)}
emitted = sorted({d for _, d in tags})
n_root = 0
for dotted in emitted:
if (label, dotted) in ALLOWLIST or any(
dotted.startswith(o + ".") and dotted.count(".") > o.count(".") + 1
for o in opaque_roots):
skipped += 1
continue
checked += 1
n_root += 1
if dotted not in mirror_paths:
missing.append((dotted.split(".")[-1], dotted))
kinds.append((label, "FIELD-BY-FIELD against %s %s.%s (%d path(s))" % (receiver, mdir, mtype, n_root)))
if missing:
convictions.append((label, receiver, missing))
continue
kinds.append((label, "name-reachability only (a tag found ANYWHERE in %s passes)" % receiver))
for tag, dotted in sorted(seen_tags.items()):
if tag in GENERIC:
skipped += 1
@@ -609,6 +651,8 @@ def run(root_override=None, quiet=False):
if not quiet:
print("wire-contract gate — %d tag(s) checked across %d declared wire(s); "
"%d skipped (generic / opaque / allowlisted)" % (checked, len(ROOTS), skipped))
for label, kind in kinds:
print(" %-52s %s" % (label, kind))
if convictions:
if not quiet:
for label, receiver, missing in convictions: