R-426: decoy suite for the published gate, against a fake Gitea
scripts/test_gate_decoys.py (new; COVERS "published"): an http.server on 127.0.0.1 stands in for Gitea through the gate's existing GITEA_BASE seam, proxies stripped, so no case reaches the real registry. Facts convicted: a tag whose package 404s, a tag tree without the configs, a package one patch past the newest tag (never tagged), a patch-gap orphan, a missing package that lexical sorting would drop out of the retention window. Inconclusive, never a pass: tags api 500, a non-JSON 200, Gitea unreachable. Passed: a clean registry, a non-semver tag, a version older than the retention window (BY DESIGN). 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:
@@ -0,0 +1,236 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""test_gate_decoys.py — can this repo's gates be fooled by a LABEL? (R-421, R-426)
|
||||
|
||||
The same instrument as `felhom.eu/scripts/test_gate_decoys.py`: a decoy is the LABEL without the
|
||||
FACT, and a gate that passes on the label alone — or refuses the genuine article — is a live hole.
|
||||
Every gate is asserted in BOTH directions: the decoy must be convicted, the genuine article passed.
|
||||
|
||||
Covered here (the `COVERS` literal is AST-read by `felhom.eu/scripts/decoy_coverage_gate.py`, which
|
||||
never imports this file):
|
||||
|
||||
published check-published-versions.py, against a FAKE Gitea (see below).
|
||||
release-complete check-release-complete.py, in a scratch clone whose `origin` is a scratch bare
|
||||
repository, against the same fake Gitea.
|
||||
reuse-refs, instructions, observations
|
||||
the three SHARED felhom.eu scripts, run against a scratch clone of THIS repo —
|
||||
so the decoy is planted in the agent's own REUSE.md / CLAUDE.md / REPORT.md and
|
||||
coverage is per input, not per script.
|
||||
|
||||
NEVER THE REAL GITEA. Both network gates read `GITEA_BASE` from the environment (CI already sets it
|
||||
to the in-cluster URL); here it points at an `http.server` bound to 127.0.0.1 inside this process,
|
||||
and every proxy variable is removed from the child's environment so urllib cannot route around it.
|
||||
A test that asked the real registry would pass or fail on whatever was published that day — the
|
||||
constant-for-measurement shape — and would reach the network from a hook.
|
||||
|
||||
NEVER THE REAL TREE. Every planted file lives in a scratch directory: a workspace that holds a
|
||||
clone of this repo beside symlinks to the sibling clones the shared scripts reach across to.
|
||||
|
||||
Run from the repo root: python3 scripts/test_gate_decoys.py
|
||||
Exit 0 every decoy judged correctly · 1 a decoy passed or a genuine article was refused.
|
||||
"""
|
||||
import http.server
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import socketserver
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
|
||||
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
PARENT = os.path.dirname(ROOT)
|
||||
|
||||
# ── WHAT THIS FILE COVERS ────────────────────────────────────────────────────────────────────────
|
||||
# Read by felhom.eu/scripts/decoy_coverage_gate.py, which AST-parses this literal. A gate named here
|
||||
# MUST have a decoy below that has been seen to fail.
|
||||
COVERS = {
|
||||
"published": "against a FAKE Gitea: a tag whose package 404s, a tag whose tree lacks the configs, a package one "
|
||||
"version past the newest tag (published, never tagged), a patch-gap orphan, a missing package that "
|
||||
"lexical sorting would drop out of the retention window (0.9.x vs 0.10.x); a tags api answering 500 "
|
||||
"or a non-JSON 200 is INCONCLUSIVE, never a pass - vs a clean registry, a non-semver tag and a version "
|
||||
"older than the retention window (not asserted, BY DESIGN) (R-426)",
|
||||
}
|
||||
|
||||
fails = []
|
||||
ran = 0
|
||||
|
||||
|
||||
def report(name, rc, out, expect_rc, must=()):
|
||||
global ran
|
||||
ran += 1
|
||||
missing = [m for m in must if m not in out]
|
||||
if rc == expect_rc and not missing:
|
||||
print(" ok %-62s rc=%d (expected %d)" % (name, rc, expect_rc))
|
||||
else:
|
||||
hole = expect_rc != 0 and rc == 0
|
||||
fails.append("%s: rc=%d expected %d%s; missing %s\n%s" % (
|
||||
name, rc, expect_rc, " - LIVE HOLE" if hole else "", missing, out[-900:]))
|
||||
|
||||
|
||||
# ── the fake Gitea ───────────────────────────────────────────────────────────────────────────────
|
||||
class Fake(object):
|
||||
"""What the fake registry serves. Reset per case."""
|
||||
|
||||
def reset(self):
|
||||
self.tags = [] # tag names, as the tags api lists them
|
||||
self.packages = set() # versions whose generic package downloads
|
||||
self.raw = set() # versions whose tag tree serves the probe config
|
||||
self.tags_status = 200
|
||||
self.tags_body = None # override bytes for the tags api
|
||||
self.pkg_status = None # override status for EVERY package request
|
||||
self.hits = []
|
||||
|
||||
|
||||
FAKE = Fake()
|
||||
FAKE.reset()
|
||||
PKG_PREFIX = "/api/packages/admin/generic/felhom-agent/"
|
||||
RAW_PREFIX = "/admin/felhom-agent/raw/tag/v"
|
||||
|
||||
|
||||
class Handler(http.server.BaseHTTPRequestHandler):
|
||||
def log_message(self, *a):
|
||||
pass
|
||||
|
||||
def _answer(self, status, body=b""):
|
||||
self.send_response(status)
|
||||
self.send_header("Content-Length", str(len(body)))
|
||||
self.end_headers()
|
||||
if self.command != "HEAD":
|
||||
self.wfile.write(body)
|
||||
|
||||
def do_GET(self):
|
||||
p = self.path
|
||||
FAKE.hits.append(p)
|
||||
if p.startswith("/api/v1/repos/admin/felhom-agent/tags"):
|
||||
body = FAKE.tags_body if FAKE.tags_body is not None else \
|
||||
json.dumps([{"name": t} for t in FAKE.tags]).encode()
|
||||
return self._answer(FAKE.tags_status, body)
|
||||
if p.startswith(PKG_PREFIX):
|
||||
if FAKE.pkg_status is not None:
|
||||
return self._answer(FAKE.pkg_status)
|
||||
v = p[len(PKG_PREFIX):].split("/", 1)[0]
|
||||
return self._answer(200, b"ELF") if v in FAKE.packages else self._answer(404)
|
||||
if p.startswith(RAW_PREFIX):
|
||||
v = p[len(RAW_PREFIX):].split("/", 1)[0]
|
||||
ok = v in FAKE.raw and p.endswith("/configs/felhom-agent.service")
|
||||
return self._answer(200, b"[Unit]\n") if ok else self._answer(404)
|
||||
return self._answer(404)
|
||||
|
||||
do_HEAD = do_GET
|
||||
|
||||
|
||||
class Server(socketserver.ThreadingMixIn, http.server.HTTPServer):
|
||||
daemon_threads = True
|
||||
|
||||
|
||||
def child_env(base):
|
||||
env = {k: v for k, v in os.environ.items() if "proxy" not in k.lower()}
|
||||
env["GITEA_BASE"] = base
|
||||
env["NO_PROXY"] = env["no_proxy"] = "127.0.0.1,localhost"
|
||||
return env
|
||||
|
||||
|
||||
def run(argv, cwd, env=None):
|
||||
# input="" — a child must never inherit (and block on) this process's stdin
|
||||
p = subprocess.run(argv, cwd=cwd, env=env, capture_output=True, text=True, input="")
|
||||
return p.returncode, p.stdout + p.stderr
|
||||
|
||||
|
||||
def sh(argv, cwd):
|
||||
rc, out = run(argv, cwd)
|
||||
if rc != 0:
|
||||
raise SystemExit("setup command failed (%s): %s" % (" ".join(argv), out))
|
||||
return out.strip()
|
||||
|
||||
|
||||
# ── published ────────────────────────────────────────────────────────────────────────────────────
|
||||
def published_cases(base):
|
||||
gate = os.path.join(ROOT, "scripts", "check-published-versions.py")
|
||||
keep = json.load(io.open(os.path.join(ROOT, "scripts", "retention-policy.json"),
|
||||
encoding="utf-8"))["generic_versions_kept"]
|
||||
TAGS = ["v0.150.%d" % i for i in range(3)] # inside any retention window >= 3
|
||||
VERS = [t[1:] for t in TAGS]
|
||||
|
||||
def case(name, setup, expect_rc, must=()):
|
||||
FAKE.reset()
|
||||
FAKE.tags = list(TAGS)
|
||||
FAKE.packages = set(VERS)
|
||||
FAKE.raw = set(VERS)
|
||||
setup()
|
||||
rc, out = run([sys.executable, gate], ROOT, child_env(base))
|
||||
if not FAKE.hits:
|
||||
fails.append("published/%s: the gate never asked the fake Gitea - the seam is not wired" % name)
|
||||
report("published: " + name, rc, out, expect_rc, must)
|
||||
|
||||
case("GENUINE: every tag downloadable and serving its configs", lambda: None, 0,
|
||||
("ALL RELEASED VERSIONS INSTALLABLE",))
|
||||
case("GENUINE: a non-semver tag is not a release", lambda: FAKE.tags.append("v0.150.2-rc1"), 0,
|
||||
("ALL RELEASED VERSIONS INSTALLABLE",))
|
||||
case("FACT: a tag whose package 404s", lambda: FAKE.packages.discard("0.150.1"), 1,
|
||||
("FAIL v0.150.1", "binary NOT downloadable"))
|
||||
case("FACT: a tag whose tree does not serve the configs", lambda: FAKE.raw.discard("0.150.2"), 1,
|
||||
("FAIL v0.150.2", "does not serve"))
|
||||
case("FACT: published one patch past the newest tag, never tagged",
|
||||
lambda: FAKE.packages.add("0.150.3"), 1, ("PUBLISHED VERSION(S) WITH NO TAG", "v0.150.3"))
|
||||
case("FACT: published in a patch GAP between two tags",
|
||||
lambda: (FAKE.tags.remove("v0.150.1"),), 1, ("v0.150.1 is downloadable", "has no git tag"))
|
||||
|
||||
def lexical():
|
||||
# keep+1 tags: 0.9.0 and 0.10.0..0.10.<keep-1>. By SEMVER the oldest is 0.9.0 (dropped); by
|
||||
# STRING sort "0.10.0" is the smallest and would be the one dropped - so its missing package
|
||||
# is convicted only if the window is cut by semver.
|
||||
FAKE.tags = ["v0.9.0"] + ["v0.10.%d" % i for i in range(keep)]
|
||||
FAKE.packages = set(t[1:] for t in FAKE.tags) - {"0.10.0"}
|
||||
FAKE.raw = set(t[1:] for t in FAKE.tags)
|
||||
case("FACT: a missing package lexical sorting would drop (0.10.0 vs 0.9.0)", lexical, 1,
|
||||
("FAIL v0.10.0",))
|
||||
|
||||
def retired():
|
||||
FAKE.tags = ["v0.9.0"] + ["v0.10.%d" % i for i in range(keep)]
|
||||
FAKE.packages = set(t[1:] for t in FAKE.tags) - {"0.9.0"}
|
||||
FAKE.raw = set(t[1:] for t in FAKE.tags)
|
||||
case("BY DESIGN: a version older than the retention window is not asserted", retired, 0,
|
||||
("NOT ASSERTED", "0.9.0"))
|
||||
|
||||
def five_hundred():
|
||||
FAKE.tags_status = 500
|
||||
case("INCONCLUSIVE: the tags api answers 500", five_hundred, 2, ("INCONCLUSIVE",))
|
||||
|
||||
def html():
|
||||
FAKE.tags_body = b"<html>sign in</html>"
|
||||
case("INCONCLUSIVE: the tags api answers a 200 that is not JSON", html, 2, ("INCONCLUSIVE",))
|
||||
|
||||
# an unreachable Gitea: a port nothing listens on
|
||||
s = Server(("127.0.0.1", 0), Handler)
|
||||
dead = "http://127.0.0.1:%d" % s.server_address[1]
|
||||
s.server_close()
|
||||
rc, out = run([sys.executable, gate], ROOT, child_env(dead))
|
||||
report("published: INCONCLUSIVE: Gitea unreachable", rc, out, 2, ("INCONCLUSIVE", "URLs tried"))
|
||||
|
||||
|
||||
def main():
|
||||
srv = Server(("127.0.0.1", 0), Handler)
|
||||
threading.Thread(target=srv.serve_forever, daemon=True).start()
|
||||
base = "http://127.0.0.1:%d" % srv.server_address[1]
|
||||
ws = tempfile.mkdtemp(prefix="agent-decoys-")
|
||||
print("agent gate decoys — fake Gitea at %s, scratch %s" % (base, ws))
|
||||
try:
|
||||
published_cases(base)
|
||||
finally:
|
||||
srv.shutdown()
|
||||
srv.server_close()
|
||||
shutil.rmtree(ws, ignore_errors=True)
|
||||
if fails:
|
||||
print()
|
||||
for f in fails:
|
||||
print("FAIL: %s" % f)
|
||||
return 1
|
||||
print("\nagent gate decoys OK — %d case(s), every label judged on its fact (R-421)" % ran)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user