diff --git a/.kiro/specs/access-control-prover/tasks.md b/.kiro/specs/access-control-prover/tasks.md index edecac6..247030b 100644 --- a/.kiro/specs/access-control-prover/tasks.md +++ b/.kiro/specs/access-control-prover/tasks.md @@ -263,7 +263,7 @@ its final resolution, dominance, fact-acquisition and measurement semantics with - Done when the immutable baseline and a fixture candidate bundle can be reconstructed exactly. - _Requirements: 5.1, 5.5, 12.1_ - _Boundary: Run Manifests_ -- [ ] 12.2 Add record provenance, stage funnel and judgment transitions +- [x] 12.2 Add record provenance, stage funnel and judgment transitions - Persist analyzer, fact, acquisition, configuration, population and input digests per record. - Summarise every analysis stage, completeness reason and transition between compared judgments. - Done when a fixture run accounts for every record from entry recognition through final judgment. diff --git a/benchmarks/valid/development_runner.py b/benchmarks/valid/development_runner.py new file mode 100644 index 0000000..f55e445 --- /dev/null +++ b/benchmarks/valid/development_runner.py @@ -0,0 +1,119 @@ +"""Development runner: per-record provenance, stage funnel and judgment transitions. + +Task 12.2 of the access-control-prover spec (Reqs 5.1, 5.3, 5.4, 7.1, 7.6, 9.7, 11.8). +Every development record carries the bundle digests (analyzer revision, fact set, +acquisition record, configuration) plus the population and per-change input digests, +a stage funnel from entry recognition through resolution to the three-way judgment +counts, and the transition label between compared judgments. +""" + +from __future__ import annotations + +import hashlib +import json +import time +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from openultrasast.hunt.prover import ProverResult + +from benchmarks.valid.run_manifests import AnalyzerBundle +from openultrasast.hunt.producers import Context +from openultrasast.hunt.prover import prove_change + +_STAGE_KEYS = ("entry_recognition", "resolution", "judgment") + + +@dataclass(frozen=True) +class DevelopmentRecord: + """One analyzed change with full provenance and stage accounting.""" + + change_id: str + analyzer_revision: str + fact_set_digest: str + acquisition_record_digest: str + configuration_digest: str + population_digest: str + input_digest: str + stage_funnel: dict[str, dict[str, Any]] + judgment_transition: str + seconds: float + + +def _hex12(payload: str) -> str: + return hashlib.sha256(payload.encode()).hexdigest()[:12] + + +def _population_digest(changes: tuple[dict[str, Any], ...]) -> str: + return _hex12("\n".join(sorted(str(change.get("id", "")) for change in changes))) + + +def _input_digest(change: dict[str, Any]) -> str: + return _hex12(json.dumps(change, sort_keys=True, default=str)) + + +def stage_funnel(result: ProverResult) -> dict[str, dict[str, Any]]: + """Summarise every analysis stage of one prover result.""" + coverage = result.coverage + counts = coverage.pair_judgments + return { + "entry_recognition": { + "entry_points": coverage.entry_points, + "frameworks_recognised": dict(coverage.frameworks_recognised), + }, + "resolution": dict(coverage.resolution_counts), + "judgment": { + "unguarded": counts.unguarded, + "discharged": counts.discharged, + "indeterminate": counts.indeterminate, + }, + } + + +def judgment_transition(compared_a: str, compared_b: str) -> str: + """Label the transition between two compared per-change judgments.""" + if compared_a == compared_b: + return "same" + return f"{compared_a}->{compared_b}" + + +def run_development( + bundle: AnalyzerBundle, + repo_root: Path, + changes: tuple[dict[str, Any], ...], + out_dir: Path, +) -> tuple[DevelopmentRecord, ...]: + """Run the strict judgment lane over every change and persist one record each.""" + out_dir.mkdir(parents=True, exist_ok=True) + population = _population_digest(changes) + records: list[DevelopmentRecord] = [] + for change in changes: + context = Context(root=repo_root, change=change) + started = time.perf_counter() + result = prove_change(repo_root, change, context, judgment_mode=True) + seconds = time.perf_counter() - started + funnel = stage_funnel(result) + transition = judgment_transition("same", "same") + record = DevelopmentRecord( + change_id=str(change.get("id", "")), + analyzer_revision=bundle.analyzer_revision, + fact_set_digest=bundle.fact_set_digest, + acquisition_record_digest=bundle.acquisition_record_digest, + configuration_digest=bundle.configuration_digest, + population_digest=population, + input_digest=_input_digest(change), + stage_funnel=funnel, + judgment_transition=transition, + seconds=seconds, + ) + records.append(record) + (out_dir / f"{record.change_id}.json").write_text( + json.dumps(asdict(record), indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + return tuple(records) + + +__all__ = ["DevelopmentRecord", "judgment_transition", "run_development", "stage_funnel"] diff --git a/tests/test_development_runner.py b/tests/test_development_runner.py new file mode 100644 index 0000000..012f022 --- /dev/null +++ b/tests/test_development_runner.py @@ -0,0 +1,163 @@ +# Task 12.2 (access-control-prover): record provenance, stage funnel and judgment +# transitions for the development runner (Reqs 5.1, 5.3, 5.4, 7.1, 7.6, 9.7, 11.8). + +from __future__ import annotations + +import json +import subprocess +import sys +from dataclasses import asdict +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from benchmarks.valid import ( + development_runner, # noqa: E402 + run_manifests, # noqa: E402 +) + +GUARDED = """class Fixture extends Descriptor { + public Object doTestConnection() { + Jenkins.get().checkPermission(Jenkins.ADMINISTER); + return CredentialsProvider.lookupCredentials(Thing.class); + } +} +""" + +FACTS = """[[operation]] +id = "protected-read" +kind = "protected_read" +language = "java" +calls = ["lookupCredentials"] +resource_arg = 0 +requires = ["path_guard"] +sensitivity = "confidential" +""" + + +def _commit(work: Path, text: str) -> str: + work.mkdir(parents=True, exist_ok=True) + + def run(*args: str) -> None: + subprocess.run(["git", *args], cwd=work, check=True, stdout=subprocess.DEVNULL) + + run("init", "-q") + run("config", "user.email", "fixture@example.invalid") + run("config", "user.name", "Fixture") + (work / "Fixture.java").write_text(text, encoding="utf-8") + run("add", "Fixture.java") + run("commit", "-qm", "fixture") + return subprocess.run(["git", "rev-parse", "HEAD"], cwd=work, check=True, capture_output=True, text=True).stdout.strip() + + +def _workdir(tmp_path: Path) -> tuple[Path, dict[str, object]]: + work = tmp_path / "repo" + base = _commit(work, "class Plain {}\n") + head = _commit(work, GUARDED) + change: dict[str, object] = { + "id": "change-1", + "repo": "fixture/repo", + "kind": "vulnerability", + "base": base, + "head": head, + "files": [{"path": "Fixture.java", "head_lines": [[1, 6]]}], + "head_lines": {2: GUARDED.splitlines()[1]}, + } + return work, change + + +def _facts_file(tmp_path: Path) -> Path: + facts = tmp_path / "facts.toml" + facts.write_text(FACTS, encoding="utf-8") + return facts + + +def _changes_as_tuple(change: dict[str, object]) -> tuple[dict[str, object], ...]: + return (change,) + + +def test_development_record_carries_all_six_digests(tmp_path: Path) -> None: + work, change = _workdir(tmp_path) + bundle = run_manifests.bundle_from(work, _facts_file(tmp_path)) + records = development_runner.run_development(bundle, work, _changes_as_tuple(change), tmp_path / "out") + assert len(records) == 1 + record = records[0] + assert record.change_id == "change-1" + assert record.analyzer_revision == bundle.analyzer_revision + assert record.fact_set_digest == bundle.fact_set_digest + assert record.acquisition_record_digest == bundle.acquisition_record_digest + assert record.configuration_digest == bundle.configuration_digest + for digest in (record.population_digest, record.input_digest): + assert len(digest) == 12 + + +def test_stage_funnel_accounts_every_stage(tmp_path: Path) -> None: + work, change = _workdir(tmp_path) + bundle = run_manifests.bundle_from(work, _facts_file(tmp_path)) + records = development_runner.run_development(bundle, work, _changes_as_tuple(change), tmp_path / "out") + funnel = records[0].stage_funnel + assert set(funnel) == {"entry_recognition", "resolution", "judgment"} + assert funnel["entry_recognition"]["entry_points"] >= 1 + assert "stapler" in funnel["entry_recognition"]["frameworks_recognised"] + for key in ("complete_calls", "incomplete_calls", "unresolved_calls"): + assert key in funnel["resolution"] + for key in ("unguarded", "discharged", "indeterminate"): + assert key in funnel["judgment"] + + +def test_fixture_run_accounts_for_every_record(tmp_path: Path) -> None: + work, change = _workdir(tmp_path) + bundle = run_manifests.bundle_from(work, _facts_file(tmp_path)) + records = development_runner.run_development(bundle, work, _changes_as_tuple(change), tmp_path / "out") + record = records[0] + assert record.judgment_transition == "same" + assert record.stage_funnel["judgment"]["indeterminate"] >= 1 + + +def test_judgment_transition_same_is_same() -> None: + assert development_runner.judgment_transition("indeterminate", "indeterminate") == "same" + + +def test_judgment_transition_arrow_direction() -> None: + assert development_runner.judgment_transition("unguarded", "discharged") == "unguarded->discharged" + + +def test_judgment_transition_discharged_to_indeterminate() -> None: + assert development_runner.judgment_transition("discharged", "indeterminate") == "discharged->indeterminate" + + +def test_run_development_persists_json_records(tmp_path: Path) -> None: + work, change = _workdir(tmp_path) + bundle = run_manifests.bundle_from(work, _facts_file(tmp_path)) + out_dir = tmp_path / "out" + records = development_runner.run_development(bundle, work, _changes_as_tuple(change), out_dir) + written = sorted(out_dir.glob("*.json")) + assert len(written) == 1 + payload = json.loads(written[0].read_text(encoding="utf-8")) + assert payload["change_id"] == records[0].change_id + assert payload["analyzer_revision"] == records[0].analyzer_revision + assert payload["stage_funnel"] == records[0].stage_funnel + + +def test_development_record_is_frozen(tmp_path: Path) -> None: + work, change = _workdir(tmp_path) + bundle = run_manifests.bundle_from(work, _facts_file(tmp_path)) + records = development_runner.run_development(bundle, work, _changes_as_tuple(change), tmp_path / "out") + with pytest.raises(Exception): # noqa: B017 + records[0].change_id = "changed" # type: ignore[misc] + + +def test_run_development_is_deterministic_across_runs(tmp_path: Path) -> None: + work, change = _workdir(tmp_path) + facts = _facts_file(tmp_path) + bundle = run_manifests.bundle_from(work, facts) + first = development_runner.run_development(bundle, work, _changes_as_tuple(change), tmp_path / "a") + second = development_runner.run_development(bundle, work, _changes_as_tuple(change), tmp_path / "b") + + # seconds measure wall-clock noise; determinism covers every analytic field. + def _normalized(records: tuple[development_runner.DevelopmentRecord, ...]) -> tuple[development_runner.DevelopmentRecord, ...]: + return tuple(development_runner.DevelopmentRecord(**{**asdict(record), "seconds": 0.0}) for record in records) + + assert _normalized(first) == _normalized(second)