#!/usr/bin/env python3 """Small candidate/transport acceptance tests; payloads are synthetic, not builds.""" import hashlib from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import importlib.util import json import os from pathlib import Path import shutil import subprocess import sys import tempfile import threading from types import SimpleNamespace import unittest ROOT = Path(__file__).resolve().parents[1] SPEC = importlib.util.spec_from_file_location("adapter", ROOT / "scripts/release-contract.py") ADAPTER = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(ADAPTER) class CandidateTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory(prefix="redflag-acceptance-") self.addCleanup(self.tmp.cleanup) self.root = Path(self.tmp.name) self.source = self.root / "artifacts" self.candidate = self.root / "candidate" self.commit = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=ROOT, text=True).strip() for target in sorted(ADAPTER.TARGETS): directory = self.source / ("release-" + target) directory.mkdir(parents=True) os_name, arch = target.split("-") snippets = [] for component in ("server", "agent", "helper"): if component == "helper" and os_name == "windows": continue name = f"redflag-{component}-{target}" + (".exe" if os_name == "windows" else "") payload = ("synthetic " + name).encode() (directory / name).write_bytes(payload) snippets.append(dict(filename=name, platform=os_name if component == "agent" else component + "-" + os_name, architecture=arch, sha256=hashlib.sha256(payload).hexdigest(), size=len(payload))) if target == "linux-amd64": (directory / "redflag_0.2.9.4_amd64.deb").write_bytes(b"synthetic deb") if target == "windows-amd64": (directory / "RedFlagSetup-windows-amd64.msi").write_bytes(b"synthetic msi") (directory / ("release-" + target + ".artifacts.json")).write_text(json.dumps(snippets)) ADAPTER.assemble(SimpleNamespace(artifacts=str(self.source), candidate=str(self.candidate), commit=self.commit, version="0.2.9.4", tag="v0.2.9.4", run_id=os.environ.get("GITHUB_RUN_ID", "123"))) def verify(self): ADAPTER.verify(self.candidate, "0.2.9.4", self.commit, "v0.2.9.4") def rewrite_receipt(self): (self.candidate / "files.sha256").write_text("".join( f"{ADAPTER.CONTRACT.sha256(p)} {p.name}\n" for p in sorted(self.candidate.iterdir()) if p.name != "files.sha256")) def test_clean_candidate_and_legacy_fields(self): self.verify() manifest = json.loads((self.candidate / "manifest.json").read_text()) self.assertTrue(any(a.get("platform") == "server-linux" for a in manifest["artifacts"])) def test_altered_bytes_rejected(self): (self.candidate / "RedFlagSetup-windows-amd64.msi").write_bytes(b"tampered") with self.assertRaisesRegex(ValueError, "receipt digest mismatch"): self.verify() def test_forged_builder_with_fresh_receipt_rejected(self): path = self.candidate / "manifest.json" data = json.loads(path.read_text()) data["build"]["builder_id"] = "other-builder" path.write_text(json.dumps(data)) self.rewrite_receipt() with self.assertRaisesRegex(ValueError, "unexpected builder"): self.verify() def test_wrong_repository_with_consistent_provenance_rejected(self): path = self.candidate / "manifest.json" data = json.loads(path.read_text()) data["source"]["repository"] = "https://example.invalid/source" path.write_text(json.dumps(data)) path = self.candidate / "provenance.json" data = json.loads(path.read_text()) data["predicate"]["buildDefinition"]["resolvedDependencies"][0]["uri"] = "https://example.invalid/source" path.write_text(json.dumps(data)) self.rewrite_receipt() with self.assertRaisesRegex(ValueError, "unexpected source repository"): self.verify() def test_duplicate_receipt_rejected(self): path = self.candidate / "files.sha256" path.write_text(path.read_text() + path.read_text().splitlines()[0] + "\n") with self.assertRaisesRegex(ValueError, "duplicate receipt"): self.verify() def test_traversal_receipt_rejected(self): (self.candidate / "files.sha256").write_text("0" * 64 + " ../victim\n") with self.assertRaisesRegex(ValueError, "unsafe candidate filename"): self.verify() def test_unreceipted_file_rejected(self): (self.candidate / "extra").write_bytes(b"extra") with self.assertRaisesRegex(ValueError, "unreceipted"): self.verify() def test_custody_transport_and_tampered_download(self): shelf, uploads = {}, [] class Shelf(BaseHTTPRequestHandler): def log_message(self, *args): pass def do_PUT(self): self.assert_auth() data = self.rfile.read(int(self.headers["Content-Length"])) status = 409 if self.path in shelf else 201 if status == 201: shelf[self.path] = data uploads.append(self.path) self.send_response(status) self.end_headers() def do_GET(self): self.assert_auth() self.send_response(200 if self.path in shelf else 404) self.end_headers() self.wfile.write(shelf.get(self.path, b"")) def assert_auth(self): assert self.headers["Authorization"] == "token synthetic-test-token" server = ThreadingHTTPServer(("127.0.0.1", 0), Shelf) self.addCleanup(server.server_close) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() self.addCleanup(server.shutdown) env = dict(os.environ, GITHUB_SERVER_URL=f"http://127.0.0.1:{server.server_port}", PACKAGE_WRITE_TOKEN="synthetic-test-token") def run(script, *args, ok=True): result = subprocess.run(["bash", str(ROOT / "scripts" / script), *map(str, args)], env=env, capture_output=True, text=True) self.assertEqual(result.returncode == 0, ok, result.stdout + result.stderr) run("stage-release-candidate.sh", self.candidate, "artifacts", "0.2.9.4") self.assertTrue(uploads[-1].endswith("/files.sha256")) self.assertEqual(len(uploads), len(list(self.candidate.iterdir()))) run("stage-release-candidate.sh", self.candidate, "artifacts", "0.2.9.4") token = self.root / "token" token.write_text("synthetic-test-token") digest = ADAPTER.CONTRACT.sha256(self.candidate / "files.sha256") output = self.root / "fetched" run("fetch-release-candidate.sh", output, "artifacts", "0.2.9.4", digest, token) ADAPTER.verify(output, "0.2.9.4", self.commit, "v0.2.9.4") run("fetch-release-candidate.sh", output, "artifacts", "0.2.9.4", digest, token, ok=False) shelf[next(p for p in shelf if p.endswith(".msi"))] = b"tampered" run("fetch-release-candidate.sh", self.root / "bad", "artifacts", "0.2.9.4", digest, token, ok=False) run("fetch-release-candidate.sh", self.root / "wrong-receipt", "artifacts", "0.2.9.4", "0" * 64, token, ok=False) if __name__ == "__main__": if len(sys.argv) == 3 and sys.argv[1] == "--export-fixture": fixture = CandidateTests() try: fixture.setUp() fixture.verify() shutil.copytree(fixture.candidate, sys.argv[2]) print("SYNTHETIC admission fixture: not a signed release or installable software") print("ADMISSION_RECEIPT_SHA256=" + ADAPTER.CONTRACT.sha256(fixture.candidate / "files.sha256")) finally: fixture.doCleanups() else: unittest.main()