67e26be2d9
The public tree and its history contain only the listed paths. Earlier projection history remains preserved internally. Source-Sha: 913fde029b935671833254797f0f20f1eb9fabba Policy-Sha: 913fde029b935671833254797f0f20f1eb9fabba Tree-Digest: 180ae530c1058a2a5c89837bdce2d323ae83e669e38590ca72e75b8d92b7262f
165 lines
8 KiB
Python
165 lines
8 KiB
Python
#!/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()
|