diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index 09a51f7..ff4f607 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -146,7 +146,7 @@ - Mark label-influenced matching behavior or promotion as spent before the relation can participate in evaluation. - Done when valid matches yield canonical effects while partial and ambiguous cases remain non-determinate with named reasons. - _Requirements: 4.1, 4.2, 4.3, 10.2, 11.2, 11.3, 11.5, 13.4_ -- [ ] 5.4 Revalidate fact-derived relations and enforce the dependency boundary +- [x] 5.4 Revalidate fact-derived relations and enforce the dependency boundary - Re-open referenced packs, facts, match evidence, review policy, lineage, custody, and resolved bytes for every fact-derived relation. - Reject undeclared roles, invalid scope, inconsistent content, and fact provenance that cannot be reproduced. - Require reviewed semantic facts for determinate behavior attributed to excluded dependencies. diff --git a/src/openultrasast/core/factpack.py b/src/openultrasast/core/factpack.py index 38d95ef..6a7e606 100644 --- a/src/openultrasast/core/factpack.py +++ b/src/openultrasast/core/factpack.py @@ -11,7 +11,13 @@ from types import MappingProxyType from openultrasast.core import _factpack_digest as digest_projection from openultrasast.core import canonical from openultrasast.core._legacy_factpack import FactPackError, admit_fact_pack, admit_pack_file, load_pack_file, verify_fact_pack -from openultrasast.core.custody import LineagePurpose, PopulationCustodyPolicy, PopulationCustodyPort +from openultrasast.core.custody import ( + LineagePurpose, + PopulationCustodyPolicy, + PopulationCustodyPort, + PopulationUseReceipt, + validate_population_custody_log, +) from openultrasast.core.family import FamilyContract, family_contract_digest, validate_contract from openultrasast.core.lineage import LineageExpectation, LineageManifest, validate_lineage from openultrasast.core.semantics import RelationKind @@ -366,6 +372,49 @@ def _lineage_for(evidence: KnowledgeEvidence, lineages: Sequence[LineageManifest return matches[0] +def _authoritative_semantic_fact_receipts( + selected_lineage: LineageManifest, + lineages: Sequence[LineageManifest], + pack: FactPack, + protected_population_digest: str | None, + custody: PopulationCustodyPort, + custody_policy: PopulationCustodyPolicy, +) -> tuple[PopulationUseReceipt, ...]: + populations = {use.source_population_digest for lineage in lineages for use in lineage.uses} + if protected_population_digest is not None: + populations.add(protected_population_digest) + relevant: list[PopulationUseReceipt] = [] + for population_digest in sorted(populations): + log = custody.current_log(population_digest) + validate_population_custody_log(log, custody_policy, current=custody) + relevant.extend( + receipt + for receipt in log.receipts + if receipt.request.purpose is LineagePurpose.SEMANTIC_FACT + and receipt.request.influenced_artifact_digest == pack.normalized_content_digest + and (receipt.request.arm_id == selected_lineage.arm_id or receipt.request.labels_accessed) + ) + return tuple(relevant) + + +def _refuse_omitted_semantic_fact_uses( + lineage: LineageManifest, + receipts: tuple[PopulationUseReceipt, ...], + protected_population_digest: str | None, +) -> None: + declared = {use.population_use_receipt_digest for use in lineage.uses} + for receipt in receipts: + if receipt.receipt_digest in declared: + continue + request = receipt.request + if request.labels_accessed: + reasons = [f"label_access_forbidden:{request.population_digest}"] + if request.population_digest == protected_population_digest: + reasons.append(f"protected_population_labels:{request.population_digest}") + raise FactPackValidationError("semantic fact lineage is contaminated: " + ",".join(reasons)) + raise FactPackValidationError(f"semantic fact lineage omitted current use: {receipt.receipt_digest}") + + def _validate_evidence_lineage( pack: FactPack, evidence: KnowledgeEvidence, @@ -390,6 +439,15 @@ def _validate_evidence_lineage( custody, custody_policy, ) + current_receipts = _authoritative_semantic_fact_receipts( + lineage, + lineages, + pack, + protected_population_digest, + custody, + custody_policy, + ) + _refuse_omitted_semantic_fact_uses(lineage, current_receipts, protected_population_digest) if not decision.allowed: raise FactPackValidationError("fact lineage refused: " + ",".join(decision.reasons)) diff --git a/src/openultrasast/core/factvalidate.py b/src/openultrasast/core/factvalidate.py new file mode 100644 index 0000000..bdc03c4 --- /dev/null +++ b/src/openultrasast/core/factvalidate.py @@ -0,0 +1,233 @@ +"""Replay validation for fact-derived relations and dependency boundaries.""" + +from __future__ import annotations + +import json +from collections.abc import Mapping, Sequence +from typing import cast + +from openultrasast.core import canonical +from openultrasast.core.custody import PopulationCustodyPolicy, PopulationCustodyPort +from openultrasast.core.evidence import EvidenceRecord +from openultrasast.core.factapply import apply_fact +from openultrasast.core.factmatch import FACT_MATCH_SCHEMA, FactMatchEvidence, StructuralObservation +from openultrasast.core.factpack import FactPack, ReviewIndependencePolicy +from openultrasast.core.family import FamilyContract, validate_record +from openultrasast.core.lineage import LineageManifest +from openultrasast.core.semantics import ProvenanceKind, SemanticRelation, SourceLocation + + +class FactRelationValidationError(ValueError): + """A fact-derived relation cannot be reproduced from its named inputs.""" + + +def _mapping(value: object, keys: set[str], field: str) -> Mapping[str, object]: + if not isinstance(value, Mapping) or not all(isinstance(key, str) for key in value) or set(value) != keys: + raise FactRelationValidationError(f"{field} must be an exact canonical object") + return cast(Mapping[str, object], value) + + +def _text(value: object, field: str) -> str: + if not isinstance(value, str) or not value: + raise FactRelationValidationError(f"{field} must be nonempty text") + return value + + +def _integer(value: object, field: str, *, minimum: int) -> int: + if isinstance(value, bool) or not isinstance(value, int) or value < minimum: + raise FactRelationValidationError(f"{field} must be an integer of at least {minimum}") + return value + + +def _optional_text(value: object, field: str) -> str | None: + if value is None: + return None + return _text(value, field) + + +def _optional_integer(value: object, field: str) -> int | None: + if value is None: + return None + return _integer(value, field, minimum=0) + + +def _json_value(value: object) -> canonical.JsonValue: + if value is None or isinstance(value, (str, bool, int, float)): + canonical.canonical_bytes(value) + return value + if isinstance(value, list): + return tuple(_json_value(item) for item in value) + if isinstance(value, Mapping) and all(isinstance(key, str) for key in value): + return {cast(str, key): _json_value(item) for key, item in value.items()} + raise FactRelationValidationError("match artifact contains a value outside the canonical JSON profile") + + +def _location(value: object, field: str) -> SourceLocation: + item = _mapping(value, {"path", "line", "column", "symbol"}, field) + return SourceLocation( + _text(item["path"], f"{field}.path"), + _integer(item["line"], f"{field}.line", minimum=1), + _optional_integer(item["column"], f"{field}.column"), + _optional_text(item["symbol"], f"{field}.symbol"), + ) + + +def _reference(value: object, field: str) -> canonical.ContentRef: + item = _mapping( + value, + {"artifact_id", "content_digest", "media_type", "byte_length", "retention_policy_digest", "storage_receipt_digest"}, + field, + ) + return canonical.ContentRef( + _text(item["artifact_id"], f"{field}.artifact_id"), + _text(item["content_digest"], f"{field}.content_digest"), + _text(item["media_type"], f"{field}.media_type"), + _integer(item["byte_length"], f"{field}.byte_length", minimum=0), + _text(item["retention_policy_digest"], f"{field}.retention_policy_digest"), + _text(item["storage_receipt_digest"], f"{field}.storage_receipt_digest"), + ) + + +def _observation(value: object, index: int) -> StructuralObservation: + field = f"observations[{index}]" + item = _mapping(value, {"requirement_id", "location", "observed", "evidence"}, field) + return StructuralObservation( + _text(item["requirement_id"], f"{field}.requirement_id"), + _location(item["location"], f"{field}.location"), + _json_value(item["observed"]), + _reference(item["evidence"], f"{field}.evidence"), + ) + + +def _match_from_bytes(content: bytes) -> FactMatchEvidence: + try: + parsed = json.loads(content.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise FactRelationValidationError(f"match artifact is not canonical JSON: {exc}") from exc + normalized = _json_value(parsed) + if canonical.canonical_bytes(normalized) != content: + raise FactRelationValidationError("match artifact bytes are not canonical") + item = _mapping( + parsed, + { + "schema_version", + "pack_digest", + "fact_id", + "product_location", + "observations", + "matcher_revision", + "matcher_behavior_digest", + "matcher_lineage_manifest_digest", + }, + "match artifact", + ) + if item["schema_version"] != FACT_MATCH_SCHEMA: + raise FactRelationValidationError("match artifact has an unknown schema version") + observations = item["observations"] + if not isinstance(observations, list): + raise FactRelationValidationError("match observations must be an array") + return FactMatchEvidence( + _text(item["pack_digest"], "match.pack_digest"), + _text(item["fact_id"], "match.fact_id"), + _location(item["product_location"], "match.product_location"), + tuple(_observation(value, index) for index, value in enumerate(observations)), + _text(item["matcher_revision"], "match.matcher_revision"), + _text(item["matcher_behavior_digest"], "match.matcher_behavior_digest"), + _text(item["matcher_lineage_manifest_digest"], "match.matcher_lineage_manifest_digest"), + canonical.bytes_digest(content), + ) + + +def _pack_for(relation: SemanticRelation, packs: Sequence[FactPack]) -> FactPack: + digest = relation.provenance.semantic_pack_digest + matches = tuple(pack for pack in packs if pack.normalized_content_digest == digest) + if len(matches) != 1: + raise FactRelationValidationError("semantic pack identity is missing or ambiguous") + return matches[0] + + +def _replay_relation( + relation: SemanticRelation, + record: EvidenceRecord, + contract: FamilyContract, + packs: Sequence[FactPack], + population_digest: str, + lineages: Sequence[LineageManifest], + custody: PopulationCustodyPort, + custody_policy: PopulationCustodyPolicy, + review_policy: ReviewIndependencePolicy, + resolver: canonical.ContentResolver, +) -> None: + pack = _pack_for(relation, packs) + if (pack.family_id, pack.family_contract_digest, pack.ecosystem) != ( + record.packet.family_id, + record.packet.family_contract_digest, + record.packet.ecosystem, + ): + raise FactRelationValidationError("semantic pack scope does not match the evidence record") + reference = relation.provenance.fact_match + if reference is None or relation.provenance.semantic_fact_id is None: + raise FactRelationValidationError("semantic fact provenance is incomplete") + content = canonical.resolve_content( + reference, + resolver, + trusted_store_identities=review_policy.trusted_store_identities, + maximum_bytes=review_policy.maximum_evidence_bytes, + ) + match = _match_from_bytes(content) + replayed = apply_fact( + pack, + relation.provenance.semantic_fact_id, + match, + reference, + contract, + population_digest, + lineages, + custody, + custody_policy, + review_policy, + resolver, + ) + if relation not in replayed: + raise FactRelationValidationError("fact-derived relation does not equal replayed declared effect") + + +def validate_fact_relations( + record: EvidenceRecord, + contract: FamilyContract, + packs: Sequence[FactPack], + evaluation_population_digest: str, + lineages: Sequence[LineageManifest], + custody: PopulationCustodyPort, + custody_policy: PopulationCustodyPolicy, + review_policy: ReviewIndependencePolicy, + resolver: canonical.ContentResolver, +) -> None: + """Re-open and replay every fact-derived relation in an evidence record.""" + try: + validate_record(contract, record) + if record.packet.provenance.population_digest != evaluation_population_digest: + raise FactRelationValidationError("record has an invalid evaluation population scope") + if not isinstance(packs, Sequence) or not all(isinstance(pack, FactPack) for pack in packs): + raise FactRelationValidationError("packs must be a sequence of FactPack values") + for relation in record.packet.transitions: + if relation.provenance.kind is ProvenanceKind.SEMANTIC_FACT: + _replay_relation( + relation, + record, + contract, + packs, + evaluation_population_digest, + lineages, + custody, + custody_policy, + review_policy, + resolver, + ) + except FactRelationValidationError: + raise + except (LookupError, TypeError, ValueError) as exc: + raise FactRelationValidationError(f"fact relation replay failed: {exc}") from exc + + +__all__ = ["FactRelationValidationError", "validate_fact_relations"] diff --git a/tests/test_core_factapply.py b/tests/test_core_factapply.py index b85336c..6a9f763 100644 --- a/tests/test_core_factapply.py +++ b/tests/test_core_factapply.py @@ -7,7 +7,7 @@ from dataclasses import dataclass, replace import pytest from detection_core_fixtures import InMemoryContentResolver, InMemoryPopulationCustody, custody_contract, stored_artifact -from openultrasast.core import canonical, provenance +from openultrasast.core import canonical, evidence, provenance from openultrasast.core.evidence import NeutralOutcome from openultrasast.core.factapply import FactApplicationError, apply_fact from openultrasast.core.factmatch import ( @@ -34,7 +34,9 @@ from openultrasast.core.factpack import ( derive_review_evidence_digest, derive_review_independence_policy_digest, ) +from openultrasast.core.factvalidate import FactRelationValidationError, validate_fact_relations from openultrasast.core.family import ( + MISSING_REVIEWED_DEPENDENCY, DecisionRule, FamilyContract, FieldSpec, @@ -42,15 +44,20 @@ from openultrasast.core.family import ( RelationOriginPolicy, ResidualQuestion, RoleSpec, + derive_security_object_id, family_contract_digest, ) from openultrasast.core.semantics import ( + EvidenceProvenance, FieldKind, + FieldValue, ProvenanceKind, RelationKind, RelationStatus, + SemanticNode, SemanticRelation, SourceLocation, + make_relation, ) @@ -68,21 +75,47 @@ class ApplyCase: population_digest: str -def _contract() -> FamilyContract: +class MultiPopulationCustody(InMemoryPopulationCustody): + def __init__( + self, + logs: tuple[provenance.PopulationCustodyLog, ...], + policy: provenance.PopulationCustodyPolicy, + ) -> None: + self._logs = {log.population_digest: log for log in logs} + self._multi_policy = policy + + def current_log(self, population_digest: str) -> provenance.PopulationCustodyLog: + try: + return self._logs[population_digest] + except KeyError as exc: + raise provenance.PopulationCustodyError("population custody log not found") from exc + + def append(self, request: provenance.PopulationUseRequest, custodian: str) -> provenance.PopulationUseReceipt: + prior = self.current_log(request.population_digest) + receipt = provenance.record_population_use(request, prior, custodian, self._multi_policy) + self._logs[request.population_digest] = provenance.append_population_use(prior, receipt, self._multi_policy) + return receipt + + +def _contract(*, fact_only: bool = False) -> FamilyContract: + origin = RelationOriginPolicy.EXCLUDED_DEPENDENCY_FACT_ONLY if fact_only else RelationOriginPolicy.PRODUCT_OR_REVIEWED_FACT return FamilyContract( "reference-family", "1", - "family.v2", + evidence.EVIDENCE_PACKET_SCHEMA, ("python",), ObjectSpec("resource", (FieldSpec("name", FieldKind.TEXT, True),), ("name",)), ( - RoleSpec("flow", (), (RelationKind.CALL,), RelationOriginPolicy.PRODUCT_OR_REVIEWED_FACT), + RoleSpec("flow", (), (RelationKind.CALL,), origin), RoleSpec("source", (FieldSpec("name", FieldKind.TEXT, False),), (), RelationOriginPolicy.PRODUCT_OR_REVIEWED_FACT), RoleSpec("target", (FieldSpec("name", FieldKind.TEXT, False),), (), RelationOriginPolicy.PRODUCT_OR_REVIEWED_FACT), ), - (DecisionRule("unknown", NeutralOutcome.INDETERMINATE, (), (RelationStatus.UNRESOLVED,)),), - (ResidualQuestion("classify", ("flow",), ("unknown",)),), - ("missing",), + ( + DecisionRule("safe", NeutralOutcome.SATISFIED, ("flow",), (RelationStatus.COMPLETE,)), + DecisionRule("unknown", NeutralOutcome.INDETERMINATE, (), (RelationStatus.INCOMPLETE, RelationStatus.UNRESOLVED)), + ), + (ResidualQuestion("classify", ("flow",), ("safe", "unknown")),), + ("missing", MISSING_REVIEWED_DEPENDENCY), ) @@ -94,8 +127,9 @@ def _append_use( *, labels_accessed: bool, arm_id: str, + population_digest: str | None = None, ) -> provenance.LineageUse: - current = custody.current_log(canonical.content_digest("population")) + current = custody.current_log(population_digest or canonical.content_digest("population")) request = provenance.PopulationUseRequest( current.population_digest, arm_id, @@ -152,8 +186,9 @@ def _case( matcher_labels: bool = False, promotion_labels: bool = False, omitted_promotion_labels: bool = False, + fact_only: bool = False, ) -> ApplyCase: - contract = _contract() + contract = _contract(fact_only=fact_only) evidence, review_resolver, reviewer = _review_evidence() review_policy = _review_policy(reviewer) effects = effects or (RelationEffect(RelationKind.CALL, "flow", "source", "target", ("matched",)),) @@ -283,6 +318,128 @@ def _apply(case: ApplyCase) -> tuple[SemanticRelation, ...]: ) +def _case_with_distinct_evaluation_population() -> ApplyCase: + case = _case(fact_only=True) + source_log = case.custody.current_log(case.population_digest) + evaluation_digest = canonical.content_digest("distinct-evaluation-population") + evaluation_log = provenance.empty_population_custody_log( + evaluation_digest, + case.custody_policy.reviewed_custodian_digests[0], + ) + custody = MultiPopulationCustody((source_log, evaluation_log), case.custody_policy) + return replace(case, custody=custody, population_digest=evaluation_digest) + + +def _case_with_three_populations() -> tuple[ApplyCase, str]: + case = _case(fact_only=True) + custodian = case.custody_policy.reviewed_custodian_digests[0] + source_log = case.custody.current_log(case.population_digest) + fact_empty = provenance.empty_population_custody_log(case.population_digest, custodian) + fact_log = provenance.append_population_use(fact_empty, source_log.receipts[0], case.custody_policy) + matcher_population = canonical.content_digest("distinct-matcher-population") + matcher_empty = provenance.empty_population_custody_log(matcher_population, custodian) + matcher_request = provenance.PopulationUseRequest( + matcher_population, + "matcher-arm", + provenance.LineagePurpose.MATCHING_BEHAVIOR, + False, + case.match.matcher_behavior_digest, + matcher_empty.log_digest, + ) + matcher_receipt = provenance.record_population_use(matcher_request, matcher_empty, custodian, case.custody_policy) + matcher_log = provenance.append_population_use(matcher_empty, matcher_receipt, case.custody_policy) + matcher_use = provenance.LineageUse( + provenance.LineagePurpose.MATCHING_BEHAVIOR, + matcher_population, + False, + case.match.matcher_behavior_digest, + matcher_receipt.receipt_digest, + ) + matcher_lineage = _manifest("matcher-arm", case.match.matcher_behavior_digest, (matcher_use,)) + provisional_match = replace( + case.match, + matcher_lineage_manifest_digest=matcher_lineage.manifest_digest, + match_digest="0" * 64, + ) + match = replace(provisional_match, match_digest=derive_fact_match_digest(provisional_match)) + match_reference, match_resolver = stored_artifact(fact_match_bytes(match), artifact_id="fact-match") + evaluation_digest = canonical.content_digest("third-evaluation-population") + evaluation_log = provenance.empty_population_custody_log(evaluation_digest, custodian) + custody = MultiPopulationCustody((fact_log, matcher_log, evaluation_log), case.custody_policy) + artifacts = {**case.resolver.artifacts, **match_resolver.artifacts} + changed = replace( + case, + match=match, + match_reference=match_reference, + lineages=(case.lineages[0], matcher_lineage), + custody=custody, + resolver=InMemoryContentResolver(artifacts), + population_digest=evaluation_digest, + ) + return changed, matcher_population + + +def _stamp(case: ApplyCase) -> provenance.ProvenanceStamp: + return provenance.ProvenanceStamp( + canonical.content_digest("analyzer"), + family_contract_digest(case.contract), + canonical.content_digest("fact-set"), + canonical.content_digest("acquisition"), + canonical.content_digest("configuration"), + case.population_digest, + canonical.content_digest("change-input"), + ) + + +def _record(case: ApplyCase, relation: SemanticRelation) -> evidence.EvidenceRecord: + fields = (FieldValue("name", FieldKind.TEXT, "resource"),) + scope = evidence.AnalyzedScope( + "base", + "head", + canonical.content_digest("change"), + (evidence.SourceRegion(case.match.product_location.path, 1, 30),), + ) + reasons = relation.reasons if relation.status is not RelationStatus.COMPLETE else () + packet = evidence.EvidencePacket( + evidence.EVIDENCE_PACKET_SCHEMA, + case.contract.family_id, + family_contract_digest(case.contract), + derive_security_object_id(case.contract.security_object, fields), + case.pack.ecosystem, + scope, + (relation,), + (), + fields, + reasons, + _stamp(case), + ) + outcome = NeutralOutcome.SATISFIED if relation.status is RelationStatus.COMPLETE else NeutralOutcome.INDETERMINATE + decision = evidence.Decision( + "safe" if relation.status is RelationStatus.COMPLETE else "unknown", + outcome, + evidence.EvidenceBasis.DETERMINISTIC, + scope, + reasons, + evidence.derive_packet_digest(packet), + provenance.DecisionProvenance(None, None, None, None, None, None), + ) + return evidence.EvidenceRecord(packet, decision, "") + + +def _validate(case: ApplyCase, record: evidence.EvidenceRecord, *, packs: tuple[FactPack, ...] | None = None) -> None: + validate_fact_relations( + record, + case.contract, + (case.pack,) if packs is None else packs, + case.population_digest, + case.lineages, + case.custody, + case.custody_policy, + case.review_policy, + case.resolver, + ) + + def test_complete_match_materializes_only_declared_effect_with_exact_provenance() -> None: case = _case() @@ -366,3 +523,267 @@ def test_omitted_current_promotion_spend_cannot_hide_behind_clean_pack_lineage() assert case.custody.current_log(case.population_digest).status is provenance.PopulationStatus.SPENT with pytest.raises(FactApplicationError, match="promotion.*protected_population_labels"): _apply(case) + + +def test_reviewed_dependency_relation_replays_from_every_named_input() -> None: + case = _case(fact_only=True) + (relation,) = _apply(case) + record = _record(case, relation) + + _validate(case, record) + + +def test_product_only_dependency_behavior_is_refused() -> None: + case = _case(fact_only=True) + (reviewed,) = _apply(case) + product = replace( + reviewed, + provenance=EvidenceProvenance(ProvenanceKind.PRODUCT_CODE, case.match.match_digest, None, None, None), + relation_id="", + ) + + with pytest.raises(FactRelationValidationError, match="reviewed semantic fact"): + _validate(case, _record(case, product)) + + +def test_missing_dependency_fact_remains_named_and_indeterminate() -> None: + case = _case(fact_only=True) + location = case.match.product_location + missing = make_relation( + kind=RelationKind.CALL, + role="flow", + source=SemanticNode("source", "source", location), + target=SemanticNode("target", "target", location), + location=location, + status=RelationStatus.INCOMPLETE, + conditions=("matched",), + reasons=(MISSING_REVIEWED_DEPENDENCY,), + provenance=EvidenceProvenance(ProvenanceKind.PRODUCT_CODE, case.match.match_digest, None, None, None), + ) + record = _record(case, missing) + + _validate(case, record) + + assert record.decision.outcome is NeutralOutcome.INDETERMINATE + assert record.decision.reasons == (MISSING_REVIEWED_DEPENDENCY,) + + +def test_fact_relation_replay_refuses_undeclared_role_and_invalid_scope() -> None: + case = _case(fact_only=True) + (relation,) = _apply(case) + undeclared = replace(relation, role="undeclared", relation_id="") + with pytest.raises(FactRelationValidationError, match="undeclared semantic role"): + _validate(case, _record(case, undeclared)) + + record = _record(case, relation) + packet = replace(record.packet, ecosystem="javascript") + decision = replace(record.decision, packet_digest=evidence.derive_packet_digest(packet)) + with pytest.raises(FactRelationValidationError, match="unsupported ecosystem scope"): + _validate(case, evidence.EvidenceRecord(packet, decision, "")) + + +def test_fact_relation_replay_refuses_inconsistent_relation_and_fact_identity() -> None: + case = _case(fact_only=True) + (relation,) = _apply(case) + inconsistent = replace(relation, conditions=("substituted",), relation_id="") + with pytest.raises(FactRelationValidationError, match="does not equal replayed"): + _validate(case, _record(case, inconsistent)) + + wrong_origin = replace(relation.provenance, semantic_fact_id="other.fact") + wrong_fact = replace(relation, provenance=wrong_origin, relation_id="") + with pytest.raises(FactRelationValidationError, match="fact identity"): + _validate(case, _record(case, wrong_fact)) + + +def test_fact_relation_replay_refuses_substituted_resolved_match_bytes() -> None: + case = _case(fact_only=True) + (relation,) = _apply(case) + wrong_ref, wrong_resolver = stored_artifact(b'{"plausible":"but-wrong"}', artifact_id="wrong-replay-match") + wrong_origin = replace( + relation.provenance, + source_content_digest=wrong_ref.content_digest, + fact_match=wrong_ref, + ) + wrong_relation = replace(relation, provenance=wrong_origin, relation_id="") + changed = replace( + case, + resolver=InMemoryContentResolver({**case.resolver.artifacts, **wrong_resolver.artifacts}), + ) + + with pytest.raises(FactRelationValidationError, match="match artifact"): + _validate(changed, _record(changed, wrong_relation)) + + +def test_fact_relation_replay_refuses_ambiguous_pack_and_changed_current_custody() -> None: + case = _case(fact_only=True) + (relation,) = _apply(case) + record = _record(case, relation) + with pytest.raises(FactRelationValidationError, match="pack identity is missing or ambiguous"): + _validate(case, record, packs=(case.pack, case.pack)) + + _append_use( + case.custody, + case.custody_policy.reviewed_custodian_digests[0], + provenance.LineagePurpose.PROMOTION, + case.pack.normalized_content_digest, + labels_accessed=True, + arm_id="other-arm", + ) + with pytest.raises(FactRelationValidationError, match="promotion lineage is contaminated"): + _validate(case, record) + + +def test_fact_relation_replay_refuses_omitted_current_semantic_fact_label_use() -> None: + case = _case(fact_only=True) + (relation,) = _apply(case) + record = _record(case, relation) + _append_use( + case.custody, + case.custody_policy.reviewed_custodian_digests[0], + provenance.LineagePurpose.SEMANTIC_FACT, + case.pack.normalized_content_digest, + labels_accessed=True, + arm_id="fact-arm", + ) + + assert case.custody.current_log(case.population_digest).status is provenance.PopulationStatus.SPENT + with pytest.raises(FactRelationValidationError, match="semantic fact.*protected_population_labels"): + _validate(case, record) + + +def test_fact_relation_replay_refuses_cross_arm_protected_semantic_fact_label_use() -> None: + case = _case(fact_only=True) + (relation,) = _apply(case) + record = _record(case, relation) + _append_use( + case.custody, + case.custody_policy.reviewed_custodian_digests[0], + provenance.LineagePurpose.SEMANTIC_FACT, + case.pack.normalized_content_digest, + labels_accessed=True, + arm_id="other-arm", + ) + + assert case.custody.current_log(case.population_digest).status is provenance.PopulationStatus.SPENT + with pytest.raises(FactRelationValidationError, match="semantic fact.*protected_population_labels"): + _validate(case, record) + + +def test_fact_relation_replay_refuses_cross_arm_label_use_from_lineage_source_population() -> None: + case = _case_with_distinct_evaluation_population() + (relation,) = _apply(case) + record = _record(case, relation) + source_population = canonical.content_digest("population") + _append_use( + case.custody, + case.custody_policy.reviewed_custodian_digests[0], + provenance.LineagePurpose.SEMANTIC_FACT, + case.pack.normalized_content_digest, + labels_accessed=True, + arm_id="other-arm", + ) + + assert source_population != case.population_digest + assert case.custody.current_log(source_population).status is provenance.PopulationStatus.SPENT + with pytest.raises(FactRelationValidationError, match="semantic fact.*label_access_forbidden"): + _validate(case, record) + + +def test_fact_relation_replay_refuses_label_use_from_other_supplied_lineage_population() -> None: + case, matcher_population = _case_with_three_populations() + (relation,) = _apply(case) + record = _record(case, relation) + _append_use( + case.custody, + case.custody_policy.reviewed_custodian_digests[0], + provenance.LineagePurpose.SEMANTIC_FACT, + case.pack.normalized_content_digest, + labels_accessed=True, + arm_id="other-arm", + population_digest=matcher_population, + ) + + assert matcher_population not in { + case.population_digest, + case.lineages[0].uses[0].source_population_digest, + } + assert case.custody.current_log(matcher_population).status is provenance.PopulationStatus.SPENT + with pytest.raises(FactRelationValidationError, match="semantic fact.*label_access_forbidden"): + _validate(case, record) + + +@pytest.mark.parametrize( + ("purpose", "subject", "labels_accessed"), + ( + (provenance.LineagePurpose.SEMANTIC_FACT, "exact-pack", False), + (provenance.LineagePurpose.MATCHING_BEHAVIOR, "exact-pack", True), + (provenance.LineagePurpose.SEMANTIC_FACT, "wrong-digest", True), + ), +) +def test_fact_relation_replay_ignores_unrelated_cross_arm_source_receipts( + purpose: provenance.LineagePurpose, + subject: str, + labels_accessed: bool, +) -> None: + case = _case_with_distinct_evaluation_population() + (relation,) = _apply(case) + record = _record(case, relation) + influenced = case.pack.normalized_content_digest if subject == "exact-pack" else canonical.content_digest(subject) + _append_use( + case.custody, + case.custody_policy.reviewed_custodian_digests[0], + purpose, + influenced, + labels_accessed=labels_accessed, + arm_id="other-arm", + ) + + _validate(case, record) + + +@pytest.mark.parametrize( + ("purpose", "subject", "labels_accessed"), + ( + (provenance.LineagePurpose.SEMANTIC_FACT, "exact-pack", False), + (provenance.LineagePurpose.MATCHING_BEHAVIOR, "exact-pack", True), + (provenance.LineagePurpose.SEMANTIC_FACT, "wrong-digest", True), + ), +) +def test_fact_relation_replay_ignores_unrelated_receipts_in_other_lineage_population( + purpose: provenance.LineagePurpose, + subject: str, + labels_accessed: bool, +) -> None: + case, matcher_population = _case_with_three_populations() + (relation,) = _apply(case) + record = _record(case, relation) + influenced = case.pack.normalized_content_digest if subject == "exact-pack" else canonical.content_digest(subject) + _append_use( + case.custody, + case.custody_policy.reviewed_custodian_digests[0], + purpose, + influenced, + labels_accessed=labels_accessed, + arm_id="other-arm", + population_digest=matcher_population, + ) + + _validate(case, record) + + +def test_fact_relation_replay_refuses_omitted_current_semantic_fact_use_without_labels() -> None: + case = _case(fact_only=True) + (relation,) = _apply(case) + record = _record(case, relation) + _append_use( + case.custody, + case.custody_policy.reviewed_custodian_digests[0], + provenance.LineagePurpose.SEMANTIC_FACT, + case.pack.normalized_content_digest, + labels_accessed=False, + arm_id="fact-arm", + ) + + assert case.custody.current_log(case.population_digest).status is provenance.PopulationStatus.AVAILABLE + with pytest.raises(FactRelationValidationError, match="semantic fact lineage omitted current use"): + _validate(case, record)