From edd8a7ec87fe0eb48d42dea8f59f7e63288c3f17 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sat, 31 Oct 2026 14:08:35 +0100 Subject: [PATCH] feat(detection-core): enforce append-only population custody --- .kiro/specs/detection-core/tasks.md | 2 +- src/openultrasast/core/custody.py | 363 +++++++++++++++++++++++++++ src/openultrasast/core/provenance.py | 40 +++ tests/detection_core_fixtures.py | 40 ++- tests/test_core_custody.py | 211 ++++++++++++++++ 5 files changed, 654 insertions(+), 2 deletions(-) create mode 100644 src/openultrasast/core/custody.py create mode 100644 tests/test_core_custody.py diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index 3da594f..8bb4beb 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -50,7 +50,7 @@ - _Requirements: 10.1, 10.4_ - _Boundary: Provenance and Lineage_ - _Depends: 1.2_ -- [ ] 2.3 Enforce append-only population custody +- [x] 2.3 Enforce append-only population custody - Construct and validate contiguous reviewed-custodian receipt chains against the complete current log. - Require compare-and-swap requests to name the expected log head and derive the next sequence and predecessor. - Make spent status monotonic and reject old prefixes, omitted receipts, broken chains, and later availability claims. diff --git a/src/openultrasast/core/custody.py b/src/openultrasast/core/custody.py new file mode 100644 index 0000000..4b2118b --- /dev/null +++ b/src/openultrasast/core/custody.py @@ -0,0 +1,363 @@ +"""Pure append-only population custody contracts.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, replace +from enum import Enum +from typing import Protocol + +from openultrasast.core import canonical + +CUSTODY_LOG_SCHEMA = "openultrasast.detection-core.population-custody-log.v2" +CUSTODY_POLICY_SCHEMA = "openultrasast.detection-core.population-custody-policy.v2" +CUSTODY_REQUEST_SCHEMA = "openultrasast.detection-core.population-use-request.v2" +CUSTODY_RECEIPT_SCHEMA = "openultrasast.detection-core.population-use-receipt.v2" +_DIGEST = re.compile(r"[0-9a-f]{64}") + + +def _require_digest(value: object, field_name: str) -> None: + if not isinstance(value, str) or _DIGEST.fullmatch(value) is None: + raise canonical.DigestFormatError(f"{field_name} must be a full lowercase SHA-256 digest") + + +def _require_text(value: object, field_name: str) -> None: + if not isinstance(value, str) or not value: + raise ValueError(f"{field_name} must be nonempty text") + + +class PopulationCustodyError(ValueError): + """A population custody chain or compare-and-swap request is invalid.""" + + +class LineagePurpose(Enum): + PROMPT = "prompt" + EXAMPLE = "example" + THRESHOLD = "threshold" + FITTED_MODEL = "fitted_model" + SEMANTIC_FACT = "semantic_fact" + MATCHING_BEHAVIOR = "matching_behavior" + DECISION_MAPPING = "decision_mapping" + PROMOTION = "promotion" + ADMISSION_LABEL_OPEN = "admission_label_open" + + +class PopulationStatus(Enum): + AVAILABLE = "available" + SPENT = "spent" + + +@dataclass(frozen=True) +class PopulationUseRequest: + population_digest: str + arm_id: str + purpose: LineagePurpose + labels_accessed: bool + influenced_artifact_digest: str | None + expected_log_digest: str + + def __post_init__(self) -> None: + _require_digest(self.population_digest, "population_digest") + _require_text(self.arm_id, "arm_id") + if not isinstance(self.purpose, LineagePurpose): + raise TypeError("purpose must be a LineagePurpose") + if not isinstance(self.labels_accessed, bool): + raise TypeError("labels_accessed must be boolean") + if self.influenced_artifact_digest is not None: + _require_digest(self.influenced_artifact_digest, "influenced_artifact_digest") + _require_digest(self.expected_log_digest, "expected_log_digest") + + +@dataclass(frozen=True) +class PopulationUseReceipt: + request: PopulationUseRequest + request_digest: str + population_digest: str + custodian_identity: str + sequence: int + predecessor_digest: str | None + prior_status: PopulationStatus + status: PopulationStatus + reasons: tuple[str, ...] + receipt_digest: str + + def __post_init__(self) -> None: + if not isinstance(self.request, PopulationUseRequest): + raise TypeError("request must be a PopulationUseRequest") + for name in ("request_digest", "population_digest", "custodian_identity", "receipt_digest"): + _require_digest(getattr(self, name), name) + if self.predecessor_digest is not None: + _require_digest(self.predecessor_digest, "predecessor_digest") + if isinstance(self.sequence, bool) or not isinstance(self.sequence, int) or self.sequence <= 0: + raise ValueError("sequence must be a positive integer") + if not isinstance(self.prior_status, PopulationStatus) or not isinstance(self.status, PopulationStatus): + raise TypeError("receipt status must be a PopulationStatus") + if not isinstance(self.reasons, tuple) or not all(isinstance(reason, str) and reason for reason in self.reasons): + raise TypeError("reasons must be a tuple of nonempty text") + + +@dataclass(frozen=True) +class PopulationCustodyPolicy: + reviewed_custodian_digests: tuple[str, ...] + policy_digest: str + + def __post_init__(self) -> None: + if not isinstance(self.reviewed_custodian_digests, tuple) or not self.reviewed_custodian_digests: + raise ValueError("reviewed_custodian_digests must be a nonempty tuple") + for digest in self.reviewed_custodian_digests: + _require_digest(digest, "reviewed_custodian_digest") + if len(set(self.reviewed_custodian_digests)) != len(self.reviewed_custodian_digests): + raise ValueError("reviewed_custodian_digests must be unique") + _require_digest(self.policy_digest, "policy_digest") + + +@dataclass(frozen=True) +class PopulationCustodyLog: + population_digest: str + custodian_identity: str + receipts: tuple[PopulationUseReceipt, ...] + receipt_count: int + head_receipt_digest: str | None + status: PopulationStatus + log_digest: str + + def __post_init__(self) -> None: + _require_digest(self.population_digest, "population_digest") + _require_digest(self.custodian_identity, "custodian_identity") + if not isinstance(self.receipts, tuple) or not all(isinstance(item, PopulationUseReceipt) for item in self.receipts): + raise TypeError("receipts must be a tuple of PopulationUseReceipt values") + if isinstance(self.receipt_count, bool) or not isinstance(self.receipt_count, int) or self.receipt_count < 0: + raise ValueError("receipt_count must be a nonnegative integer") + if self.head_receipt_digest is not None: + _require_digest(self.head_receipt_digest, "head_receipt_digest") + if not isinstance(self.status, PopulationStatus): + raise TypeError("status must be a PopulationStatus") + _require_digest(self.log_digest, "log_digest") + + +class PopulationCustodyPort(Protocol): + def current_log(self, population_digest: str) -> PopulationCustodyLog: ... + + +def _request_value(request: PopulationUseRequest) -> canonical.JsonValue: + return { + "schema_version": CUSTODY_REQUEST_SCHEMA, + "population_digest": request.population_digest, + "arm_id": request.arm_id, + "purpose": request.purpose.value, + "labels_accessed": request.labels_accessed, + "influenced_artifact_digest": request.influenced_artifact_digest, + "expected_log_digest": request.expected_log_digest, + } + + +def derive_population_use_request_digest(request: PopulationUseRequest) -> str: + return canonical.content_digest(_request_value(request)) + + +def _receipt_value(receipt: PopulationUseReceipt) -> canonical.JsonValue: + return { + "schema_version": CUSTODY_RECEIPT_SCHEMA, + "request": _request_value(receipt.request), + "request_digest": receipt.request_digest, + "population_digest": receipt.population_digest, + "custodian_identity": receipt.custodian_identity, + "sequence": receipt.sequence, + "predecessor_digest": receipt.predecessor_digest, + "prior_status": receipt.prior_status.value, + "status": receipt.status.value, + "reasons": receipt.reasons, + } + + +def derive_population_use_receipt_digest(receipt: PopulationUseReceipt) -> str: + return canonical.content_digest(_receipt_value(receipt)) + + +def derive_population_custody_policy_digest(reviewed_custodian_digests: tuple[str, ...]) -> str: + return canonical.content_digest( + {"schema_version": CUSTODY_POLICY_SCHEMA, "reviewed_custodian_digests": tuple(sorted(reviewed_custodian_digests))} + ) + + +def _log_digest(population: str, custodian: str, receipts: tuple[PopulationUseReceipt, ...], status: PopulationStatus) -> str: + return canonical.content_digest( + { + "schema_version": CUSTODY_LOG_SCHEMA, + "population_digest": population, + "custodian_identity": custodian, + "receipt_digests": tuple(receipt.receipt_digest for receipt in receipts), + "receipt_count": len(receipts), + "head_receipt_digest": receipts[-1].receipt_digest if receipts else None, + "status": status.value, + } + ) + + +def empty_population_custody_log(population_digest: str, custodian_identity: str) -> PopulationCustodyLog: + return PopulationCustodyLog( + population_digest, + custodian_identity, + (), + 0, + None, + PopulationStatus.AVAILABLE, + _log_digest(population_digest, custodian_identity, (), PopulationStatus.AVAILABLE), + ) + + +def _validate_custody_policy(policy: PopulationCustodyPolicy) -> None: + expected = derive_population_custody_policy_digest(policy.reviewed_custodian_digests) + if policy.policy_digest != expected: + raise PopulationCustodyError("population custody policy digest cannot be re-derived") + + +def _validate_receipt_link( + log: PopulationCustodyLog, + receipt: PopulationUseReceipt, + sequence: int, + prefix: tuple[PopulationUseReceipt, ...], + status: PopulationStatus, +) -> None: + predecessor = prefix[-1].receipt_digest if prefix else None + if receipt.sequence != sequence: + raise PopulationCustodyError("receipt sequence is not contiguous") + if receipt.predecessor_digest != predecessor: + raise PopulationCustodyError("receipt predecessor does not match complete chain") + if receipt.population_digest != log.population_digest or receipt.request.population_digest != log.population_digest: + raise PopulationCustodyError("receipt population does not match log") + if receipt.custodian_identity != log.custodian_identity: + raise PopulationCustodyError("receipt custodian does not match log") + expected = _log_digest(log.population_digest, log.custodian_identity, prefix, status) + if receipt.request.expected_log_digest != expected: + raise PopulationCustodyError("receipt expected log does not match prefix") + + +def _validate_receipt_identity(receipt: PopulationUseReceipt, status: PopulationStatus) -> PopulationStatus: + expected_status = PopulationStatus.SPENT if status is PopulationStatus.SPENT or receipt.request.labels_accessed else status + if receipt.prior_status is not status or receipt.status is not expected_status: + raise PopulationCustodyError("receipt status is not monotonic") + expected_reasons = ("labels_accessed",) if receipt.request.labels_accessed else () + if status is PopulationStatus.SPENT and not receipt.request.labels_accessed: + expected_reasons = ("population_already_spent",) + if receipt.reasons != expected_reasons: + raise PopulationCustodyError("receipt reasons do not match status transition") + if receipt.request_digest != derive_population_use_request_digest(receipt.request): + raise PopulationCustodyError("request digest cannot be re-derived") + if receipt.receipt_digest != derive_population_use_receipt_digest(receipt): + raise PopulationCustodyError("receipt digest cannot be re-derived") + return expected_status + + +def _validate_log_metadata(log: PopulationCustodyLog, status: PopulationStatus) -> None: + if log.receipt_count != len(log.receipts): + raise PopulationCustodyError("receipt count does not match complete chain") + head = log.receipts[-1].receipt_digest if log.receipts else None + if log.head_receipt_digest != head: + raise PopulationCustodyError("head receipt does not match complete chain") + if log.status is not status: + raise PopulationCustodyError("log status does not match receipt chain") + if log.log_digest != _log_digest(log.population_digest, log.custodian_identity, log.receipts, status): + raise PopulationCustodyError("log digest cannot be re-derived") + + +def validate_population_custody_log( + log: PopulationCustodyLog, + policy: PopulationCustodyPolicy, + *, + current: PopulationCustodyPort | None = None, +) -> None: + _validate_custody_policy(policy) + if log.custodian_identity not in policy.reviewed_custodian_digests: + raise PopulationCustodyError("population custodian is not reviewed") + status = PopulationStatus.AVAILABLE + prefix: tuple[PopulationUseReceipt, ...] = () + for sequence, receipt in enumerate(log.receipts, 1): + _validate_receipt_link(log, receipt, sequence, prefix, status) + status = _validate_receipt_identity(receipt, status) + prefix = (*prefix, receipt) + _validate_log_metadata(log, status) + if current is not None: + current_log = current.current_log(log.population_digest) + validate_population_custody_log(current_log, policy) + if current_log.log_digest != log.log_digest: + raise PopulationCustodyError("candidate is not the current complete log") + + +def record_population_use( + request: PopulationUseRequest, + prior: PopulationCustodyLog, + custodian_identity: str, + policy: PopulationCustodyPolicy, +) -> PopulationUseReceipt: + validate_population_custody_log(prior, policy) + if custodian_identity not in policy.reviewed_custodian_digests: + raise PopulationCustodyError("population custodian is not reviewed") + if custodian_identity != prior.custodian_identity: + raise PopulationCustodyError("population custodian does not match log") + if request.population_digest != prior.population_digest: + raise PopulationCustodyError("request population does not match log") + if request.expected_log_digest != prior.log_digest: + raise PopulationCustodyError("request expected log does not match current head") + status = PopulationStatus.SPENT if prior.status is PopulationStatus.SPENT or request.labels_accessed else prior.status + reasons: tuple[str, ...] + if request.labels_accessed: + reasons = ("labels_accessed",) + else: + reasons = ("population_already_spent",) if prior.status is PopulationStatus.SPENT else () + provisional = PopulationUseReceipt( + request, + derive_population_use_request_digest(request), + prior.population_digest, + custodian_identity, + prior.receipt_count + 1, + prior.head_receipt_digest, + prior.status, + status, + reasons, + "0" * 64, + ) + return replace(provisional, receipt_digest=derive_population_use_receipt_digest(provisional)) + + +def append_population_use( + prior: PopulationCustodyLog, receipt: PopulationUseReceipt, policy: PopulationCustodyPolicy +) -> PopulationCustodyLog: + validate_population_custody_log(prior, policy) + if receipt.request.expected_log_digest != prior.log_digest: + raise PopulationCustodyError("receipt expected log does not match prior log") + receipts = (*prior.receipts, receipt) + provisional = PopulationCustodyLog( + prior.population_digest, + prior.custodian_identity, + receipts, + len(receipts), + receipt.receipt_digest, + receipt.status, + "0" * 64, + ) + result = replace( + provisional, + log_digest=_log_digest(provisional.population_digest, provisional.custodian_identity, receipts, provisional.status), + ) + validate_population_custody_log(result, policy) + return result + + +__all__ = [ + "LineagePurpose", + "PopulationCustodyError", + "PopulationCustodyLog", + "PopulationCustodyPolicy", + "PopulationCustodyPort", + "PopulationStatus", + "PopulationUseReceipt", + "PopulationUseRequest", + "append_population_use", + "derive_population_custody_policy_digest", + "derive_population_use_receipt_digest", + "derive_population_use_request_digest", + "empty_population_custody_log", + "record_population_use", + "validate_population_custody_log", +] diff --git a/src/openultrasast/core/provenance.py b/src/openultrasast/core/provenance.py index 8b4a472..01eb85e 100644 --- a/src/openultrasast/core/provenance.py +++ b/src/openultrasast/core/provenance.py @@ -11,6 +11,27 @@ from enum import Enum from openultrasast.core import canonical from openultrasast.core.canonical import ContentRef, ContentResolver +from openultrasast.core.custody import ( + CUSTODY_LOG_SCHEMA, + CUSTODY_POLICY_SCHEMA, + CUSTODY_RECEIPT_SCHEMA, + CUSTODY_REQUEST_SCHEMA, + LineagePurpose, + PopulationCustodyError, + PopulationCustodyLog, + PopulationCustodyPolicy, + PopulationCustodyPort, + PopulationStatus, + PopulationUseReceipt, + PopulationUseRequest, + append_population_use, + derive_population_custody_policy_digest, + derive_population_use_receipt_digest, + derive_population_use_request_digest, + empty_population_custody_log, + record_population_use, + validate_population_custody_log, +) NAMED_INPUTS_SCHEMA = "openultrasast.detection-core.named-inputs.v2" PROVENANCE_POLICY_SCHEMA = "openultrasast.detection-core.provenance-policy.v2" @@ -291,6 +312,25 @@ def validate_provenance( __all__ = [ + "CUSTODY_LOG_SCHEMA", + "CUSTODY_POLICY_SCHEMA", + "CUSTODY_RECEIPT_SCHEMA", + "CUSTODY_REQUEST_SCHEMA", + "LineagePurpose", + "PopulationCustodyError", + "PopulationCustodyLog", + "PopulationCustodyPolicy", + "PopulationCustodyPort", + "PopulationStatus", + "PopulationUseReceipt", + "PopulationUseRequest", + "append_population_use", + "derive_population_custody_policy_digest", + "derive_population_use_receipt_digest", + "derive_population_use_request_digest", + "empty_population_custody_log", + "record_population_use", + "validate_population_custody_log", "DecisionProvenance", "ModelCallIdentity", "NAMED_INPUTS_SCHEMA", diff --git a/tests/detection_core_fixtures.py b/tests/detection_core_fixtures.py index ceb3956..62eea82 100644 --- a/tests/detection_core_fixtures.py +++ b/tests/detection_core_fixtures.py @@ -4,9 +4,10 @@ from __future__ import annotations from collections.abc import Mapping from dataclasses import dataclass +from threading import Lock from types import MappingProxyType -from openultrasast.core import canonical +from openultrasast.core import canonical, provenance @dataclass(frozen=True) @@ -59,3 +60,40 @@ def stored_artifact( ) resolver = InMemoryContentResolver({artifact_id: canonical.ResolvedContent(content=content, receipt=receipt)}) return reference, resolver + + +class InMemoryPopulationCustody: + """Deterministic compare-and-swap fixture; production persistence stays external.""" + + def __init__(self, initial: provenance.PopulationCustodyLog, policy: provenance.PopulationCustodyPolicy) -> None: + provenance.validate_population_custody_log(initial, policy) + self._log = initial + self._policy = policy + self._lock = Lock() + + def current_log(self, population_digest: str) -> provenance.PopulationCustodyLog: + if population_digest != self._log.population_digest: + raise provenance.PopulationCustodyError("population custody log not found") + return self._log + + def append(self, request: provenance.PopulationUseRequest, custodian: str) -> provenance.PopulationUseReceipt: + with self._lock: + receipt = provenance.record_population_use(request, self._log, custodian, self._policy) + self._log = provenance.append_population_use(self._log, receipt, self._policy) + return receipt + + +def custody_contract() -> tuple[ + provenance.PopulationCustodyLog, + provenance.PopulationCustodyPolicy, + str, +]: + """Create one reviewed custodian and an empty reproducible population log.""" + population = canonical.content_digest("population") + custodian = canonical.content_digest("reviewed-custodian") + reviewed = (custodian,) + policy = provenance.PopulationCustodyPolicy( + reviewed_custodian_digests=reviewed, + policy_digest=provenance.derive_population_custody_policy_digest(reviewed), + ) + return provenance.empty_population_custody_log(population, custodian), policy, custodian diff --git a/tests/test_core_custody.py b/tests/test_core_custody.py new file mode 100644 index 0000000..15f8e35 --- /dev/null +++ b/tests/test_core_custody.py @@ -0,0 +1,211 @@ +"""Append-only population custody contracts (Req 12.2, 13.1, 13.4).""" + +from __future__ import annotations + +from concurrent.futures import ThreadPoolExecutor +from dataclasses import replace + +import detection_core_fixtures as fixtures +import pytest + +from openultrasast.core import canonical, provenance + + +def _request( + log: provenance.PopulationCustodyLog, + *, + labels_accessed: bool = False, + purpose: provenance.LineagePurpose | None = None, +) -> provenance.PopulationUseRequest: + return provenance.PopulationUseRequest( + population_digest=log.population_digest, + arm_id="arm-v2", + purpose=purpose or provenance.LineagePurpose.SEMANTIC_FACT, + labels_accessed=labels_accessed, + influenced_artifact_digest=canonical.content_digest("artifact"), + expected_log_digest=log.log_digest, + ) + + +def test_empty_log_and_policy_have_deterministic_identities() -> None: + first, policy, _ = fixtures.custody_contract() + second, _, _ = fixtures.custody_contract() + assert first == second + assert first.receipts == () + assert first.receipt_count == 0 + assert first.head_receipt_digest is None + assert first.status is provenance.PopulationStatus.AVAILABLE + provenance.validate_population_custody_log(first, policy) + + +def test_record_derives_cas_sequence_predecessor_and_receipt_identity() -> None: + prior, policy, custodian = fixtures.custody_contract() + receipt = provenance.record_population_use(_request(prior), prior, custodian, policy) + assert receipt.sequence == 1 + assert receipt.predecessor_digest is None + assert receipt.prior_status is provenance.PopulationStatus.AVAILABLE + assert receipt.status is provenance.PopulationStatus.AVAILABLE + assert receipt.request_digest == provenance.derive_population_use_request_digest(receipt.request) + assert receipt.receipt_digest == provenance.derive_population_use_receipt_digest(receipt) + + +def test_append_constructs_and_validates_a_contiguous_complete_chain() -> None: + prior, policy, custodian = fixtures.custody_contract() + first = provenance.record_population_use(_request(prior), prior, custodian, policy) + one = provenance.append_population_use(prior, first, policy) + second = provenance.record_population_use(_request(one), one, custodian, policy) + complete = provenance.append_population_use(one, second, policy) + provenance.validate_population_custody_log(complete, policy) + assert complete.receipt_count == 2 + assert complete.head_receipt_digest == second.receipt_digest + assert second.predecessor_digest == first.receipt_digest + + +def test_compare_and_swap_refuses_an_old_log_head() -> None: + prior, policy, custodian = fixtures.custody_contract() + stale = _request(prior) + first = provenance.record_population_use(stale, prior, custodian, policy) + current = provenance.append_population_use(prior, first, policy) + with pytest.raises(provenance.PopulationCustodyError, match="expected log"): + provenance.record_population_use(stale, current, custodian, policy) + + +def test_label_access_spends_population_and_status_is_monotonic() -> None: + prior, policy, custodian = fixtures.custody_contract() + spent_receipt = provenance.record_population_use(_request(prior, labels_accessed=True), prior, custodian, policy) + spent = provenance.append_population_use(prior, spent_receipt, policy) + later = provenance.record_population_use(_request(spent), spent, custodian, policy) + complete = provenance.append_population_use(spent, later, policy) + assert spent_receipt.status is provenance.PopulationStatus.SPENT + assert later.prior_status is provenance.PopulationStatus.SPENT + assert later.status is provenance.PopulationStatus.SPENT + assert complete.status is provenance.PopulationStatus.SPENT + + +def test_broken_or_omitted_receipts_are_refused() -> None: + prior, policy, custodian = fixtures.custody_contract() + first = provenance.record_population_use(_request(prior), prior, custodian, policy) + one = provenance.append_population_use(prior, first, policy) + second = provenance.record_population_use(_request(one), one, custodian, policy) + complete = provenance.append_population_use(one, second, policy) + broken = replace(second, predecessor_digest=None) + with pytest.raises(provenance.PopulationCustodyError, match="predecessor"): + provenance.validate_population_custody_log(replace(complete, receipts=(first, broken)), policy) + with pytest.raises(provenance.PopulationCustodyError, match="sequence"): + provenance.validate_population_custody_log( + replace(complete, receipts=(second,), receipt_count=1, head_receipt_digest=second.receipt_digest), policy + ) + + +def test_log_metadata_and_receipt_identities_are_rederived() -> None: + prior, policy, custodian = fixtures.custody_contract() + receipt = provenance.record_population_use(_request(prior), prior, custodian, policy) + current = provenance.append_population_use(prior, receipt, policy) + for changed, message in ( + (replace(current, receipt_count=2), "count"), + (replace(current, head_receipt_digest=None), "head"), + (replace(current, log_digest="0" * 64), "log digest"), + (replace(current, status=provenance.PopulationStatus.SPENT), "status"), + ): + with pytest.raises(provenance.PopulationCustodyError, match=message): + provenance.validate_population_custody_log(changed, policy) + with pytest.raises(provenance.PopulationCustodyError, match="receipt digest"): + provenance.validate_population_custody_log(replace(current, receipts=(replace(receipt, receipt_digest="0" * 64),)), policy) + + +def test_unreviewed_custodian_and_tampered_policy_are_refused() -> None: + prior, policy, custodian = fixtures.custody_contract() + with pytest.raises(provenance.PopulationCustodyError, match="reviewed"): + provenance.record_population_use(_request(prior), prior, canonical.content_digest("other"), policy) + with pytest.raises(provenance.PopulationCustodyError, match="policy"): + provenance.record_population_use(_request(prior), prior, custodian, replace(policy, policy_digest="0" * 64)) + + +def test_reviewed_custodian_cannot_take_over_an_existing_log() -> None: + prior, _, custodian = fixtures.custody_contract() + other = canonical.content_digest("other-reviewed-custodian") + reviewed = (custodian, other) + policy = provenance.PopulationCustodyPolicy( + reviewed, + provenance.derive_population_custody_policy_digest(reviewed), + ) + with pytest.raises(provenance.PopulationCustodyError, match="does not match log"): + provenance.record_population_use(_request(prior), prior, other, policy) + + +def test_receipt_reasons_are_derived_from_the_status_transition() -> None: + prior, policy, custodian = fixtures.custody_contract() + receipt = provenance.record_population_use(_request(prior), prior, custodian, policy) + tampered = replace(receipt, reasons=("labels_accessed",), receipt_digest="0" * 64) + tampered = replace(tampered, receipt_digest=provenance.derive_population_use_receipt_digest(tampered)) + with pytest.raises(provenance.PopulationCustodyError, match="reasons"): + provenance.append_population_use(prior, tampered, policy) + + +def test_current_log_check_refuses_an_old_but_internally_valid_prefix() -> None: + prior, policy, custodian = fixtures.custody_contract() + port = fixtures.InMemoryPopulationCustody(prior, policy) + port.append(_request(prior), custodian) + provenance.validate_population_custody_log(port.current_log(prior.population_digest), policy, current=port) + with pytest.raises(provenance.PopulationCustodyError, match="current complete log"): + provenance.validate_population_custody_log(prior, policy, current=port) + + +def test_current_log_check_validates_the_ports_complete_chain() -> None: + prior, policy, _ = fixtures.custody_contract() + + class InvalidPort: + def current_log(self, population_digest: str) -> provenance.PopulationCustodyLog: + assert population_digest == prior.population_digest + return replace(prior, log_digest="0" * 64) + + with pytest.raises(provenance.PopulationCustodyError, match="log digest"): + provenance.validate_population_custody_log(prior, policy, current=InvalidPort()) + + +def test_concurrent_requests_have_exactly_one_cas_successor() -> None: + prior, policy, custodian = fixtures.custody_contract() + port = fixtures.InMemoryPopulationCustody(prior, policy) + left = _request(prior, labels_accessed=True, purpose=provenance.LineagePurpose.ADMISSION_LABEL_OPEN) + right = replace(left, arm_id="other-arm") + + def attempt( + request: provenance.PopulationUseRequest, + ) -> provenance.PopulationUseReceipt | provenance.PopulationCustodyError: + try: + return port.append(request, custodian) + except provenance.PopulationCustodyError as exc: + return exc + + with ThreadPoolExecutor(max_workers=2) as pool: + results = tuple(pool.map(attempt, (left, right))) + winners = tuple(result for result in results if isinstance(result, provenance.PopulationUseReceipt)) + failures = tuple(result for result in results if isinstance(result, provenance.PopulationCustodyError)) + assert len(winners) == len(failures) == 1 + assert "expected log" in str(failures[0]) + current = port.current_log(prior.population_digest) + assert current.receipts == winners + assert current.status is provenance.PopulationStatus.SPENT + + +def test_fixture_cannot_restore_a_spent_population() -> None: + prior, policy, custodian = fixtures.custody_contract() + port = fixtures.InMemoryPopulationCustody(prior, policy) + port.append(_request(prior, labels_accessed=True), custodian) + spent = port.current_log(prior.population_digest) + port.append(_request(spent), custodian) + current = port.current_log(prior.population_digest) + assert current.status is provenance.PopulationStatus.SPENT + restored = replace(current, status=provenance.PopulationStatus.AVAILABLE) + with pytest.raises(provenance.PopulationCustodyError, match="status"): + provenance.validate_population_custody_log(restored, policy) + + +@pytest.mark.parametrize("field", ["population_digest", "expected_log_digest"]) +def test_request_digest_fields_are_runtime_validated(field: str) -> None: + prior, _, _ = fixtures.custody_contract() + with pytest.raises(canonical.DigestFormatError): + if field == "population_digest": + replace(_request(prior), population_digest="short") + else: + replace(_request(prior), expected_log_digest="short")