diff --git a/scripts/test_gate_decoys.py b/scripts/test_gate_decoys.py new file mode 100644 index 0000000..63e5ad0 --- /dev/null +++ b/scripts/test_gate_decoys.py @@ -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.. 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"sign in" + 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())