From 3f241bc595a6d7323fc9c2c668f0c30935779f2f Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Thu, 29 Oct 2026 20:49:39 +0100 Subject: [PATCH] feat(access-control-prover): 11.5 acquisition provenance and label-isolated mode --- .kiro/specs/access-control-prover/tasks.md | 2 +- src/openultrasast/tools/fact_init.py | 59 ++++++++++++++++++++++ tests/test_fact_init.py | 56 ++++++++++++++++++++ 3 files changed, 116 insertions(+), 1 deletion(-) diff --git a/.kiro/specs/access-control-prover/tasks.md b/.kiro/specs/access-control-prover/tasks.md index 3a1cabc..e28c23d 100644 --- a/.kiro/specs/access-control-prover/tasks.md +++ b/.kiro/specs/access-control-prover/tasks.md @@ -235,7 +235,7 @@ its final resolution, dominance, fact-acquisition and measurement semantics with - _Requirements: 2.9, 2.10_ - _Boundary: Fact Initialization_ - _Depends: 11.3_ -- [ ] 11.5 Record acquisition provenance and enforce label-isolated mode +- [x] 11.5 Record acquisition provenance and enforce label-isolated mode - Persist input revision, evidence digest, accepted/rejected proposals and output digest. - Reject admission acquisition inputs containing vulnerability, fix-location or ordinary-change labels. - Done when changed output after labels are opened marks the population spent. diff --git a/src/openultrasast/tools/fact_init.py b/src/openultrasast/tools/fact_init.py index 1820fc3..c1a5b26 100644 --- a/src/openultrasast/tools/fact_init.py +++ b/src/openultrasast/tools/fact_init.py @@ -157,11 +157,70 @@ def promote_proposals(candidates: tuple[CandidateRecord, ...], promoted: tuple[C __all__ = [ + "AcquisitionRecord", + "AdmissionLabelsExposed", "CandidateRecord", "PromotionResult", + "acquire", "extract_all", "extract_fact_rows", "extract_legacy_strategies", + "population_is_spent", "promote_proposals", "verify_candidate", ] + + +class AdmissionLabelsExposed(ValueError): + """Raised when acquisition input carries admission evaluation labels (Req 12.2).""" + + +FORBIDDEN_LABEL_MARKERS: tuple[str, ...] = ("vulnerability", "fix-location", "ordinary-change") + + +@dataclass(frozen=True) +class AcquisitionRecord: + """Provenance record of one acquisition pass (Req 12.1).""" + + input_revision: str + evidence_digest: str + accepted_names: tuple[str, ...] + rejected_names: tuple[str, ...] + output_digest: str + + +def _corpus_digest(records: tuple[CandidateRecord, ...]) -> str: + lines = sorted( + f"{record.kind}\x1f{record.name}\x1f{record.source}\x1f{record.version}\x1f{record.region}\x1f{record.payload}" + for record in records + ) + return hashlib.sha256("\n".join(lines).encode()).hexdigest()[:12] + + +def acquire( + candidates: tuple[CandidateRecord, ...], + promoted: tuple[CandidateRecord, ...], + *, + input_labels: tuple[str, ...] = (), +) -> AcquisitionRecord: + """Run one label-isolated acquisition pass (Reqs 12.1, 12.2). + + The label guard runs first: an input carrying admission evaluation labels is + rejected before any promotion happens, leaving the promoted corpus untouched. + """ + for label in input_labels: + if any(marker in label for marker in FORBIDDEN_LABEL_MARKERS): + raise AdmissionLabelsExposed(label) + result = promote_proposals(candidates, promoted) + return AcquisitionRecord( + input_revision=_corpus_digest(promoted), + evidence_digest=_corpus_digest(candidates), + accepted_names=tuple(sorted(record.name for record in result.accepted)), + rejected_names=tuple(sorted(record.name for record, _ in result.rejected)), + output_digest=_corpus_digest(result.promoted_after), + ) + + +def population_is_spent(blind_record: AcquisitionRecord, later_output_digest: str) -> bool: + """A changed acquisition output after labels opened spends the population (Req 12.3).""" + return later_output_digest != blind_record.output_digest diff --git a/tests/test_fact_init.py b/tests/test_fact_init.py index 51b2cde..ee0c29a 100644 --- a/tests/test_fact_init.py +++ b/tests/test_fact_init.py @@ -14,11 +14,15 @@ from pathlib import Path import pytest from openultrasast.tools.fact_init import ( + AcquisitionRecord, + AdmissionLabelsExposed, CandidateRecord, PromotionResult, + acquire, extract_all, extract_fact_rows, extract_legacy_strategies, + population_is_spent, promote_proposals, verify_candidate, ) @@ -238,3 +242,55 @@ def test_candidate_record_is_frozen() -> None: ) with pytest.raises(Exception): # noqa: B017 record.name = "other" # type: ignore[misc] + + +def test_label_blind_acquire_records_provenance() -> None: + record = acquire((_proposal(), _contradicted_proposal()), _promoted_corpus()) + assert isinstance(record, AcquisitionRecord) + assert _VERSION_PATTERN.fullmatch(record.input_revision) + assert _VERSION_PATTERN.fullmatch(record.evidence_digest) + assert _VERSION_PATTERN.fullmatch(record.output_digest) + assert record.accepted_names == ("proposed-read",) + assert record.rejected_names == ("promoted-read",) + + +def test_admission_labels_are_rejected_before_promotion() -> None: + promoted = _promoted_corpus() + with pytest.raises(AdmissionLabelsExposed): + acquire((_proposal(),), promoted, input_labels=("vulnerability-fix-label",)) + assert promoted == _promoted_corpus() + + +def test_every_forbidden_marker_raises() -> None: + for label in ("fix-location:Handler.java:4", "ordinary-change:repo-a", "vulnerability:case-9"): + with pytest.raises(AdmissionLabelsExposed): + acquire((_proposal(),), _promoted_corpus(), input_labels=(label,)) + + +def test_acquire_is_deterministic() -> None: + first = acquire((_proposal(),), _promoted_corpus()) + second = acquire((_proposal(),), _promoted_corpus()) + assert first == second + + +def test_population_is_spent_detects_output_change() -> None: + blind = acquire((_proposal(),), _promoted_corpus()) + assert population_is_spent(blind, blind.output_digest) is False + assert population_is_spent(blind, "0" * 12) is True + + +def test_acquisition_record_is_frozen() -> None: + record = acquire((_proposal(),), _promoted_corpus()) + with pytest.raises(Exception): # noqa: B017 + record.output_digest = "0" * 12 # type: ignore[misc] + + +def _contradicted_proposal() -> CandidateRecord: + return CandidateRecord( + kind="operation", + name="promoted-read", + source="proposals.toml", + version="a" * 12, + region="proposals.toml:operation:promoted-read", + payload=(("calls", "write"), ("kind", "protected_read")), + )