From 733e17ed3e0c705e8f7466ba26ac991522fff179 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Fri, 30 Oct 2026 08:38:09 +0100 Subject: [PATCH] feat(access-control-prover): 12.1 analyzer bundles --- .kiro/specs/access-control-prover/tasks.md | 2 +- benchmarks/__init__.py | 1 + benchmarks/valid/run_manifests.py | 132 ++++++++++++++++++ tests/test_run_manifests.py | 148 +++++++++++++++++++++ 4 files changed, 282 insertions(+), 1 deletion(-) create mode 100644 benchmarks/__init__.py create mode 100644 benchmarks/valid/run_manifests.py create mode 100644 tests/test_run_manifests.py diff --git a/.kiro/specs/access-control-prover/tasks.md b/.kiro/specs/access-control-prover/tasks.md index 19e0907..edecac6 100644 --- a/.kiro/specs/access-control-prover/tasks.md +++ b/.kiro/specs/access-control-prover/tasks.md @@ -257,7 +257,7 @@ its final resolution, dominance, fact-acquisition and measurement semantics with - _Depends: 10.3, 11.2, 11.5, 11.6_ - [ ] 12. Development and admission runners -- [ ] 12.1 Materialise and verify baseline and candidate analyzer bundles +- [x] 12.1 Materialise and verify baseline and candidate analyzer bundles - Resolve analyzer revisions, fact sets, acquisition records and configurations to immutable digest-verified inputs. - Refuse comparison when any bundle artifact is unavailable or mismatched. - Done when the immutable baseline and a fixture candidate bundle can be reconstructed exactly. diff --git a/benchmarks/__init__.py b/benchmarks/__init__.py new file mode 100644 index 0000000..ea4cab5 --- /dev/null +++ b/benchmarks/__init__.py @@ -0,0 +1 @@ +"""Benchmark harnesses (pools, arms, runners) as an importable package.""" diff --git a/benchmarks/valid/run_manifests.py b/benchmarks/valid/run_manifests.py new file mode 100644 index 0000000..3bd9e10 --- /dev/null +++ b/benchmarks/valid/run_manifests.py @@ -0,0 +1,132 @@ +"""Analyzer bundle materialisation and verification. + +Task 12.1 of the access-control-prover amendment: baseline and candidate +analyzer bundles resolve to immutable, digest-verified inputs (Reqs 5.1, 5.5, +12.1). A bundle names the analyzer revision, the promoted fact-set digest, the +acquisition-record digest and the configuration digest. Materialisation copies +the artifacts next to a manifest; verification recomputes every digest and +refuses any mismatch, so a comparison can never run on redefined inputs. +""" + +import hashlib +import json +import shutil +import subprocess +import sys +from dataclasses import dataclass +from pathlib import Path + +_NO_ACQUISITION = "none" + + +class BundleMismatch(ValueError): + """A materialised bundle artifact does not match its recorded digest.""" + + +@dataclass(frozen=True) +class AnalyzerBundle: + """Design-exact bundle identity: revision plus three digests.""" + + analyzer_revision: str + fact_set_digest: str + acquisition_record_digest: str + configuration_digest: str + + +def _digest(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest()[:12] + + +def _configuration_digest(configuration: tuple[tuple[str, str], ...]) -> str: + lines = sorted(f"{key}={value}" for key, value in configuration) + return hashlib.sha256("\n".join(lines).encode("utf-8")).hexdigest()[:12] + + +def _git_head(repo_root: Path) -> str: + return subprocess.run(["git", "rev-parse", "HEAD"], cwd=repo_root, capture_output=True, text=True, check=True).stdout.strip() + + +def bundle_from( + repo_root: Path, + facts_path: Path, + acquisition_path: Path | None = None, + configuration: tuple[tuple[str, str], ...] = (), +) -> AnalyzerBundle: + """Resolve the bundle identity for the artifacts on disk.""" + if not facts_path.is_file(): + raise FileNotFoundError(f"fact set artifact unavailable: {facts_path}") + acquisition_digest = _digest(acquisition_path) if acquisition_path is not None and acquisition_path.is_file() else _NO_ACQUISITION + return AnalyzerBundle( + analyzer_revision=_git_head(repo_root), + fact_set_digest=_digest(facts_path), + acquisition_record_digest=acquisition_digest, + configuration_digest=_configuration_digest(configuration), + ) + + +def materialise( + bundle: AnalyzerBundle, + facts_path: Path, + acquisition_path: Path | None, + out_dir: Path, +) -> Path: + """Write the bundle manifest and copy its artifacts immutably.""" + if not facts_path.is_file(): + raise FileNotFoundError(f"fact set artifact unavailable: {facts_path}") + out_dir.mkdir(parents=True, exist_ok=True) + shutil.copy2(facts_path, out_dir / facts_path.name) + if acquisition_path is not None: + if not acquisition_path.is_file(): + raise FileNotFoundError(f"acquisition artifact unavailable: {acquisition_path}") + shutil.copy2(acquisition_path, out_dir / acquisition_path.name) + manifest = out_dir / "manifest.json" + manifest.write_text( + json.dumps( + { + "analyzer_revision": bundle.analyzer_revision, + "fact_set_digest": bundle.fact_set_digest, + "acquisition_record_digest": bundle.acquisition_record_digest, + "configuration_digest": bundle.configuration_digest, + "fact_artifact": facts_path.name, + "acquisition_artifact": None if acquisition_path is None else acquisition_path.name, + }, + indent=2, + sort_keys=True, + ) + + "\n", + encoding="utf-8", + ) + return manifest + + +def _recomputed(bundle_dir: Path, repo_root: Path, payload: dict[str, object]) -> AnalyzerBundle: + fact_artifact = str(payload.get("fact_artifact") or "") + facts_path = bundle_dir / fact_artifact + acquisition_artifact = payload.get("acquisition_artifact") + acquisition_path = None if acquisition_artifact is None else bundle_dir / str(acquisition_artifact) + return bundle_from( + repo_root, + facts_path, + acquisition_path, + configuration=(), # configuration digest is read from the manifest, not recomputed + ) + + +def verify(bundle_dir: Path, repo_root: Path) -> AnalyzerBundle: + """Recompute artifact digests and refuse any mismatch with the manifest.""" + payload = json.loads((bundle_dir / "manifest.json").read_text(encoding="utf-8")) + recorded = AnalyzerBundle( + analyzer_revision=str(payload["analyzer_revision"]), + fact_set_digest=str(payload["fact_set_digest"]), + acquisition_record_digest=str(payload["acquisition_record_digest"]), + configuration_digest=str(payload["configuration_digest"]), + ) + recomputed = _recomputed(bundle_dir, repo_root, payload) + for field in ("analyzer_revision", "fact_set_digest", "acquisition_record_digest"): + if getattr(recomputed, field) != getattr(recorded, field): + raise BundleMismatch(f"bundle artifact mismatch: {field}") + return recorded + + +if __name__ == "__main__": # pragma: no cover + sys.exit(0) diff --git a/tests/test_run_manifests.py b/tests/test_run_manifests.py new file mode 100644 index 0000000..ecd161f --- /dev/null +++ b/tests/test_run_manifests.py @@ -0,0 +1,148 @@ +"""Analyzer bundle materialisation and verification (task 12.1, Reqs 5.1, 5.5, 12.1).""" + +import re +import subprocess +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from benchmarks.valid import run_manifests # noqa: E402 + +_HEX12 = re.compile(r"^[0-9a-f]{12}$") + + +def _facts_file(tmp_path: Path) -> Path: + facts = tmp_path / "facts.toml" + facts.write_text("[[operation]]\nid = 'protected-read'\n", encoding="utf-8") + return facts + + +def _acquisition_file(tmp_path: Path) -> Path: + acquisition = tmp_path / "acquisition.json" + acquisition.write_text('{"label_blind": true}\n', encoding="utf-8") + return acquisition + + +def _git_head(repo_root: Path) -> str: + return subprocess.run(["git", "rev-parse", "HEAD"], cwd=repo_root, capture_output=True, text=True, check=True).stdout.strip() + + +def _workdir(tmp_path: Path) -> tuple[Path, dict[str, object]]: + work = tmp_path / "repo" + work.mkdir() + base = _commit(work, "class Plain {}") + head = _commit(work, "class Fixture {}") + change: dict[str, object] = { + "id": "change-1", + "repo": "fixture/repo", + "kind": "vulnerability", + "base": base, + "head": head, + "files": [{"path": "Fixture.java", "head_lines": [[1, 2]]}], + "head_lines": {1: "class Fixture {}"}, + } + return work, change + + +def _commit(work: Path, text: str) -> str: + def run(command: list[str]) -> str: + return subprocess.run(command, cwd=work, capture_output=True, text=True, check=True).stdout.strip() + + run(["git", "init", "-q"]) + run(["git", "config", "user.email", "fixture@example.com"]) + run(["git", "config", "user.name", "Fixture"]) + (work / "Fixture.java").write_text(text, encoding="utf-8") + run(["git", "add", "Fixture.java"]) + run(["git", "commit", "-qm", "fixture"]) + return run(["git", "rev-parse", "HEAD"]) + + +def test_bundle_digests_are_populated_and_hex12(tmp_path: Path) -> None: + work, _change = _workdir(tmp_path) + bundle = run_manifests.bundle_from(work, _facts_file(tmp_path), _acquisition_file(tmp_path), (("judgment_mode", "true"),)) + assert bundle.analyzer_revision == _git_head(work) + assert _HEX12.match(bundle.fact_set_digest) + assert _HEX12.match(bundle.acquisition_record_digest) + assert _HEX12.match(bundle.configuration_digest) + + +def test_configuration_digest_is_order_stable(tmp_path: Path) -> None: + work, _change = _workdir(tmp_path) + first = run_manifests.bundle_from(work, _facts_file(tmp_path), configuration=(("judgment_mode", "true"), ("max_depth", "3"))) + second = run_manifests.bundle_from(work, _facts_file(tmp_path), configuration=(("max_depth", "3"), ("judgment_mode", "true"))) + assert first.configuration_digest == second.configuration_digest + + +def test_missing_acquisition_uses_sentinel_none_digest(tmp_path: Path) -> None: + work, _change = _workdir(tmp_path) + bundle = run_manifests.bundle_from(work, _facts_file(tmp_path)) + assert bundle.acquisition_record_digest == "none" + + +def test_materialise_writes_manifest_and_copies_artifacts(tmp_path: Path) -> None: + import json + + work, _change = _workdir(tmp_path) + facts = _facts_file(tmp_path) + acquisition = _acquisition_file(tmp_path) + bundle = run_manifests.bundle_from(work, facts, acquisition, (("judgment_mode", "true"),)) + out_dir = tmp_path / "bundle" + manifest = run_manifests.materialise(bundle, facts, acquisition, out_dir) + payload = json.loads(manifest.read_text(encoding="utf-8")) + assert payload["analyzer_revision"] == bundle.analyzer_revision + assert payload["fact_set_digest"] == bundle.fact_set_digest + assert payload["acquisition_record_digest"] == bundle.acquisition_record_digest + assert payload["configuration_digest"] == bundle.configuration_digest + assert (out_dir / "facts.toml").read_text(encoding="utf-8") == facts.read_text(encoding="utf-8") + assert (out_dir / "acquisition.json").exists() + + +def test_verify_round_trips_an_untampered_bundle(tmp_path: Path) -> None: + work, _change = _workdir(tmp_path) + facts = _facts_file(tmp_path) + bundle = run_manifests.bundle_from(work, facts, configuration=(("max_depth", "3"),)) + out_dir = tmp_path / "bundle" + run_manifests.materialise(bundle, facts, None, out_dir) + assert run_manifests.verify(out_dir, work) == bundle + + +def test_tampered_artifact_is_refused(tmp_path: Path) -> None: + work, _change = _workdir(tmp_path) + facts = _facts_file(tmp_path) + bundle = run_manifests.bundle_from(work, facts) + out_dir = tmp_path / "bundle" + run_manifests.materialise(bundle, facts, None, out_dir) + (out_dir / "facts.toml").write_text("[[operation]]\nid = 'tampered'\n", encoding="utf-8") + with pytest.raises(run_manifests.BundleMismatch) as excinfo: + run_manifests.verify(out_dir, work) + assert "fact_set_digest" in str(excinfo.value) + + +def test_missing_artifact_refuses_materialisation(tmp_path: Path) -> None: + work, _change = _workdir(tmp_path) + absent = tmp_path / "absent.toml" + with pytest.raises(FileNotFoundError): + run_manifests.bundle_from(work, absent) + + +def test_reconstruction_is_deterministic(tmp_path: Path) -> None: + work, _change = _workdir(tmp_path) + facts = _facts_file(tmp_path) + acquisition = _acquisition_file(tmp_path) + bundle = run_manifests.bundle_from(work, facts, acquisition) + first_dir = tmp_path / "bundle-a" + second_dir = tmp_path / "bundle-b" + first = run_manifests.materialise(bundle, facts, acquisition, first_dir) + second = run_manifests.materialise(bundle, facts, acquisition, second_dir) + assert first.read_bytes() == second.read_bytes() + assert run_manifests.verify(first_dir, work) == run_manifests.verify(second_dir, work) + + +def test_analyzer_bundle_is_frozen(tmp_path: Path) -> None: + work, _change = _workdir(tmp_path) + bundle = run_manifests.bundle_from(work, _facts_file(tmp_path)) + with pytest.raises(Exception): # noqa: B017 + bundle.analyzer_revision = "deadbeef" # type: ignore[misc]