felhom-agent/scripts/test_gate_decoys.py

#!/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)",
    "release-complete": "the newest `## vX.Y.Z` with no tag anywhere, a tag parked on an unrelated commit, a tag "
                        "with no package; a registry 500 is INCONCLUSIVE - vs the genuine release, an `## Unreleased` "
                        "heading above it, a newer version named only in prose or under `###`, a tag that only "
                        "origin has (the shallow-CI shape); a LOCAL-only tag passes BY DESIGN (CI's fresh clone "
                        "and the published gate's converse probe are what see it) (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"))


# ── release-complete ─────────────────────────────────────────────────────────────────────────────
def release_cases(base, ws):
    bare = os.path.join(ws, "origin.git")
    work = os.path.join(ws, "rc-work")
    sh(["git", "clone", "-q", "--bare", "--no-tags", "file://" + ROOT, bare], ws)
    sh(["git", "clone", "-q", "--no-tags", "file://" + bare, work], ws)
    sh(["git", "config", "user.email", "decoy@gate.invalid"], work)
    sh(["git", "config", "user.name", "decoy"], work)
    # the WORKING-TREE gate, so the file under test is the one being edited, not HEAD's
    shutil.copy(os.path.join(ROOT, "scripts", "check-release-complete.py"),
                os.path.join(work, "scripts", "check-release-complete.py"))
    sh(["git", "add", "scripts/check-release-complete.py"], work)
    sh(["git", "commit", "-q", "--allow-empty", "-m", "the gate under test"], work)
    base_sha = sh(["git", "rev-parse", "HEAD"], work)
    ch = os.path.join(work, "CHANGELOG.md")
    original = io.open(ch, encoding="utf-8").read()
    V = "9.9.9"

    def case(name, top, expect_rc, must=(), tag=None, origin_tag=False, packaged=True, pkg_status=None):
        FAKE.reset()
        if packaged:
            FAKE.packages = {V}
        FAKE.pkg_status = pkg_status
        try:
            io.open(ch, "w", encoding="utf-8").write(top + original)
            sh(["git", "commit", "-q", "-am", name], work)
            if tag == "head":
                sh(["git", "tag", "-a", "v" + V, "-m", "decoy", "HEAD"], work)
            elif tag == "unrelated":
                empty = sh(["git", "mktree"], work)   # stdin is "" — the empty tree, written to this repo
                orphan = sh(["git", "commit-tree", "-m", "unrelated", empty], work)
                sh(["git", "tag", "-a", "v" + V, "-m", "decoy", orphan], work)
            if origin_tag:
                sh(["git", "push", "-q", "origin", "HEAD:refs/tags/v" + V], work)
            rc, out = run([sys.executable, os.path.join(work, "scripts", "check-release-complete.py")],
                          work, child_env(base))
            report("release-complete: " + name, rc, out, expect_rc, must)
        finally:
            run(["git", "tag", "-d", "v" + V], work)
            run(["git", "push", "-q", "origin", ":refs/tags/v" + V], work)
            sh(["git", "reset", "-q", "--hard", base_sha], work)

    HEAD = "## v%s — 2026-10-06\n\n- decoy release\n\n" % V
    case("GENUINE: tagged at HEAD and published", HEAD, 0,
         ("newest CHANGELOG version: v9.9.9", "is tagged, placed and published"), tag="head")
    case("GENUINE: an `## Unreleased` heading above the release", "## Unreleased\n\n- wip\n\n" + HEAD, 0,
         ("newest CHANGELOG version: v9.9.9",), tag="head")
    case("GENUINE: a newer version named only in prose and under ###",
         "The `## v10.0.0` heading is not written yet.\n### v10.0.0 notes\n\n" + HEAD, 0,
         ("newest CHANGELOG version: v9.9.9",), tag="head")
    case("GENUINE: the tag only on origin (the shallow-CI shape)", HEAD, 0,
         ("exists on origin",), origin_tag=True)
    case("BY DESIGN: a LOCAL-only tag passes (CI's fresh clone sees only origin)", HEAD, 0,
         ("an ancestor of HEAD",), tag="head")
    case("FACT: the newest heading has no tag anywhere", HEAD, 1, ("DOES NOT EXIST",))
    case("FACT: a tag parked on an unrelated commit", HEAD, 1, ("NOT an ancestor",), tag="unrelated")
    case("FACT: tagged, never published", HEAD, 1, ("IS NOT PUBLISHED",), tag="head", packaged=False)
    case("INCONCLUSIVE: the registry answers 500", HEAD, 2, ("INCONCLUSIVE",), tag="head", pkg_status=500)
    case("FACT beats INCONCLUSIVE: no tag AND the registry answers 500", HEAD, 1, ("DOES NOT EXIST",),
         pkg_status=500)


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)
        release_cases(base, ws)
    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())