diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index a562ab3..cabae5a 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -343,7 +343,7 @@ - _Requirements: 8.1, 8.3, 8.4, 8.5, 8.6, 14.3, 15.5_ - _Boundary: Benchmark Readiness Adapter_ - _Depends: 4.4, 8.2_ -- [ ] 9.6 Integrate benchmark-owned admission scoring and guarded custody +- [x] 9.6 Integrate benchmark-owned admission scoring and guarded custody - Delegate neutral scoring to core while retaining population persistence and run orchestration in the benchmark. - Make the custodian the sole append-only writer and combine durable compare-and-swap spend plus label release as one guarded operation. - After the winning spend, produce and independently attest the canonical case release over the exact immutable result, evidence, change, and input identities supplied to scoring. diff --git a/benchmarks/valid/admission_run.py b/benchmarks/valid/admission_run.py index a45ac5f..dfeae8e 100644 --- a/benchmarks/valid/admission_run.py +++ b/benchmarks/valid/admission_run.py @@ -18,13 +18,19 @@ from __future__ import annotations import hashlib import json import tomllib +from collections.abc import Callable, Sequence +from dataclasses import dataclass from pathlib import Path from typing import Any from benchmarks.unseen.replay import canonical -from benchmarks.valid.admission_scoring import AdmissionCase, OrdinaryChange, admission_report +from benchmarks.valid.admission_scoring import AdmissionCase, OrdinaryChange, admission_report, score_v2_admission from benchmarks.valid.arm_regression import BundleDigests, _input_digest, _prove_one +from benchmarks.valid.population_custody import AdmissionSpendAuthority, GuardedPopulationCustodian, ReleaseReview from benchmarks.valid.prover_admission import AdmissionManifest, validate_admission_manifest +from openultrasast.core.admission_case import AdmissionCaseResult +from openultrasast.core.admission_report import AdmissionReplay +from openultrasast.core.admission_spend import authorize_admission from openultrasast.preprocess import detect_language from openultrasast.source import canonical_language, ranges @@ -37,12 +43,57 @@ __all__ = [ "load_admission_population", "locate_function", "run_admission", + "run_v2_admission", + "V2AdmissionRunInput", "site_judgment", ] ARM_NAME = "access-control-admission" +@dataclass(frozen=True) +class V2AdmissionRunInput: + authority: AdmissionSpendAuthority + custodian: GuardedPopulationCustodian + load_labels: Callable[[], bytes] + analyze_labels: Callable[[bytes], Sequence[AdmissionCaseResult]] + review_release: ReleaseReview + + +def run_v2_admission(value: V2AdmissionRunInput) -> AdmissionReplay: + """Run one synthetic v2 admission through guarded custody and core scoring.""" + authority = value.authority + preparation = authorize_admission( + authority.manifest, + authority.denominator, + authority.readiness, + authority.context, + value.custodian, + authority.resolver, + ) + guarded = value.custodian.spend_and_release_labels(authority, preparation, value.load_labels) + cases = tuple(value.analyze_labels(guarded.labels)) + release = value.custodian.release_cases( + authority, + preparation, + guarded.receipt, + cases, + value.review_release, + ) + return score_v2_admission( + authority.manifest, + authority.denominator, + authority.readiness, + preparation, + guarded.receipt, + release, + cases, + authority.context, + value.custodian, + authority.resolver, + ) + + def _site_traces(record: dict[str, Any], path: str, start: int, end: int) -> list[dict[str, Any]]: """Traces whose operation site falls inside the labelled range.""" matched = [] diff --git a/benchmarks/valid/admission_scoring.py b/benchmarks/valid/admission_scoring.py index cbac425..5bc1580 100644 --- a/benchmarks/valid/admission_scoring.py +++ b/benchmarks/valid/admission_scoring.py @@ -10,13 +10,23 @@ indeterminate and never converted to clean results; ordinary changes whose review found a real missing-authorization defect are refused before scoring. """ +from collections.abc import Sequence from dataclasses import dataclass from typing import Any from benchmarks.unseen import score from benchmarks.valid.prover_admission import AdmissionManifest +from openultrasast.core import canonical +from openultrasast.core.admission_case import AdmissionCaseRelease, AdmissionCaseResult +from openultrasast.core.admission_context import AdmissionValidationContext +from openultrasast.core.admission_report import AdmissionReplay, score_admission, validate_admission_replay +from openultrasast.core.admission_review import AdmissionManifestV2 +from openultrasast.core.admission_spend import AdmissionPreparation +from openultrasast.core.custody import PopulationCustodyPort, PopulationUseReceipt +from openultrasast.core.measurement import DenominatorPolicy +from openultrasast.core.readiness import ReadinessResult -__all__ = ["AdmissionCase", "AdmissionScoringError", "OrdinaryChange", "admission_report"] +__all__ = ["AdmissionCase", "AdmissionScoringError", "OrdinaryChange", "admission_report", "score_v2_admission"] _JUDGMENTS = ("unguarded", "discharged", "indeterminate") @@ -45,6 +55,35 @@ class OrdinaryChange: mislabeled_defect: bool +def score_v2_admission( + manifest: AdmissionManifestV2, + denominator: DenominatorPolicy, + readiness: Sequence[ReadinessResult], + preparation: AdmissionPreparation, + receipt: PopulationUseReceipt, + release: AdmissionCaseRelease, + cases: Sequence[AdmissionCaseResult], + context: AdmissionValidationContext, + custody: PopulationCustodyPort, + resolver: canonical.ContentResolver, +) -> AdmissionReplay: + """Delegate v2 scoring and authoritative replay to detection core.""" + replay = score_admission( + manifest, + denominator, + tuple(readiness), + preparation, + receipt, + release, + tuple(cases), + context, + custody, + resolver, + ) + validate_admission_replay(replay, context, custody, resolver) + return replay + + def _refuse_inputs( manifest: AdmissionManifest, cases: tuple[AdmissionCase, ...], diff --git a/benchmarks/valid/population_custody.py b/benchmarks/valid/population_custody.py new file mode 100644 index 0000000..134a9e2 --- /dev/null +++ b/benchmarks/valid/population_custody.py @@ -0,0 +1,243 @@ +"""Benchmark-owned guarded population custody for v2 admission.""" + +from __future__ import annotations + +from collections.abc import Callable, Sequence +from dataclasses import dataclass, replace +from typing import Protocol + +from openultrasast.core import canonical +from openultrasast.core.admission_case import ( + ADMISSION_CASE_RELEASE_SCHEMA, + AdmissionCaseRelease, + AdmissionCaseResult, + ReleasedCaseBinding, + derive_admission_case_release_digest, + derive_admission_case_release_payload_digest, + derive_released_case_binding_digest, + validate_admission_case_release, +) +from openultrasast.core.admission_context import AdmissionValidationContext +from openultrasast.core.admission_review import AdmissionManifestV2 +from openultrasast.core.admission_spend import ( + AdmissionPreparation, + AdmissionSpendError, + authorize_admission, + validate_pre_label_spend, +) +from openultrasast.core.custody import ( + PopulationCustodyError, + PopulationCustodyLog, + PopulationCustodyPolicy, + PopulationCustodyPort, + PopulationUseReceipt, + append_population_use, + record_population_use, + validate_population_custody_log, +) +from openultrasast.core.factpack import IndependenceAttestation +from openultrasast.core.measurement import DenominatorPolicy +from openultrasast.core.readiness import ReadinessResult + +ReleaseReview = Callable[[str, str], IndependenceAttestation] + + +class GuardedCustodyError(RuntimeError): + """A guarded append or label release did not complete.""" + + +class CompareAndSwapCustodyStore(Protocol): + """Durable storage seam owned by the benchmark custodian.""" + + def current_log(self, population_digest: str) -> PopulationCustodyLog: ... + + def compare_and_swap(self, expected_log_digest: str, replacement: PopulationCustodyLog) -> bool: ... + + +@dataclass(frozen=True) +class GuardedLabelRelease: + receipt: PopulationUseReceipt + labels: bytes + + +@dataclass(frozen=True) +class AdmissionSpendAuthority: + manifest: AdmissionManifestV2 + denominator: DenominatorPolicy + readiness: tuple[ReadinessResult, ...] + context: AdmissionValidationContext + resolver: canonical.ContentResolver + + +class GuardedPopulationCustodian: + """Sole writer for one admission population.""" + + def __init__( + self, + population_digest: str, + custodian_identity: str, + policy: PopulationCustodyPolicy, + store: CompareAndSwapCustodyStore, + background: PopulationCustodyPort, + ) -> None: + self.population_digest = population_digest + self.custodian_identity = custodian_identity + self.policy = policy + self.store = store + self.background = background + + def current_log(self, population_digest: str) -> PopulationCustodyLog: + if population_digest == self.population_digest: + return self.store.current_log(population_digest) + return self.background.current_log(population_digest) + + def spend_and_release_labels( + self, + authority: AdmissionSpendAuthority, + preparation: AdmissionPreparation, + load_labels: Callable[[], bytes], + ) -> GuardedLabelRelease: + expected = self._authorize(authority) + if preparation != expected: + raise GuardedCustodyError("supplied preparation is not the authoritative core preparation") + request = preparation.label_release_request + if request.population_digest != self.population_digest: + raise GuardedCustodyError("label-release request names another population") + prior = self.store.current_log(self.population_digest) + try: + validate_population_custody_log(prior, self.policy) + receipt = record_population_use(request, prior, self.custodian_identity, self.policy) + replacement = append_population_use(prior, receipt, self.policy) + except PopulationCustodyError as exc: + raise GuardedCustodyError(f"custody successor refused: {exc}") from exc + if not self.store.compare_and_swap(prior.log_digest, replacement): + raise GuardedCustodyError("compare-and-swap lost the current custody head") + committed = self.store.current_log(self.population_digest) + validate_population_custody_log(committed, self.policy) + if committed != replacement: + raise GuardedCustodyError("durable custody append does not match the committed successor") + self._validate_spend(authority, preparation, receipt) + labels = load_labels() + if type(labels) is not bytes or not labels: + raise GuardedCustodyError("label release returned a zero-byte or invalid input") + return GuardedLabelRelease(receipt, labels) + + def _authorize(self, authority: AdmissionSpendAuthority) -> AdmissionPreparation: + try: + return authorize_admission( + authority.manifest, + authority.denominator, + authority.readiness, + authority.context, + self, + authority.resolver, + ) + except AdmissionSpendError as exc: + raise GuardedCustodyError(f"admission preparation refused: {exc}") from exc + + def _validate_spend( + self, + authority: AdmissionSpendAuthority, + preparation: AdmissionPreparation, + receipt: PopulationUseReceipt, + ) -> None: + try: + validate_pre_label_spend( + authority.manifest, + authority.denominator, + authority.readiness, + authority.context, + preparation, + receipt, + self, + authority.resolver, + ) + except AdmissionSpendError as exc: + raise GuardedCustodyError(f"committed admission spend refused: {exc}") from exc + + def release_cases( + self, + authority: AdmissionSpendAuthority, + preparation: AdmissionPreparation, + receipt: PopulationUseReceipt, + cases: Sequence[AdmissionCaseResult], + review_release: ReleaseReview, + ) -> AdmissionCaseRelease: + selected = tuple(cases) + release = _build_release( + authority.manifest, + preparation, + receipt, + selected, + review_release, + self.custodian_identity, + ) + return validate_admission_case_release( + authority.manifest, + authority.denominator, + authority.readiness, + authority.context, + preparation, + receipt, + release, + selected, + self, + authority.resolver, + ) + + +def _binding(case: AdmissionCaseResult) -> ReleasedCaseBinding: + evidence = tuple(sorted(item.input_digest for item in case.evidence_inputs)) + changes = tuple(sorted({item.record.packet.provenance.change_input_digest for item in case.evidence_inputs})) + draft = ReleasedCaseBinding( + case.case_id_digest, + case.scope, + case.role, + case.pair_id_digest, + case.result_digest, + evidence, + changes, + case.input_manifest.manifest_digest, + "0" * 64, + ) + return replace(draft, binding_digest=derive_released_case_binding_digest(draft)) + + +def _build_release( + manifest: AdmissionManifestV2, + preparation: AdmissionPreparation, + receipt: PopulationUseReceipt, + cases: tuple[AdmissionCaseResult, ...], + review_release: ReleaseReview, + custodian_identity: str, +) -> AdmissionCaseRelease: + bindings = tuple(sorted((_binding(case) for case in cases), key=lambda item: item.case_id_digest)) + draft = AdmissionCaseRelease( + ADMISSION_CASE_RELEASE_SCHEMA, + manifest.identity.population_digest, + manifest.manifest_digest, + preparation.authorization.authorization_digest, + preparation.preparation_digest, + receipt.receipt_digest, + custodian_identity, + bindings, + "0" * 64, + manifest.review, + "0" * 64, + ) + payload = derive_admission_case_release_payload_digest(draft) + review = review_release(payload, custodian_identity) + if type(review) is not IndependenceAttestation: + raise GuardedCustodyError("release reviewer returned an invalid attestation") + reviewed = replace(draft, release_payload_digest=payload, review=review) + return replace(reviewed, release_digest=derive_admission_case_release_digest(reviewed)) + + +__all__ = [ + "AdmissionSpendAuthority", + "CompareAndSwapCustodyStore", + "GuardedCustodyError", + "GuardedLabelRelease", + "GuardedPopulationCustodian", + "ReleaseReview", +] diff --git a/tests/test_detection_core_admission_adapter.py b/tests/test_detection_core_admission_adapter.py new file mode 100644 index 0000000..2d23e54 --- /dev/null +++ b/tests/test_detection_core_admission_adapter.py @@ -0,0 +1,302 @@ +"""Benchmark-owned guarded admission integration for detection-core Task 9.6.""" + +from __future__ import annotations + +import sys +from concurrent.futures import ThreadPoolExecutor +from dataclasses import replace +from pathlib import Path +from threading import Lock + +import pytest +from test_core_admission_metrics import _setup + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from benchmarks.valid.admission_run import V2AdmissionRunInput, run_v2_admission +from benchmarks.valid.admission_scoring import score_v2_admission +from benchmarks.valid.population_custody import ( + AdmissionSpendAuthority, + GuardedCustodyError, + GuardedPopulationCustodian, + ReleaseReview, +) +from openultrasast.core.admission_case import AdmissionCaseResult, derive_admission_case_result_digest +from openultrasast.core.admission_report import AdmissionDecision +from openultrasast.core.admission_spend import AdmissionPreparation, authorize_admission +from openultrasast.core.canonical import content_digest +from openultrasast.core.custody import LineagePurpose, PopulationCustodyLog, PopulationStatus +from openultrasast.core.factpack import IndependenceAttestation, derive_independence_attestation_digest + + +class SimulatedCrash(RuntimeError): + pass + + +class _Store: + def __init__(self, initial: PopulationCustodyLog, *, fail_before_append: bool = False) -> None: + self.log = initial + self.fail_before_append = fail_before_append + self.lock = Lock() + + def current_log(self, population_digest: str) -> PopulationCustodyLog: + if population_digest != self.log.population_digest: + raise LookupError("population not found") + return self.log + + def compare_and_swap(self, expected_log_digest: str, replacement: PopulationCustodyLog) -> bool: + with self.lock: + if self.fail_before_append: + raise SimulatedCrash("before durable append") + if self.log.log_digest != expected_log_digest: + return False + self.log = replacement + return True + + +def _custodian(setup: object, store: _Store) -> GuardedPopulationCustodian: + population = setup.manifest.identity.population_digest # type: ignore[attr-defined] + current = setup.custody.current_log(population) # type: ignore[attr-defined] + return GuardedPopulationCustodian( + population, + current.custodian_identity, + setup.context.evidence.custody_policy, # type: ignore[attr-defined] + store, + setup.custody, # type: ignore[attr-defined] + ) + + +def _label_file(tmp_path: Path) -> Path: + path = tmp_path / "labels.json" + path.write_bytes(b'{"labels":["vulnerable","fixed","ordinary"]}') + assert path.stat().st_size > 0 + return path + + +def _reviewer(setup: object) -> ReleaseReview: + def review(payload: str, producer: str) -> IndependenceAttestation: + draft = IndependenceAttestation( + payload, + producer, + setup.manifest.review.reviewer_identity, # type: ignore[attr-defined] + setup.manifest.review.conflict_disclosure, # type: ignore[attr-defined] + "0" * 64, + ) + return replace(draft, attestation_digest=derive_independence_attestation_digest(draft)) + + return review + + +def _authority(setup: object) -> AdmissionSpendAuthority: + return AdmissionSpendAuthority( + setup.manifest, # type: ignore[attr-defined] + setup.denominator, # type: ignore[attr-defined] + setup.readiness, # type: ignore[attr-defined] + setup.context, # type: ignore[attr-defined] + setup.resolver, # type: ignore[attr-defined] + ) + + +def _request(setup: object, custodian: GuardedPopulationCustodian, path: Path) -> V2AdmissionRunInput: + def analyze(labels: bytes) -> tuple[AdmissionCaseResult, ...]: + assert labels == path.read_bytes() + return setup.cases # type: ignore[attr-defined,no-any-return] + + return V2AdmissionRunInput( + _authority(setup), + custodian, + path.read_bytes, + analyze, + _reviewer(setup), + ) + + +def _prepare(setup: object, custodian: GuardedPopulationCustodian) -> AdmissionPreparation: + authority = _authority(setup) + return authorize_admission( + authority.manifest, + authority.denominator, + authority.readiness, + authority.context, + custodian, + authority.resolver, + ) + + +def test_synthetic_admission_runs_end_to_end_after_nonzero_label_read(tmp_path: Path) -> None: + setup = _setup() + population = setup.manifest.identity.population_digest + store = _Store(setup.custody.current_log(population)) + custodian = _custodian(setup, store) + + replay = run_v2_admission(_request(setup, custodian, _label_file(tmp_path))) + + assert replay.report.decision in (AdmissionDecision.GO, AdmissionDecision.NO_GO) + assert replay.label_release_receipt.status is PopulationStatus.SPENT + assert replay.case_release.review.producer_identity == custodian.custodian_identity + assert replay.case_release.review.reviewer_identity != custodian.custodian_identity + assert store.log.status is PopulationStatus.SPENT + + +def test_crash_before_append_releases_no_labels_and_preserves_available(tmp_path: Path) -> None: + setup = _setup() + population = setup.manifest.identity.population_digest + store = _Store(setup.custody.current_log(population), fail_before_append=True) + custodian = _custodian(setup, store) + preparation = _prepare(setup, custodian) + reads = 0 + + def labels() -> bytes: + nonlocal reads + reads += 1 + return _label_file(tmp_path).read_bytes() + + with pytest.raises(SimulatedCrash, match="before durable append"): + custodian.spend_and_release_labels(_authority(setup), preparation, labels) + assert reads == 0 + assert store.log.status is PopulationStatus.AVAILABLE + + +@pytest.mark.parametrize("mutation", ("preparation", "authorization", "purpose", "influenced")) +def test_invalid_preparation_never_spends_or_releases_labels(mutation: str) -> None: + setup = _setup() + population = setup.manifest.identity.population_digest + store = _Store(setup.custody.current_log(population)) + custodian = _custodian(setup, store) + preparation = _prepare(setup, custodian) + if mutation == "preparation": + changed = replace(preparation, preparation_digest="0" * 64) + elif mutation == "authorization": + authorization = replace(preparation.authorization, authorization_digest="0" * 64) + changed = replace(preparation, authorization=authorization) + elif mutation == "purpose": + request = replace(preparation.label_release_request, purpose=LineagePurpose.PROMOTION) + changed = replace(preparation, label_release_request=request) + else: + request = replace(preparation.label_release_request, influenced_artifact_digest=content_digest("substituted")) + changed = replace(preparation, label_release_request=request) + reads = 0 + + def labels() -> bytes: + nonlocal reads + reads += 1 + return b"escaped-labels" + + with pytest.raises(GuardedCustodyError): + custodian.spend_and_release_labels(_authority(setup), changed, labels) + assert reads == 0 + assert store.log.status is PopulationStatus.AVAILABLE + + +def test_crash_after_append_leaves_population_spent(tmp_path: Path) -> None: + setup = _setup() + population = setup.manifest.identity.population_digest + store = _Store(setup.custody.current_log(population)) + custodian = _custodian(setup, store) + preparation = _prepare(setup, custodian) + + def crash() -> bytes: + _label_file(tmp_path).read_bytes() + raise SimulatedCrash("after durable append") + + with pytest.raises(SimulatedCrash, match="after durable append"): + custodian.spend_and_release_labels(_authority(setup), preparation, crash) + assert store.log.status is PopulationStatus.SPENT + assert store.log.receipt_count > 0 + + +def test_concurrent_preparations_release_labels_to_one_cas_winner(tmp_path: Path) -> None: + setup = _setup() + population = setup.manifest.identity.population_digest + store = _Store(setup.custody.current_log(population)) + custodian = _custodian(setup, store) + preparation = _prepare(setup, custodian) + path = _label_file(tmp_path) + reads = 0 + lock = Lock() + + def labels() -> bytes: + nonlocal reads + with lock: + reads += 1 + return path.read_bytes() + + def attempt() -> object: + try: + return custodian.spend_and_release_labels(_authority(setup), preparation, labels) + except GuardedCustodyError as exc: + return exc + + with ThreadPoolExecutor(max_workers=2) as pool: + results = tuple(pool.map(lambda _: attempt(), range(2))) + assert sum(not isinstance(item, GuardedCustodyError) for item in results) == 1 + assert reads == 1 + assert store.log.status is PopulationStatus.SPENT + + +def test_swapped_operational_payload_is_refused_before_metrics(tmp_path: Path) -> None: + setup = _setup() + population = setup.manifest.identity.population_digest + store = _Store(setup.custody.current_log(population)) + custodian = _custodian(setup, store) + preparation = _prepare(setup, custodian) + authority = _authority(setup) + guarded = custodian.spend_and_release_labels(authority, preparation, _label_file(tmp_path).read_bytes) + release = custodian.release_cases( + authority, + preparation, + guarded.receipt, + setup.cases, + _reviewer(setup), + ) + changed = replace(setup.cases[0], total_reply_ms=setup.cases[0].total_reply_ms + 1) + changed = replace(changed, result_digest=derive_admission_case_result_digest(changed)) + + with pytest.raises(ValueError, match="release|binding|case"): + score_v2_admission( + setup.manifest, + setup.denominator, + setup.readiness, + preparation, + guarded.receipt, + release, + (changed, *setup.cases[1:]), + setup.context, + custodian, + setup.resolver, + ) + + +def test_case_release_requires_an_independent_attestation(tmp_path: Path) -> None: + setup = _setup() + population = setup.manifest.identity.population_digest + store = _Store(setup.custody.current_log(population)) + custodian = _custodian(setup, store) + preparation = _prepare(setup, custodian) + authority = _authority(setup) + guarded = custodian.spend_and_release_labels(authority, preparation, _label_file(tmp_path).read_bytes) + + def self_review(payload: str, producer: str) -> IndependenceAttestation: + draft = IndependenceAttestation(payload, producer, producer, setup.manifest.review.conflict_disclosure, "0" * 64) + return replace(draft, attestation_digest=derive_independence_attestation_digest(draft)) + + with pytest.raises(ValueError, match="independent"): + custodian.release_cases( + authority, + preparation, + guarded.receipt, + setup.cases, + self_review, + ) + + +def test_zero_byte_label_input_fails_after_monotonic_spend() -> None: + setup = _setup() + population = setup.manifest.identity.population_digest + store = _Store(setup.custody.current_log(population)) + custodian = _custodian(setup, store) + preparation = _prepare(setup, custodian) + + with pytest.raises(GuardedCustodyError, match="zero-byte"): + custodian.spend_and_release_labels(_authority(setup), preparation, lambda: b"") + assert store.log.status is PopulationStatus.SPENT