From 62c87b39ab7afdd6c6c6dc6a5df4ef4d89125e02 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sat, 31 Oct 2026 14:13:39 +0100 Subject: [PATCH] feat(detection-core): bind lineage to exact influenced artifacts --- .kiro/specs/detection-core/tasks.md | 2 +- src/openultrasast/core/lineage.py | 273 +++++++++++++++++++++ src/openultrasast/core/provenance.py | 22 ++ tests/test_core_lineage.py | 341 +++++++++++++++++++++++++++ 4 files changed, 637 insertions(+), 1 deletion(-) create mode 100644 src/openultrasast/core/lineage.py create mode 100644 tests/test_core_lineage.py diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index 8bb4beb..563540d 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -57,7 +57,7 @@ - Provide deterministic in-memory custody-port and custodian fixtures for lineage and admission tests. - Done when concurrent requests have one valid successor and no fixture can restore a spent population. - _Requirements: 12.2, 13.1, 13.4_ -- [ ] 2.4 Bind lineage purposes to exact influenced artifacts +- [x] 2.4 Bind lineage purposes to exact influenced artifacts - Require each lineage use to name its purpose, influenced artifact, source population, label access, and current-chain receipt. - Enforce required and optional use counts for prompts, examples, thresholds, fitted models, facts, matching behavior, mappings, and promotion. - Distinguish fact ancestry that forbids all label access from development lineage that may use a different spent population. diff --git a/src/openultrasast/core/lineage.py b/src/openultrasast/core/lineage.py new file mode 100644 index 0000000..9b462e9 --- /dev/null +++ b/src/openultrasast/core/lineage.py @@ -0,0 +1,273 @@ +"""Pure lineage contracts bound to authoritative population custody chains.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, replace + +from openultrasast.core import canonical +from openultrasast.core.custody import ( + LineagePurpose, + PopulationCustodyPolicy, + PopulationCustodyPort, + PopulationStatus, + PopulationUseReceipt, + validate_population_custody_log, +) + +LINEAGE_MANIFEST_SCHEMA = "openultrasast.detection-core.lineage-manifest.v2" +LINEAGE_DECISION_SCHEMA = "openultrasast.detection-core.lineage-decision.v2" +_DIGEST = re.compile(r"[0-9a-f]{64}") + + +class LineageValidationError(ValueError): + """A lineage identity or custody join cannot be reproduced.""" + + +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") + + +def _require_count(value: object, field_name: str) -> None: + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + raise ValueError(f"{field_name} must be a nonnegative integer") + + +@dataclass(frozen=True) +class LineageUse: + purpose: LineagePurpose + source_population_digest: str + labels_accessed: bool + influenced_artifact_digest: str + population_use_receipt_digest: str + + def __post_init__(self) -> None: + if not isinstance(self.purpose, LineagePurpose): + raise TypeError("purpose must be a LineagePurpose") + _require_digest(self.source_population_digest, "source_population_digest") + if not isinstance(self.labels_accessed, bool): + raise TypeError("labels_accessed must be boolean") + _require_digest(self.influenced_artifact_digest, "influenced_artifact_digest") + _require_digest(self.population_use_receipt_digest, "population_use_receipt_digest") + + +@dataclass(frozen=True) +class LineageManifest: + arm_id: str + subject_digest: str + uses: tuple[LineageUse, ...] + manifest_digest: str + + def __post_init__(self) -> None: + _require_text(self.arm_id, "arm_id") + _require_digest(self.subject_digest, "subject_digest") + if not isinstance(self.uses, tuple) or not all(isinstance(use, LineageUse) for use in self.uses): + raise TypeError("uses must be a tuple of LineageUse values") + _require_digest(self.manifest_digest, "manifest_digest") + + +@dataclass(frozen=True) +class LineageExpectation: + purpose: LineagePurpose + influenced_artifact_digest: str + minimum_uses: int + maximum_uses: int | None + + def __post_init__(self) -> None: + if not isinstance(self.purpose, LineagePurpose): + raise TypeError("purpose must be a LineagePurpose") + _require_digest(self.influenced_artifact_digest, "influenced_artifact_digest") + _require_count(self.minimum_uses, "minimum_uses") + if self.maximum_uses is not None: + _require_count(self.maximum_uses, "maximum_uses") + if self.maximum_uses < self.minimum_uses: + raise ValueError("maximum_uses must be at least minimum_uses") + + +@dataclass(frozen=True) +class LineageDecision: + allowed: bool + spent_population_digests: tuple[str, ...] + reasons: tuple[str, ...] + decision_digest: str + + def __post_init__(self) -> None: + if not isinstance(self.allowed, bool): + raise TypeError("allowed must be boolean") + if not isinstance(self.spent_population_digests, tuple): + raise TypeError("spent_population_digests must be a tuple") + for digest in self.spent_population_digests: + _require_digest(digest, "spent_population_digest") + if tuple(sorted(set(self.spent_population_digests))) != self.spent_population_digests: + raise ValueError("spent_population_digests must be sorted and unique") + 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") + if tuple(sorted(set(self.reasons))) != self.reasons: + raise ValueError("reasons must be sorted and unique") + _require_digest(self.decision_digest, "decision_digest") + + +def _use_value(use: LineageUse) -> canonical.JsonValue: + return { + "purpose": use.purpose.value, + "source_population_digest": use.source_population_digest, + "labels_accessed": use.labels_accessed, + "influenced_artifact_digest": use.influenced_artifact_digest, + "population_use_receipt_digest": use.population_use_receipt_digest, + } + + +def derive_lineage_manifest_digest(arm_id: str, subject_digest: str, uses: tuple[LineageUse, ...]) -> str: + _require_text(arm_id, "arm_id") + _require_digest(subject_digest, "subject_digest") + return canonical.content_digest( + { + "schema_version": LINEAGE_MANIFEST_SCHEMA, + "arm_id": arm_id, + "subject_digest": subject_digest, + "uses": tuple(_use_value(use) for use in uses), + } + ) + + +def _decision_value(decision: LineageDecision) -> canonical.JsonValue: + return { + "schema_version": LINEAGE_DECISION_SCHEMA, + "allowed": decision.allowed, + "spent_population_digests": decision.spent_population_digests, + "reasons": decision.reasons, + } + + +def derive_lineage_decision_digest(decision: LineageDecision) -> str: + return canonical.content_digest(_decision_value(decision)) + + +def _validate_manifest(lineage: LineageManifest, expected_subject_digest: str) -> None: + _require_digest(expected_subject_digest, "expected_subject_digest") + expected = derive_lineage_manifest_digest(lineage.arm_id, lineage.subject_digest, lineage.uses) + if lineage.manifest_digest != expected: + raise LineageValidationError("lineage manifest digest cannot be re-derived") + if lineage.subject_digest != expected_subject_digest: + raise LineageValidationError("lineage subject does not match expected subject") + receipt_digests = tuple(use.population_use_receipt_digest for use in lineage.uses) + if len(set(receipt_digests)) != len(receipt_digests): + raise LineageValidationError("duplicate lineage receipt") + + +def _expectation_index( + expectations: tuple[LineageExpectation, ...], +) -> dict[tuple[LineagePurpose, str], LineageExpectation]: + if not isinstance(expectations, tuple) or not all(isinstance(item, LineageExpectation) for item in expectations): + raise TypeError("expectations must be a tuple of LineageExpectation values") + indexed: dict[tuple[LineagePurpose, str], LineageExpectation] = {} + for expectation in expectations: + key = (expectation.purpose, expectation.influenced_artifact_digest) + if key in indexed: + raise LineageValidationError("duplicate lineage expectation") + indexed[key] = expectation + return indexed + + +def _receipt_for_use( + use: LineageUse, + arm_id: str, + custody: PopulationCustodyPort, + policy: PopulationCustodyPolicy, +) -> PopulationUseReceipt: + log = custody.current_log(use.source_population_digest) + validate_population_custody_log(log, policy, current=custody) + matches = tuple(receipt for receipt in log.receipts if receipt.receipt_digest == use.population_use_receipt_digest) + if len(matches) != 1: + raise LineageValidationError("lineage receipt is absent from the current complete custody chain") + receipt = matches[0] + request = receipt.request + if request.arm_id != arm_id: + raise LineageValidationError("lineage arm does not match its current-chain receipt") + if ( + request.population_digest != use.source_population_digest + or request.purpose is not use.purpose + or request.labels_accessed is not use.labels_accessed + or request.influenced_artifact_digest != use.influenced_artifact_digest + ): + raise LineageValidationError("lineage use does not match its current-chain receipt") + if use.labels_accessed and receipt.status is not PopulationStatus.SPENT: + raise LineageValidationError("labelled lineage receipt did not spend its population") + return receipt + + +def _count_reasons( + uses: tuple[LineageUse, ...], + expected: dict[tuple[LineagePurpose, str], LineageExpectation], +) -> list[str]: + counts = {key: 0 for key in expected} + reasons: list[str] = [] + for use in uses: + key = (use.purpose, use.influenced_artifact_digest) + if key not in expected: + reasons.append(f"unexpected_use:{use.purpose.value}:{use.influenced_artifact_digest}") + else: + counts[key] += 1 + for key, expectation in expected.items(): + count = counts[key] + label = f"{key[0].value}:{key[1]}" + if count < expectation.minimum_uses: + reasons.append(f"missing_required_use:{label}") + if expectation.maximum_uses is not None and count > expectation.maximum_uses: + reasons.append(f"too_many_uses:{label}") + return reasons + + +def _decision(allowed: bool, spent: tuple[str, ...], reasons: tuple[str, ...]) -> LineageDecision: + provisional = LineageDecision(allowed, spent, reasons, "0" * 64) + return replace(provisional, decision_digest=derive_lineage_decision_digest(provisional)) + + +def validate_lineage( + lineage: LineageManifest, + expected_subject_digest: str, + expectations: tuple[LineageExpectation, ...], + protected_population_digest: str | None, + forbid_label_access: bool, + custody: PopulationCustodyPort, + policy: PopulationCustodyPolicy, +) -> LineageDecision: + """Bind every lineage use to an exact expected artifact and current receipt.""" + _validate_manifest(lineage, expected_subject_digest) + if protected_population_digest is not None: + _require_digest(protected_population_digest, "protected_population_digest") + if not isinstance(forbid_label_access, bool): + raise TypeError("forbid_label_access must be boolean") + expected = _expectation_index(expectations) + reasons = _count_reasons(lineage.uses, expected) + spent: set[str] = set() + for use in lineage.uses: + _receipt_for_use(use, lineage.arm_id, custody, policy) + if use.labels_accessed: + spent.add(use.source_population_digest) + if forbid_label_access: + reasons.append(f"label_access_forbidden:{use.source_population_digest}") + if use.source_population_digest == protected_population_digest: + reasons.append(f"protected_population_labels:{use.source_population_digest}") + normalized_reasons = tuple(sorted(set(reasons))) + return _decision(not normalized_reasons, tuple(sorted(spent)), normalized_reasons) + + +__all__ = [ + "LINEAGE_DECISION_SCHEMA", + "LINEAGE_MANIFEST_SCHEMA", + "LineageDecision", + "LineageExpectation", + "LineageManifest", + "LineageUse", + "LineageValidationError", + "derive_lineage_decision_digest", + "derive_lineage_manifest_digest", + "validate_lineage", +] diff --git a/src/openultrasast/core/provenance.py b/src/openultrasast/core/provenance.py index 01eb85e..4c5e297 100644 --- a/src/openultrasast/core/provenance.py +++ b/src/openultrasast/core/provenance.py @@ -32,6 +32,18 @@ from openultrasast.core.custody import ( record_population_use, validate_population_custody_log, ) +from openultrasast.core.lineage import ( + LINEAGE_DECISION_SCHEMA, + LINEAGE_MANIFEST_SCHEMA, + LineageDecision, + LineageExpectation, + LineageManifest, + LineageUse, + LineageValidationError, + derive_lineage_decision_digest, + derive_lineage_manifest_digest, + validate_lineage, +) NAMED_INPUTS_SCHEMA = "openultrasast.detection-core.named-inputs.v2" PROVENANCE_POLICY_SCHEMA = "openultrasast.detection-core.provenance-policy.v2" @@ -317,6 +329,13 @@ __all__ = [ "CUSTODY_RECEIPT_SCHEMA", "CUSTODY_REQUEST_SCHEMA", "LineagePurpose", + "LINEAGE_DECISION_SCHEMA", + "LINEAGE_MANIFEST_SCHEMA", + "LineageDecision", + "LineageExpectation", + "LineageManifest", + "LineageUse", + "LineageValidationError", "PopulationCustodyError", "PopulationCustodyLog", "PopulationCustodyPolicy", @@ -343,8 +362,11 @@ __all__ = [ "ProvenanceTrustPolicy", "ProvenanceValidationError", "derive_named_inputs_digest", + "derive_lineage_decision_digest", + "derive_lineage_manifest_digest", "derive_trust_policy_digest", "stamp", "validate_provenance", + "validate_lineage", "verify_stamp", ] diff --git a/tests/test_core_lineage.py b/tests/test_core_lineage.py new file mode 100644 index 0000000..0867009 --- /dev/null +++ b/tests/test_core_lineage.py @@ -0,0 +1,341 @@ +from __future__ import annotations + +from dataclasses import replace + +import pytest +from detection_core_fixtures import InMemoryPopulationCustody, custody_contract + +from openultrasast.core import canonical, provenance + + +def _lineage( + uses: tuple[provenance.LineageUse, ...], + *, + subject: str | None = None, + arm_id: str = "arm", +) -> provenance.LineageManifest: + subject_digest = subject or canonical.content_digest("subject") + digest = provenance.derive_lineage_manifest_digest(arm_id, subject_digest, uses) + return provenance.LineageManifest(arm_id, subject_digest, uses, digest) + + +def _use( + purpose: provenance.LineagePurpose, + artifact: str, + *, + labels: bool = False, +) -> tuple[provenance.LineageUse, InMemoryPopulationCustody, provenance.PopulationCustodyPolicy]: + initial, policy, custodian = custody_contract() + port = InMemoryPopulationCustody(initial, policy) + request = provenance.PopulationUseRequest( + initial.population_digest, + "arm", + purpose, + labels, + artifact, + initial.log_digest, + ) + receipt = port.append(request, custodian) + return ( + provenance.LineageUse(purpose, initial.population_digest, labels, artifact, receipt.receipt_digest), + port, + policy, + ) + + +def _expectation( + purpose: provenance.LineagePurpose, + artifact: str, + minimum: int = 1, + maximum: int | None = 1, +) -> provenance.LineageExpectation: + return provenance.LineageExpectation(purpose, artifact, minimum, maximum) + + +@pytest.mark.parametrize( + "purpose", + [ + provenance.LineagePurpose.PROMPT, + provenance.LineagePurpose.EXAMPLE, + provenance.LineagePurpose.THRESHOLD, + provenance.LineagePurpose.FITTED_MODEL, + provenance.LineagePurpose.SEMANTIC_FACT, + provenance.LineagePurpose.MATCHING_BEHAVIOR, + provenance.LineagePurpose.DECISION_MAPPING, + provenance.LineagePurpose.PROMOTION, + ], +) +def test_every_preparation_purpose_binds_exact_artifact_and_receipt( + purpose: provenance.LineagePurpose, +) -> None: + artifact = canonical.content_digest(purpose.value) + use, port, policy = _use(purpose, artifact) + manifest = _lineage((use,)) + + decision = provenance.validate_lineage( + manifest, + manifest.subject_digest, + (_expectation(purpose, artifact),), + None, + False, + port, + policy, + ) + + assert decision.allowed + assert decision.reasons == () + assert decision.decision_digest == provenance.derive_lineage_decision_digest(decision) + + +def test_required_optional_and_maximum_use_counts_are_enforced() -> None: + required = canonical.content_digest("required") + optional = canonical.content_digest("optional") + use, port, policy = _use(provenance.LineagePurpose.PROMPT, required) + manifest = _lineage((use,)) + allowed = provenance.validate_lineage( + manifest, + manifest.subject_digest, + ( + _expectation(provenance.LineagePurpose.PROMPT, required), + _expectation(provenance.LineagePurpose.EXAMPLE, optional, 0, 1), + ), + None, + False, + port, + policy, + ) + assert allowed.allowed + + current = port.current_log(use.source_population_digest) + second_receipt = port.append( + provenance.PopulationUseRequest( + use.source_population_digest, + "arm", + provenance.LineagePurpose.PROMPT, + False, + required, + current.log_digest, + ), + current.custodian_identity, + ) + second_use = provenance.LineageUse( + provenance.LineagePurpose.PROMPT, + use.source_population_digest, + False, + required, + second_receipt.receipt_digest, + ) + too_many_manifest = _lineage((use, second_use)) + too_many = provenance.validate_lineage( + too_many_manifest, + too_many_manifest.subject_digest, + (_expectation(provenance.LineagePurpose.PROMPT, required),), + None, + False, + port, + policy, + ) + assert not too_many.allowed + assert any(reason.startswith("too_many_uses:prompt") for reason in too_many.reasons) + + missing = provenance.validate_lineage( + _lineage(()), + manifest.subject_digest, + (_expectation(provenance.LineagePurpose.PROMPT, required),), + None, + False, + port, + policy, + ) + assert not missing.allowed + assert any(reason.startswith("missing_required_use:prompt") for reason in missing.reasons) + + +def test_duplicate_receipt_cannot_inflate_a_required_use_count() -> None: + artifact = canonical.content_digest("prompt") + use, port, policy = _use(provenance.LineagePurpose.PROMPT, artifact) + manifest = _lineage((use, use)) + + with pytest.raises(provenance.LineageValidationError, match="duplicate.*receipt"): + provenance.validate_lineage( + manifest, + manifest.subject_digest, + (_expectation(provenance.LineagePurpose.PROMPT, artifact, 2, 2),), + None, + False, + port, + policy, + ) + + +def test_matching_subject_cannot_mask_redirected_artifact_or_wrong_purpose() -> None: + expected = canonical.content_digest("expected") + redirected = canonical.content_digest("redirected") + use, port, policy = _use(provenance.LineagePurpose.PROMPT, redirected) + manifest = _lineage((use,)) + + decision = provenance.validate_lineage( + manifest, + manifest.subject_digest, + (_expectation(provenance.LineagePurpose.PROMPT, expected),), + None, + False, + port, + policy, + ) + assert not decision.allowed + assert any(reason.startswith("unexpected_use:prompt") for reason in decision.reasons) + assert any(reason.startswith("missing_required_use:prompt") for reason in decision.reasons) + + wrong_purpose = replace(use, purpose=provenance.LineagePurpose.EXAMPLE) + wrong_manifest = _lineage((wrong_purpose,)) + with pytest.raises(provenance.LineageValidationError, match="does not match its current-chain receipt"): + provenance.validate_lineage( + wrong_manifest, + wrong_manifest.subject_digest, + (_expectation(provenance.LineagePurpose.PROMPT, redirected),), + None, + False, + port, + policy, + ) + + +def test_fact_ancestry_forbids_all_label_access_and_records_spent_population() -> None: + artifact = canonical.content_digest("fact") + use, port, policy = _use(provenance.LineagePurpose.SEMANTIC_FACT, artifact, labels=True) + manifest = _lineage((use,)) + + decision = provenance.validate_lineage( + manifest, + manifest.subject_digest, + (_expectation(provenance.LineagePurpose.SEMANTIC_FACT, artifact),), + None, + True, + port, + policy, + ) + + assert not decision.allowed + assert decision.spent_population_digests == (use.source_population_digest,) + assert decision.reasons == (f"label_access_forbidden:{use.source_population_digest}",) + + +def test_development_lineage_may_use_different_spent_population() -> None: + artifact = canonical.content_digest("development-model") + protected = canonical.content_digest("admission-population") + use, port, policy = _use(provenance.LineagePurpose.FITTED_MODEL, artifact, labels=True) + manifest = _lineage((use,)) + + decision = provenance.validate_lineage( + manifest, + manifest.subject_digest, + (_expectation(provenance.LineagePurpose.FITTED_MODEL, artifact),), + protected, + False, + port, + policy, + ) + + assert decision.allowed + assert decision.spent_population_digests == (use.source_population_digest,) + + +def test_protected_population_label_influence_is_refused() -> None: + artifact = canonical.content_digest("threshold") + use, port, policy = _use(provenance.LineagePurpose.THRESHOLD, artifact, labels=True) + manifest = _lineage((use,)) + + decision = provenance.validate_lineage( + manifest, + manifest.subject_digest, + (_expectation(provenance.LineagePurpose.THRESHOLD, artifact),), + use.source_population_digest, + False, + port, + policy, + ) + + assert not decision.allowed + assert decision.reasons == (f"protected_population_labels:{use.source_population_digest}",) + + +def test_incomplete_current_history_and_substituted_receipt_are_refused() -> None: + artifact = canonical.content_digest("prompt") + use, port, policy = _use(provenance.LineagePurpose.PROMPT, artifact) + absent = replace(use, population_use_receipt_digest=canonical.content_digest("absent")) + manifest = _lineage((absent,)) + + with pytest.raises(provenance.LineageValidationError, match="absent from the current complete custody chain"): + provenance.validate_lineage( + manifest, + manifest.subject_digest, + (_expectation(provenance.LineagePurpose.PROMPT, artifact),), + None, + False, + port, + policy, + ) + + +def test_receipt_issued_to_another_arm_is_refused() -> None: + artifact = canonical.content_digest("prompt") + use, port, policy = _use(provenance.LineagePurpose.PROMPT, artifact) + manifest = _lineage((use,), arm_id="different-arm") + + with pytest.raises(provenance.LineageValidationError, match="arm"): + provenance.validate_lineage( + manifest, + manifest.subject_digest, + (_expectation(provenance.LineagePurpose.PROMPT, artifact),), + None, + False, + port, + policy, + ) + + +def test_manifest_identity_subject_and_duplicate_expectation_tampering_is_refused() -> None: + artifact = canonical.content_digest("prompt") + use, port, policy = _use(provenance.LineagePurpose.PROMPT, artifact) + manifest = _lineage((use,)) + expected = _expectation(provenance.LineagePurpose.PROMPT, artifact) + + with pytest.raises(provenance.LineageValidationError, match="manifest digest"): + provenance.validate_lineage( + replace(manifest, manifest_digest=canonical.content_digest("tampered")), + manifest.subject_digest, + (expected,), + None, + False, + port, + policy, + ) + with pytest.raises(provenance.LineageValidationError, match="subject"): + provenance.validate_lineage( + manifest, + canonical.content_digest("other-subject"), + (expected,), + None, + False, + port, + policy, + ) + with pytest.raises(provenance.LineageValidationError, match="duplicate"): + provenance.validate_lineage( + manifest, + manifest.subject_digest, + (expected, expected), + None, + False, + port, + policy, + ) + + +def test_lineage_models_validate_runtime_types_and_counts() -> None: + digest = canonical.content_digest("value") + with pytest.raises(ValueError, match="maximum_uses"): + provenance.LineageExpectation(provenance.LineagePurpose.PROMPT, digest, 2, 1) + with pytest.raises(TypeError, match="labels_accessed"): + provenance.LineageUse(provenance.LineagePurpose.PROMPT, digest, 1, digest, digest) # type: ignore[arg-type]