From 087dcc723fda9a1c07575cab789d7a0026b88e10 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sat, 31 Oct 2026 15:44:48 +0100 Subject: [PATCH] feat(detection-core): validate structural matches and matcher lineage --- .kiro/specs/detection-core/tasks.md | 2 +- src/openultrasast/core/factmatch.py | 428 +++++++++++++++++++++++++++ tests/test_core_factmatch.py | 442 ++++++++++++++++++++++++++++ 3 files changed, 871 insertions(+), 1 deletion(-) create mode 100644 src/openultrasast/core/factmatch.py create mode 100644 tests/test_core_factmatch.py diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index cd50114..1c2e8a5 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -132,7 +132,7 @@ - Reject label-derived semantic-fact ancestry, unknown scope/roles, missing review, and inconsistent digests. - Done when the schema/scope/review matrix accepts only fully reviewed label-isolated packs. - _Requirements: 11.1, 11.4, 13.3_ -- [ ] 5.2 Validate complete structural observations and matcher lineage +- [x] 5.2 Validate complete structural observations and matcher lineage - Match the declared requirement set, including qualified name, shape, receiver, target, argument, and configuration evidence as applicable. - Resolve every observation artifact and require its identifier set to equal the fact’s complete structural requirements. - Bind matcher revision, behavior digest, and current lineage to the exact evaluated product location. diff --git a/src/openultrasast/core/factmatch.py b/src/openultrasast/core/factmatch.py new file mode 100644 index 0000000..e4c18f5 --- /dev/null +++ b/src/openultrasast/core/factmatch.py @@ -0,0 +1,428 @@ +"""Exact structural fact-match evidence and matcher-lineage validation.""" + +from __future__ import annotations + +import re +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from types import MappingProxyType + +from openultrasast.core import canonical +from openultrasast.core.custody import ( + LineagePurpose, + PopulationCustodyError, + PopulationCustodyPolicy, + PopulationCustodyPort, + PopulationUseReceipt, + validate_population_custody_log, +) +from openultrasast.core.factpack import ( + FactPack, + ReviewIndependencePolicy, + SemanticFact, + StructuralRequirementKind, + derive_fact_pack_content_digest, + derive_review_independence_policy_digest, +) +from openultrasast.core.lineage import LineageExpectation, LineageManifest, LineageValidationError, validate_lineage +from openultrasast.core.semantics import RelationStatus, SourceLocation + +STRUCTURAL_OBSERVATION_SCHEMA = "openultrasast.detection-core.structural-observation.v2" +FACT_MATCH_SCHEMA = "openultrasast.detection-core.fact-match.v2" +_DIGEST = re.compile(r"[0-9a-f]{64}") + + +class FactMatchValidationError(ValueError): + """A fact match identity, artifact, or lineage join cannot be reproduced.""" + + +def _require_text(value: object, field: str) -> None: + if not isinstance(value, str) or not value: + raise FactMatchValidationError(f"{field} must be nonempty text") + + +def _require_digest(value: object, field: str) -> None: + if not isinstance(value, str) or _DIGEST.fullmatch(value) is None: + raise FactMatchValidationError(f"{field} must be a full lowercase SHA-256 digest") + + +def _freeze_json(value: canonical.JsonValue) -> canonical.JsonValue: + if isinstance(value, Mapping): + return MappingProxyType({key: _freeze_json(item) for key, item in value.items()}) + if isinstance(value, tuple): + return tuple(_freeze_json(item) for item in value) + return value + + +def _location_value(location: SourceLocation) -> Mapping[str, canonical.JsonValue]: + return {"path": location.path, "line": location.line, "column": location.column, "symbol": location.symbol} + + +def _reference_value(reference: canonical.ContentRef) -> Mapping[str, canonical.JsonValue]: + return { + "artifact_id": reference.artifact_id, + "content_digest": reference.content_digest, + "media_type": reference.media_type, + "byte_length": reference.byte_length, + "retention_policy_digest": reference.retention_policy_digest, + "storage_receipt_digest": reference.storage_receipt_digest, + } + + +@dataclass(frozen=True) +class StructuralObservation: + requirement_id: str + location: SourceLocation + observed: canonical.JsonValue + evidence: canonical.ContentRef + + def __post_init__(self) -> None: + _require_text(self.requirement_id, "requirement_id") + if not isinstance(self.location, SourceLocation): + raise FactMatchValidationError("observation location must be a SourceLocation") + self.location.__post_init__() + frozen = _freeze_json(self.observed) + canonical.canonical_bytes(frozen) + object.__setattr__(self, "observed", frozen) + if not isinstance(self.evidence, canonical.ContentRef): + raise FactMatchValidationError("observation evidence must be a ContentRef") + + +def _observation_payload(observation: StructuralObservation) -> Mapping[str, canonical.JsonValue]: + return { + "schema_version": STRUCTURAL_OBSERVATION_SCHEMA, + "requirement_id": observation.requirement_id, + "location": _location_value(observation.location), + "observed": observation.observed, + } + + +def structural_observation_bytes(observation: StructuralObservation) -> bytes: + """Canonical bytes that an observation ContentRef must resolve.""" + if not isinstance(observation, StructuralObservation): + raise TypeError("observation must be a StructuralObservation") + return canonical.canonical_record_bytes( + _observation_payload(observation), + known_schema_versions=(STRUCTURAL_OBSERVATION_SCHEMA,), + ) + + +def _match_observation_value(observation: StructuralObservation) -> Mapping[str, canonical.JsonValue]: + return { + "requirement_id": observation.requirement_id, + "location": _location_value(observation.location), + "observed": observation.observed, + "evidence": _reference_value(observation.evidence), + } + + +@dataclass(frozen=True) +class FactMatchEvidence: + pack_digest: str + fact_id: str + product_location: SourceLocation + observations: tuple[StructuralObservation, ...] + matcher_revision: str + matcher_behavior_digest: str + matcher_lineage_manifest_digest: str + match_digest: str + + def __post_init__(self) -> None: + _require_digest(self.pack_digest, "pack_digest") + _require_text(self.fact_id, "fact_id") + if not isinstance(self.product_location, SourceLocation): + raise FactMatchValidationError("product_location must be a SourceLocation") + self.product_location.__post_init__() + if not isinstance(self.observations, tuple) or not all(isinstance(item, StructuralObservation) for item in self.observations): + raise FactMatchValidationError("observations must be a tuple of StructuralObservation values") + _require_text(self.matcher_revision, "matcher_revision") + _require_digest(self.matcher_behavior_digest, "matcher_behavior_digest") + _require_digest(self.matcher_lineage_manifest_digest, "matcher_lineage_manifest_digest") + _require_digest(self.match_digest, "match_digest") + + +def _match_value(match: FactMatchEvidence) -> Mapping[str, canonical.JsonValue]: + return { + "schema_version": FACT_MATCH_SCHEMA, + "pack_digest": match.pack_digest, + "fact_id": match.fact_id, + "product_location": _location_value(match.product_location), + "observations": tuple(_match_observation_value(item) for item in match.observations), + "matcher_revision": match.matcher_revision, + "matcher_behavior_digest": match.matcher_behavior_digest, + "matcher_lineage_manifest_digest": match.matcher_lineage_manifest_digest, + } + + +def derive_fact_match_digest(match: FactMatchEvidence) -> str: + if not isinstance(match, FactMatchEvidence): + raise TypeError("match must be FactMatchEvidence") + return canonical.content_digest(_match_value(match)) + + +@dataclass(frozen=True) +class FactMatchDecision: + status: RelationStatus + reasons: tuple[str, ...] + effect_eligible: bool + lineage_decision_digest: str + + def __post_init__(self) -> None: + if not isinstance(self.status, RelationStatus): + raise FactMatchValidationError("status must be a RelationStatus") + if not isinstance(self.reasons, tuple) or not all(isinstance(item, str) and item for item in self.reasons): + raise FactMatchValidationError("reasons must be a tuple of nonempty text") + if self.reasons != tuple(sorted(set(self.reasons))): + raise FactMatchValidationError("reasons must be sorted and unique") + if not isinstance(self.effect_eligible, bool): + raise FactMatchValidationError("effect_eligible must be boolean") + if self.effect_eligible is not (self.status is RelationStatus.COMPLETE): + raise FactMatchValidationError("only COMPLETE decisions may be effect eligible") + _require_digest(self.lineage_decision_digest, "lineage_decision_digest") + + +def _fact_for(pack: FactPack, fact_id: str) -> SemanticFact: + matches = tuple(fact for fact in pack.facts if fact.fact_id == fact_id) + if len(matches) != 1: + raise FactMatchValidationError("fact identity is missing or ambiguous in pack") + return matches[0] + + +def _validate_named_inputs( + pack: FactPack, + fact_id: str, + match: FactMatchEvidence, + review_policy: ReviewIndependencePolicy, +) -> SemanticFact: + if pack.normalized_content_digest != derive_fact_pack_content_digest(pack): + raise FactMatchValidationError("pack digest cannot be re-derived") + if match.pack_digest != pack.normalized_content_digest: + raise FactMatchValidationError("match pack digest does not name supplied pack") + if match.fact_id != fact_id: + raise FactMatchValidationError("match fact identity does not name requested fact") + fact = _fact_for(pack, fact_id) + if match.match_digest != derive_fact_match_digest(match): + raise FactMatchValidationError("match digest cannot be re-derived") + expected_policy = derive_review_independence_policy_digest( + review_policy.reviewed_reviewer_digests, + review_policy.forbidden_identity_pairs, + review_policy.trusted_store_identities, + review_policy.maximum_evidence_bytes, + ) + if review_policy.policy_digest != expected_policy or pack.review_policy_digest != expected_policy: + raise FactMatchValidationError("review policy identity cannot be re-derived from named inputs") + return fact + + +def _resolve_observations( + match: FactMatchEvidence, + review_policy: ReviewIndependencePolicy, + resolver: canonical.ContentResolver, +) -> None: + for observation in match.observations: + observation.__post_init__() + observation.evidence.__post_init__() + try: + resolved = canonical.resolve_content( + observation.evidence, + resolver, + trusted_store_identities=review_policy.trusted_store_identities, + maximum_bytes=review_policy.maximum_evidence_bytes, + ) + except (canonical.ArtifactResolutionError, canonical.ArtifactValidationError) as exc: + raise FactMatchValidationError(f"observation artifact cannot be validated: {exc}") from exc + if resolved != structural_observation_bytes(observation): + raise FactMatchValidationError("resolved bytes do not equal canonical observation payload") + + +def _conflicting(items: tuple[StructuralObservation, ...]) -> bool: + values = { + canonical.content_digest( + { + "location": _location_value(item.location), + "observed": item.observed, + "evidence": _reference_value(item.evidence), + } + ) + for item in items + } + return len(values) > 1 + + +def _unresolved_reasons(fact: SemanticFact, match: FactMatchEvidence) -> list[str]: + expected_ids = {item.requirement_id for item in fact.structural_requirements} + grouped: dict[str, list[StructuralObservation]] = {} + reasons: list[str] = [] + for observation in match.observations: + grouped.setdefault(observation.requirement_id, []).append(observation) + if observation.requirement_id not in expected_ids: + reasons.append(f"fabricated_observation:{observation.requirement_id}") + if observation.location != match.product_location: + reasons.append(f"ambiguous_product_location:{observation.requirement_id}") + for requirement_id, items in grouped.items(): + if len(items) > 1: + reasons.extend(("ambiguous_match", f"duplicate_observation:{requirement_id}")) + if _conflicting(tuple(items)): + reasons.append(f"conflicting_observation:{requirement_id}") + return reasons + + +def _incomplete_reasons(fact: SemanticFact, match: FactMatchEvidence) -> list[str]: + indexed = {item.requirement_id: item for item in match.observations} + reasons: list[str] = [] + present_kinds: set[StructuralRequirementKind] = set() + for requirement in fact.structural_requirements: + observation = indexed.get(requirement.requirement_id) + if observation is None: + reasons.append(f"missing_observation:{requirement.requirement_id}") + continue + present_kinds.add(requirement.kind) + if canonical.canonical_bytes(observation.observed) != canonical.canonical_bytes(requirement.expected): + reasons.append(f"predicate_mismatch:{requirement.requirement_id}") + if present_kinds == {StructuralRequirementKind.QUALIFIED_NAME}: + reasons.append("name_only_match") + return reasons + + +def _lineage_for(match: FactMatchEvidence, lineages: Sequence[LineageManifest]) -> LineageManifest: + if not isinstance(lineages, Sequence) or not all(isinstance(item, LineageManifest) for item in lineages): + raise FactMatchValidationError("lineages must be a sequence of LineageManifest values") + matches = tuple(item for item in lineages if item.manifest_digest == match.matcher_lineage_manifest_digest) + if len(matches) != 1: + raise FactMatchValidationError("matcher lineage manifest is missing or ambiguous") + return matches[0] + + +def _authoritative_matching_receipts( + lineage: LineageManifest, + match: FactMatchEvidence, + protected_population_digest: str, + custody: PopulationCustodyPort, + custody_policy: PopulationCustodyPolicy, +) -> tuple[PopulationUseReceipt, ...]: + populations = {protected_population_digest, *(use.source_population_digest for use in lineage.uses)} + relevant: list[PopulationUseReceipt] = [] + for population_digest in sorted(populations): + log = custody.current_log(population_digest) + validate_population_custody_log(log, custody_policy, current=custody) + for receipt in log.receipts: + request = receipt.request + if ( + request.arm_id == lineage.arm_id + and request.purpose is LineagePurpose.MATCHING_BEHAVIOR + and request.influenced_artifact_digest == match.matcher_behavior_digest + ): + relevant.append(receipt) + return tuple(relevant) + + +def _omitted_matching_reasons( + lineage: LineageManifest, + current_receipts: tuple[PopulationUseReceipt, ...], + protected_population_digest: str, +) -> tuple[str, ...]: + declared = {use.population_use_receipt_digest for use in lineage.uses} + reasons: list[str] = [] + for receipt in current_receipts: + if receipt.receipt_digest in declared: + continue + reasons.append(f"omitted_current_matching_use:{receipt.receipt_digest}") + if receipt.request.labels_accessed: + reasons.append(f"label_access_forbidden:{receipt.population_digest}") + if receipt.population_digest == protected_population_digest: + reasons.append(f"protected_population_labels:{receipt.population_digest}") + return tuple(sorted(set(reasons))) + + +def _validate_matcher_lineage( + match: FactMatchEvidence, + protected_population_digest: str, + lineages: Sequence[LineageManifest], + custody: PopulationCustodyPort, + custody_policy: PopulationCustodyPolicy, +) -> tuple[tuple[str, ...], str]: + lineage = _lineage_for(match, lineages) + expectation = LineageExpectation(LineagePurpose.MATCHING_BEHAVIOR, match.matcher_behavior_digest, 1, 1) + try: + decision = validate_lineage( + lineage, + match.matcher_behavior_digest, + (expectation,), + protected_population_digest, + True, + custody, + custody_policy, + ) + current_receipts = _authoritative_matching_receipts( + lineage, + match, + protected_population_digest, + custody, + custody_policy, + ) + except (LineageValidationError, PopulationCustodyError, canonical.DigestFormatError) as exc: + raise FactMatchValidationError(f"matcher lineage contract is invalid: {exc}") from exc + omitted = _omitted_matching_reasons(lineage, current_receipts, protected_population_digest) + return tuple(sorted(set((*decision.reasons, *omitted)))), decision.decision_digest + + +def _decision( + unresolved: Sequence[str], + incomplete: Sequence[str], + lineage_digest: str, +) -> FactMatchDecision: + if unresolved: + status = RelationStatus.UNRESOLVED + reasons = tuple(sorted(set(unresolved))) + elif incomplete: + status = RelationStatus.INCOMPLETE + reasons = tuple(sorted(set(incomplete))) + else: + status = RelationStatus.COMPLETE + reasons = () + return FactMatchDecision(status, reasons, status is RelationStatus.COMPLETE, lineage_digest) + + +def validate_fact_match( + pack: FactPack, + fact_id: str, + match: FactMatchEvidence, + protected_population_digest: str, + lineages: Sequence[LineageManifest], + custody: PopulationCustodyPort, + custody_policy: PopulationCustodyPolicy, + review_policy: ReviewIndependencePolicy, + resolver: canonical.ContentResolver, +) -> FactMatchDecision: + """Resolve one complete fact match and make downstream effect eligibility explicit.""" + if not isinstance(pack, FactPack) or not isinstance(match, FactMatchEvidence): + raise FactMatchValidationError("pack and match must use v2 contract types") + if not isinstance(review_policy, ReviewIndependencePolicy): + raise FactMatchValidationError("review_policy must use the v2 contract type") + match.__post_init__() + review_policy.__post_init__() + fact = _validate_named_inputs(pack, fact_id, match, review_policy) + _resolve_observations(match, review_policy, resolver) + unresolved = _unresolved_reasons(fact, match) + incomplete = _incomplete_reasons(fact, match) + lineage_reasons, lineage_digest = _validate_matcher_lineage( + match, + protected_population_digest, + lineages, + custody, + custody_policy, + ) + return _decision((*unresolved, *lineage_reasons), incomplete, lineage_digest) + + +__all__ = [ + "FACT_MATCH_SCHEMA", + "STRUCTURAL_OBSERVATION_SCHEMA", + "FactMatchDecision", + "FactMatchEvidence", + "FactMatchValidationError", + "StructuralObservation", + "derive_fact_match_digest", + "structural_observation_bytes", + "validate_fact_match", +] diff --git a/tests/test_core_factmatch.py b/tests/test_core_factmatch.py new file mode 100644 index 0000000..bd6dea6 --- /dev/null +++ b/tests/test_core_factmatch.py @@ -0,0 +1,442 @@ +"""Structural fact-match validation and matcher-lineage contracts.""" + +from __future__ import annotations + +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.factmatch import ( + FACT_MATCH_SCHEMA, + STRUCTURAL_OBSERVATION_SCHEMA, + FactMatchDecision, + FactMatchEvidence, + FactMatchValidationError, + StructuralObservation, + derive_fact_match_digest, + structural_observation_bytes, + validate_fact_match, +) +from openultrasast.core.factpack import ( + FACT_PACK_SCHEMA, + FactPack, + RelationEffect, + ReviewIndependencePolicy, + SemanticFact, + StructuralRequirement, + StructuralRequirementKind, + derive_fact_pack_content_digest, + derive_review_independence_policy_digest, +) +from openultrasast.core.semantics import RelationKind, RelationStatus, SourceLocation + + +@dataclass(frozen=True) +class MatchCase: + pack: FactPack + match: FactMatchEvidence + resolver: InMemoryContentResolver + lineage: provenance.LineageManifest + custody: InMemoryPopulationCustody + custody_policy: provenance.PopulationCustodyPolicy + review_policy: ReviewIndependencePolicy + protected_population: str + + +def _requirements() -> tuple[StructuralRequirement, ...]: + return ( + StructuralRequirement("qualified", StructuralRequirementKind.QUALIFIED_NAME, "callee", "pkg.Client.call"), + StructuralRequirement("shape", StructuralRequirementKind.AST_SHAPE, "shape", {"node": "call"}), + StructuralRequirement("receiver", StructuralRequirementKind.RECEIVER_TYPE, "receiver", "pkg.Client"), + StructuralRequirement("target", StructuralRequirementKind.CALL_TARGET, "target", "pkg.Client.call"), + StructuralRequirement("arguments", StructuralRequirementKind.ARGUMENT_SHAPE, "arguments", ("resource",)), + StructuralRequirement("configuration", StructuralRequirementKind.CONFIGURATION, "config", {"mode": "strict"}), + ) + + +def _pack(requirements: tuple[StructuralRequirement, ...] | None = None) -> FactPack: + policy = _review_policy() + fact = SemanticFact( + "stable.fact.v1", + "flow", + requirements or _requirements(), + (RelationEffect(RelationKind.CALL, "flow", "source", "target", ("matched",)),), + ) + provisional = FactPack( + FACT_PACK_SCHEMA, + canonical.content_digest("family-contract"), + "reference-family", + "python", + "python", + (fact,), + "0" * 64, + canonical.content_digest("reviewer"), + (), + canonical.content_digest(()), + policy.policy_digest, + ) + return replace(provisional, normalized_content_digest=derive_fact_pack_content_digest(provisional)) + + +def _review_policy() -> ReviewIndependencePolicy: + reviewers = (canonical.content_digest("reviewer"),) + stores = ("trusted-store",) + digest = derive_review_independence_policy_digest(reviewers, (), stores, 4096) + return ReviewIndependencePolicy(reviewers, (), stores, 4096, digest) + + +def _matcher_lineage( + behavior: str, + *, + labels_accessed: bool = False, + purpose: provenance.LineagePurpose = provenance.LineagePurpose.MATCHING_BEHAVIOR, +) -> tuple[provenance.LineageManifest, InMemoryPopulationCustody, provenance.PopulationCustodyPolicy, str]: + initial, policy, custodian = custody_contract() + custody = InMemoryPopulationCustody(initial, policy) + request = provenance.PopulationUseRequest( + initial.population_digest, + "matcher-arm", + purpose, + labels_accessed, + behavior, + initial.log_digest, + ) + receipt = custody.append(request, custodian) + use = provenance.LineageUse( + request.purpose, + request.population_digest, + labels_accessed, + behavior, + receipt.receipt_digest, + ) + manifest_digest = provenance.derive_lineage_manifest_digest("matcher-arm", behavior, (use,)) + return provenance.LineageManifest("matcher-arm", behavior, (use,), manifest_digest), custody, policy, initial.population_digest + + +def _observation( + requirement: StructuralRequirement, + location: SourceLocation, + *, + observed: canonical.JsonValue | None = None, + suffix: str = "", +) -> tuple[StructuralObservation, InMemoryContentResolver]: + value = requirement.expected if observed is None else observed + provisional_reference = canonical.ContentRef("provisional", "0" * 64, "application/json", 0, "0" * 64, "0" * 64) + provisional = StructuralObservation(requirement.requirement_id, location, value, provisional_reference) + reference, resolver = stored_artifact( + structural_observation_bytes(provisional), + artifact_id=f"observation-{requirement.requirement_id}{suffix}", + ) + return replace(provisional, evidence=reference), resolver + + +def _case(*, labels_accessed: bool = False) -> MatchCase: + pack = _pack() + location = SourceLocation("src/product.py", 17, 8, "run") + artifacts: dict[str, canonical.ResolvedContent] = {} + observations: list[StructuralObservation] = [] + for requirement in pack.facts[0].structural_requirements: + observation, resolver = _observation(requirement, location) + observations.append(observation) + artifacts.update(resolver.artifacts) + behavior = canonical.content_digest("matcher-behavior-v3") + lineage, custody, custody_policy, protected = _matcher_lineage(behavior, labels_accessed=labels_accessed) + provisional = FactMatchEvidence( + pack.normalized_content_digest, + pack.facts[0].fact_id, + location, + tuple(observations), + "matcher-v3", + behavior, + lineage.manifest_digest, + "0" * 64, + ) + match = replace(provisional, match_digest=derive_fact_match_digest(provisional)) + return MatchCase( + pack, + match, + InMemoryContentResolver(artifacts), + lineage, + custody, + custody_policy, + _review_policy(), + protected, + ) + + +def _redigest(match: FactMatchEvidence) -> FactMatchEvidence: + return replace(match, match_digest=derive_fact_match_digest(match)) + + +def _validate(case: MatchCase, match: FactMatchEvidence | None = None) -> FactMatchDecision: + return validate_fact_match( + case.pack, + case.pack.facts[0].fact_id, + match or case.match, + case.protected_population, + (case.lineage,), + case.custody, + case.custody_policy, + case.review_policy, + case.resolver, + ) + + +def test_complete_match_requires_all_six_typed_observations_and_current_lineage() -> None: + case = _case() + + decision = _validate(case) + + assert {item.kind for item in case.pack.facts[0].structural_requirements} == set(StructuralRequirementKind) + assert decision.status is RelationStatus.COMPLETE + assert decision.reasons == () + assert decision.effect_eligible + assert decision.lineage_decision_digest is not None + assert FACT_MATCH_SCHEMA.endswith(".v2") and STRUCTURAL_OBSERVATION_SCHEMA.endswith(".v2") + + +@pytest.mark.parametrize("missing", [item.requirement_id for item in _requirements()]) +def test_missing_structural_observation_is_named_incomplete(missing: str) -> None: + case = _case() + observations = tuple(item for item in case.match.observations if item.requirement_id != missing) + match = _redigest(replace(case.match, observations=observations)) + + decision = _validate(case, match) + + assert decision.status is RelationStatus.INCOMPLETE + assert f"missing_observation:{missing}" in decision.reasons + assert not decision.effect_eligible + + +def test_name_only_match_can_never_be_complete() -> None: + requirement = _requirements()[0] + pack = _pack((requirement,)) + case = _case() + match = _redigest( + replace( + case.match, + pack_digest=pack.normalized_content_digest, + observations=(case.match.observations[0],), + ) + ) + case = replace(case, pack=pack) + + decision = _validate(case, match) + + assert decision.status is RelationStatus.INCOMPLETE + assert decision.reasons == ("name_only_match",) + assert not decision.effect_eligible + + +def test_typed_predicate_mismatch_is_incomplete() -> None: + case = _case() + requirement = case.pack.facts[0].structural_requirements[2] + observation, resolver = _observation(requirement, case.match.product_location, observed="pkg.Other", suffix="-wrong") + observations = tuple(observation if item.requirement_id == requirement.requirement_id else item for item in case.match.observations) + match = _redigest(replace(case.match, observations=observations)) + case = replace(case, resolver=InMemoryContentResolver({**case.resolver.artifacts, **resolver.artifacts})) + + decision = _validate(case, match) + + assert decision.status is RelationStatus.INCOMPLETE + assert decision.reasons == ("predicate_mismatch:receiver",) + + +@pytest.mark.parametrize("mode", ("duplicate", "conflict", "fabricated")) +def test_ambiguous_or_fabricated_identifier_sets_are_unresolved(mode: str) -> None: + case = _case() + observations = case.match.observations + resolver = case.resolver + if mode == "duplicate": + observations += (observations[0],) + elif mode == "conflict": + requirement = case.pack.facts[0].structural_requirements[0] + extra, extra_resolver = _observation(requirement, case.match.product_location, observed="pkg.Other", suffix="-conflict") + observations += (extra,) + resolver = InMemoryContentResolver({**resolver.artifacts, **extra_resolver.artifacts}) + else: + fabricated = replace(observations[0], requirement_id="invented") + reference, extra_resolver = stored_artifact( + structural_observation_bytes(fabricated), + artifact_id="observation-invented", + ) + observations += (replace(fabricated, evidence=reference),) + resolver = InMemoryContentResolver({**resolver.artifacts, **extra_resolver.artifacts}) + match = _redigest(replace(case.match, observations=observations)) + + decision = _validate(replace(case, resolver=resolver), match) + + assert decision.status is RelationStatus.UNRESOLVED + assert not decision.effect_eligible + assert any(mode in reason or "ambiguous" in reason for reason in decision.reasons) + + +def test_observation_from_another_product_location_is_unresolved() -> None: + case = _case() + requirement = case.pack.facts[0].structural_requirements[1] + other = SourceLocation("src/other.py", 4, 0, None) + observation, resolver = _observation(requirement, other, suffix="-other-location") + observations = tuple(observation if item.requirement_id == requirement.requirement_id else item for item in case.match.observations) + match = _redigest(replace(case.match, observations=observations)) + case = replace(case, resolver=InMemoryContentResolver({**case.resolver.artifacts, **resolver.artifacts})) + + decision = _validate(case, match) + + assert decision.status is RelationStatus.UNRESOLVED + assert decision.reasons == ("ambiguous_product_location:shape",) + + +def test_every_observation_artifact_must_resolve_to_its_exact_canonical_payload() -> None: + case = _case() + observation = case.match.observations[0] + wrong_reference, resolver = stored_artifact(b'{"plausible":"but-wrong"}', artifact_id="wrong-payload") + observations = (replace(observation, evidence=wrong_reference),) + case.match.observations[1:] + match = _redigest(replace(case.match, observations=observations)) + case = replace(case, resolver=InMemoryContentResolver({**case.resolver.artifacts, **resolver.artifacts})) + + with pytest.raises(FactMatchValidationError, match="canonical observation payload"): + _validate(case, match) + + +@pytest.mark.parametrize( + ("field", "value", "message"), + ( + ("pack_digest", "f" * 64, "pack digest"), + ("fact_id", "other.fact", "fact identity"), + ("match_digest", "f" * 64, "match digest"), + ), +) +def test_tampered_named_match_identities_are_refused(field: str, value: str, message: str) -> None: + case = _case() + if field == "pack_digest": + match = replace(case.match, pack_digest=value) + elif field == "fact_id": + match = replace(case.match, fact_id=value) + else: + match = replace(case.match, match_digest=value) + + with pytest.raises(FactMatchValidationError, match=message): + _validate(case, match) + + +def test_match_digest_binds_revision_behavior_lineage_location_and_observations() -> None: + case = _case() + original = case.match.match_digest + variants = ( + replace(case.match, matcher_revision="matcher-v4"), + replace(case.match, matcher_behavior_digest=canonical.content_digest("other-behavior")), + replace(case.match, matcher_lineage_manifest_digest="f" * 64), + replace(case.match, product_location=SourceLocation("src/product.py", 18, 8, "run")), + replace(case.match, observations=case.match.observations[:-1]), + ) + + assert all(derive_fact_match_digest(item) != original for item in variants) + + +def test_matcher_lineage_must_resolve_exact_behavior_and_current_receipt() -> None: + case = _case() + + with pytest.raises(FactMatchValidationError, match="lineage manifest"): + validate_fact_match( + case.pack, + case.pack.facts[0].fact_id, + case.match, + case.protected_population, + (), + case.custody, + case.custody_policy, + case.review_policy, + case.resolver, + ) + + +def test_matcher_lineage_requires_exact_matching_behavior_use() -> None: + case = _case() + lineage, custody, policy, protected = _matcher_lineage( + case.match.matcher_behavior_digest, + purpose=provenance.LineagePurpose.PROMPT, + ) + match = _redigest(replace(case.match, matcher_lineage_manifest_digest=lineage.manifest_digest)) + case = replace(case, match=match, lineage=lineage, custody=custody, custody_policy=policy, protected_population=protected) + + decision = _validate(case) + + assert decision.status is RelationStatus.UNRESOLVED + assert any(reason.startswith("missing_required_use:matching_behavior:") for reason in decision.reasons) + assert any(reason.startswith("unexpected_use:prompt:") for reason in decision.reasons) + + +def test_matcher_behavior_subject_substitution_is_refused() -> None: + case = _case() + match = _redigest(replace(case.match, matcher_behavior_digest=canonical.content_digest("substituted-behavior"))) + + with pytest.raises(FactMatchValidationError, match="lineage contract"): + _validate(case, match) + + +def test_lineage_receipt_must_remain_in_current_complete_custody_chain() -> None: + case = _case() + empty, policy, _ = custody_contract() + empty_custody = InMemoryPopulationCustody(empty, policy) + + with pytest.raises(FactMatchValidationError, match="current complete custody chain"): + _validate(replace(case, custody=empty_custody, custody_policy=policy)) + + +def _append_current_matching_use(case: MatchCase, *, labels_accessed: bool) -> provenance.PopulationUseReceipt: + current = case.custody.current_log(case.protected_population) + request = provenance.PopulationUseRequest( + case.protected_population, + case.lineage.arm_id, + provenance.LineagePurpose.MATCHING_BEHAVIOR, + labels_accessed, + case.match.matcher_behavior_digest, + current.log_digest, + ) + return case.custody.append(request, current.custodian_identity) + + +def test_omitted_later_protected_label_use_refuses_clean_match_manifest() -> None: + case = _case() + omitted = _append_current_matching_use(case, labels_accessed=True) + + decision = _validate(case) + + assert decision.status is RelationStatus.UNRESOLVED + assert not decision.effect_eligible + assert f"omitted_current_matching_use:{omitted.receipt_digest}" in decision.reasons + assert f"protected_population_labels:{case.protected_population}" in decision.reasons + + +def test_omitted_later_benign_matching_use_makes_manifest_incomplete() -> None: + case = _case() + omitted = _append_current_matching_use(case, labels_accessed=False) + + decision = _validate(case) + + assert decision.status is RelationStatus.UNRESOLVED + assert not decision.effect_eligible + assert decision.reasons == (f"omitted_current_matching_use:{omitted.receipt_digest}",) + + +def test_protected_population_label_influence_refuses_effect_eligibility() -> None: + case = _case(labels_accessed=True) + + decision = _validate(case) + + assert decision.status is RelationStatus.UNRESOLVED + assert not decision.effect_eligible + assert any(reason.startswith("protected_population_labels:") for reason in decision.reasons) + assert any(reason.startswith("label_access_forbidden:") for reason in decision.reasons) + + +def test_structural_observations_deep_freeze_canonical_values() -> None: + nested_shape: dict[str, canonical.JsonValue] = {"kind": "call"} + nested: dict[str, canonical.JsonValue] = {"shape": nested_shape} + reference, _ = stored_artifact(b"{}") + observation = StructuralObservation("shape", SourceLocation("a.py", 1, None, None), nested, reference) + nested_shape["kind"] = "changed" + + assert observation.observed == {"shape": {"kind": "call"}}