diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index 869b3a1..3da594f 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -42,7 +42,7 @@ - _Requirements: 1.2, 4.1, 4.2, 4.3, 5.2_ - _Boundary: Semantic Evidence_ - _Depends: 1.2_ -- [ ] 2.2 (P) Validate complete input and decision provenance +- [x] 2.2 (P) Validate complete input and decision provenance - Stamp analyzer, family contract, fact set, acquisition, configuration, population, and change-input identities. - Require the closed named-input role set and resolve every referenced input through the trusted artifact boundary. - Keep model and proof identities outside the immutable deterministic packet. diff --git a/src/openultrasast/core/provenance.py b/src/openultrasast/core/provenance.py index df8ca5a..8b4a472 100644 --- a/src/openultrasast/core/provenance.py +++ b/src/openultrasast/core/provenance.py @@ -1,23 +1,27 @@ -"""Provenance digest stamping for the detection core (Req 6). - -Every analysis is stamped with the analyzer revision, fact-set, acquisition-record, -configuration, population, and per-change input digests. A result whose digests -cannot be re-derived from the named inputs is refused, never accepted. -""" +"""Legacy stamps and strict v2 input/decision provenance contracts.""" from __future__ import annotations import hashlib import json +import re from collections.abc import Mapping -from typing import Any +from dataclasses import dataclass, fields +from enum import Enum + +from openultrasast.core import canonical +from openultrasast.core.canonical import ContentRef, ContentResolver + +NAMED_INPUTS_SCHEMA = "openultrasast.detection-core.named-inputs.v2" +PROVENANCE_POLICY_SCHEMA = "openultrasast.detection-core.provenance-policy.v2" +_DIGEST = re.compile(r"[0-9a-f]{64}") class ProvenanceMismatch(ValueError): - """A recorded digest does not match the re-derived input identity.""" + """A legacy recorded digest does not match its re-derived identity.""" -_DIGEST_FIELDS = ( +_LEGACY_DIGEST_FIELDS = ( "analyzer_revision", "fact_set_digest", "acquisition_record_digest", @@ -41,7 +45,7 @@ def stamp( population_digest: str, change_id: str, ) -> dict[str, str]: - """The provenance stamp of one analysis (Req 6.1).""" + """Preserve the v1 provenance stamp during migration.""" return { "analyzer_revision": analyzer_revision, "fact_set_digest": fact_set_digest, @@ -52,11 +56,255 @@ def stamp( } -def verify_stamp(recorded: Mapping[str, Any], rederived: Mapping[str, Any]) -> None: - """Refuse when any digest cannot be re-derived (Req 6.2).""" - for field_name in _DIGEST_FIELDS: +def verify_stamp(recorded: Mapping[str, object], rederived: Mapping[str, object]) -> None: + """Preserve v1 refusal when a legacy digest cannot be re-derived.""" + for field_name in _LEGACY_DIGEST_FIELDS: if recorded.get(field_name) != rederived.get(field_name): raise ProvenanceMismatch(f"{field_name}: recorded {recorded.get(field_name)!r} != rederived {rederived.get(field_name)!r}") -__all__ = ["ProvenanceMismatch", "stamp", "verify_stamp"] +class ProvenanceValidationError(ValueError): + """A provenance identity cannot be reproduced from its named inputs.""" + + +class ProvenanceInputRole(Enum): + ANALYZER = "analyzer" + FAMILY_CONTRACT = "family_contract" + FACT_SET = "fact_set" + ACQUISITION = "acquisition" + CONFIGURATION = "configuration" + POPULATION = "population" + CHANGE_INPUT = "change_input" + + +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_maximum(value: object) -> None: + if isinstance(value, bool) or not isinstance(value, int) or value <= 0: + raise ValueError("maximum_input_bytes must be a positive integer") + + +def _require_stores(stores: object) -> None: + if not isinstance(stores, tuple) or not stores: + raise ValueError("trusted_store_identities must be a nonempty tuple") + for store in stores: + _require_text(store, "trusted_store_identity") + if len(set(stores)) != len(stores): + raise ValueError("trusted_store_identities must be unique") + + +@dataclass(frozen=True) +class ProvenanceStamp: + analyzer_revision: str + family_contract_digest: str + fact_set_digest: str + acquisition_record_digest: str + configuration_digest: str + population_digest: str + change_input_digest: str + + def __post_init__(self) -> None: + for item in fields(self): + _require_digest(getattr(self, item.name), item.name) + + +@dataclass(frozen=True) +class ProvenanceTrustPolicy: + trusted_store_identities: tuple[str, ...] + maximum_input_bytes: int + policy_digest: str + + def __post_init__(self) -> None: + _require_stores(self.trusted_store_identities) + _require_maximum(self.maximum_input_bytes) + _require_digest(self.policy_digest, "policy_digest") + + +@dataclass(frozen=True) +class NamedInput: + role: ProvenanceInputRole + identity: str + content: ContentRef + + def __post_init__(self) -> None: + _require_digest(self.identity, "named input identity") + if not isinstance(self.content, ContentRef): + raise TypeError("named input content must be a ContentRef") + + +@dataclass(frozen=True) +class NamedInputs: + inputs: tuple[NamedInput, ...] + manifest_digest: str + + def __post_init__(self) -> None: + if not isinstance(self.inputs, tuple) or not all(isinstance(item, NamedInput) for item in self.inputs): + raise TypeError("named inputs must be a tuple of NamedInput values") + _require_digest(self.manifest_digest, "manifest_digest") + + +@dataclass(frozen=True) +class ModelCallIdentity: + provider: str + model: str + program_digest: str + prompt_digest: str + configuration_digest: str + adapter_revision: str + + def __post_init__(self) -> None: + for field_name in ("provider", "model", "adapter_revision"): + _require_text(getattr(self, field_name), field_name) + for field_name in ("program_digest", "prompt_digest", "configuration_digest"): + _require_digest(getattr(self, field_name), field_name) + + +@dataclass(frozen=True) +class DecisionProvenance: + model_identity: ModelCallIdentity | None + model_result_digest: str | None + model_declaration_digest: str | None + proof_record_digest: str | None + proof_producer_digest: str | None + lineage_manifest_digest: str | None + + def __post_init__(self) -> None: + if self.model_identity is not None and not isinstance(self.model_identity, ModelCallIdentity): + raise TypeError("model_identity must be a ModelCallIdentity") + for item in fields(self)[1:]: + value = getattr(self, item.name) + if value is not None: + _require_digest(value, item.name) + + +def _content_ref_value(reference: ContentRef) -> 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, + } + + +def _input_value(item: NamedInput) -> canonical.JsonValue: + if not isinstance(item.role, ProvenanceInputRole): + raise ProvenanceValidationError(f"unknown named input role: {item.role!r}") + return {"role": item.role.value, "identity": item.identity, "content": _content_ref_value(item.content)} + + +def derive_named_inputs_digest(inputs: tuple[NamedInput, ...]) -> str: + """Derive an order-independent identity over every named input reference.""" + ordered = tuple(sorted(inputs, key=lambda item: item.role.value)) + payload: canonical.JsonValue = { + "schema_version": NAMED_INPUTS_SCHEMA, + "inputs": tuple(_input_value(item) for item in ordered), + } + return canonical.content_digest(payload) + + +def derive_trust_policy_digest(trusted_store_identities: tuple[str, ...], maximum_input_bytes: int) -> str: + """Derive the trust and size policy identity.""" + _require_stores(trusted_store_identities) + _require_maximum(maximum_input_bytes) + payload: canonical.JsonValue = { + "schema_version": PROVENANCE_POLICY_SCHEMA, + "trusted_store_identities": tuple(sorted(trusted_store_identities)), + "maximum_input_bytes": maximum_input_bytes, + } + return canonical.content_digest(payload) + + +_STAMP_FIELD = { + ProvenanceInputRole.ANALYZER: "analyzer_revision", + ProvenanceInputRole.FAMILY_CONTRACT: "family_contract_digest", + ProvenanceInputRole.FACT_SET: "fact_set_digest", + ProvenanceInputRole.ACQUISITION: "acquisition_record_digest", + ProvenanceInputRole.CONFIGURATION: "configuration_digest", + ProvenanceInputRole.POPULATION: "population_digest", + ProvenanceInputRole.CHANGE_INPUT: "change_input_digest", +} + + +def _index_inputs(inputs: tuple[NamedInput, ...]) -> dict[ProvenanceInputRole, NamedInput]: + indexed: dict[ProvenanceInputRole, NamedInput] = {} + for item in inputs: + if not isinstance(item.role, ProvenanceInputRole): + raise ProvenanceValidationError(f"unknown named input role: {item.role!r}") + if item.role in indexed: + raise ProvenanceValidationError(f"duplicate named input role: {item.role.value}") + indexed[item.role] = item + missing = set(ProvenanceInputRole) - set(indexed) + if missing: + names = ", ".join(sorted(role.value for role in missing)) + raise ProvenanceValidationError(f"missing named input roles: {names}") + return indexed + + +def _validate_policy(trust: ProvenanceTrustPolicy) -> None: + expected = derive_trust_policy_digest(trust.trusted_store_identities, trust.maximum_input_bytes) + if trust.policy_digest != expected: + raise ProvenanceValidationError("provenance trust policy digest cannot be re-derived") + + +def _validate_manifest(named_inputs: NamedInputs) -> None: + expected = derive_named_inputs_digest(named_inputs.inputs) + if named_inputs.manifest_digest != expected: + raise ProvenanceValidationError("named input manifest digest cannot be re-derived") + + +def validate_provenance( + stamp: ProvenanceStamp, + named_inputs: NamedInputs, + resolver: ContentResolver, + trust: ProvenanceTrustPolicy, +) -> None: + """Resolve every closed-role input and reproduce each immutable stamp identity.""" + _validate_policy(trust) + indexed = _index_inputs(named_inputs.inputs) + _validate_manifest(named_inputs) + mismatches: list[str] = [] + for role in ProvenanceInputRole: + item = indexed[role] + canonical.resolve_content( + item.content, + resolver, + trusted_store_identities=trust.trusted_store_identities, + maximum_bytes=trust.maximum_input_bytes, + ) + if item.identity != item.content.content_digest: + raise ProvenanceValidationError(f"{role.value} named identity does not match resolved content") + field_name = _STAMP_FIELD[role] + if getattr(stamp, field_name) != item.identity: + mismatches.append(field_name) + if mismatches: + raise ProvenanceValidationError(f"stamp identities do not match named inputs: {', '.join(mismatches)}") + + +__all__ = [ + "DecisionProvenance", + "ModelCallIdentity", + "NAMED_INPUTS_SCHEMA", + "NamedInput", + "NamedInputs", + "PROVENANCE_POLICY_SCHEMA", + "ProvenanceInputRole", + "ProvenanceMismatch", + "ProvenanceStamp", + "ProvenanceTrustPolicy", + "ProvenanceValidationError", + "derive_named_inputs_digest", + "derive_trust_policy_digest", + "stamp", + "validate_provenance", + "verify_stamp", +] diff --git a/tests/test_core_provenance.py b/tests/test_core_provenance.py new file mode 100644 index 0000000..d16fbe5 --- /dev/null +++ b/tests/test_core_provenance.py @@ -0,0 +1,194 @@ +"""Complete input and decision provenance contracts (Req 10.1, 10.4).""" + +from __future__ import annotations + +from dataclasses import fields, replace +from typing import cast + +import pytest +from detection_core_fixtures import InMemoryContentResolver, stored_artifact + +from openultrasast.core import canonical, provenance + +STAMP_FIELDS = ( + "analyzer_revision", + "family_contract_digest", + "fact_set_digest", + "acquisition_record_digest", + "configuration_digest", + "population_digest", + "change_input_digest", +) + + +class CountingResolver: + def __init__(self, delegate: InMemoryContentResolver) -> None: + self.delegate = delegate + self.resolved: list[str] = [] + + def resolve(self, reference: canonical.ContentRef) -> canonical.ResolvedContent: + self.resolved.append(reference.artifact_id) + return self.delegate.resolve(reference) + + +def _contract() -> tuple[ + provenance.ProvenanceStamp, + provenance.NamedInputs, + InMemoryContentResolver, + provenance.ProvenanceTrustPolicy, +]: + entries: list[provenance.NamedInput] = [] + artifacts: dict[str, canonical.ResolvedContent] = {} + identities: dict[provenance.ProvenanceInputRole, str] = {} + for role in provenance.ProvenanceInputRole: + content = canonical.canonical_bytes({"schema_version": canonical.DETECTION_CORE_SCHEMA, "role": role.value}) + reference, resolver = stored_artifact(content, artifact_id=f"input-{role.value}") + entries.append(provenance.NamedInput(role=role, identity=reference.content_digest, content=reference)) + artifacts.update(resolver.artifacts) + identities[role] = reference.content_digest + inputs = tuple(reversed(entries)) + named = provenance.NamedInputs(inputs=inputs, manifest_digest=provenance.derive_named_inputs_digest(inputs)) + stamp = provenance.ProvenanceStamp( + analyzer_revision=identities[provenance.ProvenanceInputRole.ANALYZER], + family_contract_digest=identities[provenance.ProvenanceInputRole.FAMILY_CONTRACT], + fact_set_digest=identities[provenance.ProvenanceInputRole.FACT_SET], + acquisition_record_digest=identities[provenance.ProvenanceInputRole.ACQUISITION], + configuration_digest=identities[provenance.ProvenanceInputRole.CONFIGURATION], + population_digest=identities[provenance.ProvenanceInputRole.POPULATION], + change_input_digest=identities[provenance.ProvenanceInputRole.CHANGE_INPUT], + ) + stores = ("trusted-store",) + policy = provenance.ProvenanceTrustPolicy( + trusted_store_identities=stores, + maximum_input_bytes=1024, + policy_digest=provenance.derive_trust_policy_digest(stores, 1024), + ) + return stamp, named, InMemoryContentResolver(artifacts), policy + + +def _replace_stamp_digest(stamp: provenance.ProvenanceStamp, field: str, value: str) -> provenance.ProvenanceStamp: + values = {item.name: getattr(stamp, item.name) for item in fields(stamp)} + values[field] = value + return provenance.ProvenanceStamp(**values) + + +def test_every_closed_named_input_is_resolved_before_acceptance() -> None: + stamp, named, resolver, policy = _contract() + counting = CountingResolver(resolver) + + provenance.validate_provenance(stamp, named, counting, policy) + + assert set(counting.resolved) == {item.content.artifact_id for item in named.inputs} + assert len(counting.resolved) == len(provenance.ProvenanceInputRole) == 7 + + +@pytest.mark.parametrize("field", STAMP_FIELDS) +def test_every_stamp_identity_has_a_tamper_check(field: str) -> None: + stamp, named, resolver, policy = _contract() + tampered = _replace_stamp_digest(stamp, field, "0" * 64) + with pytest.raises(provenance.ProvenanceValidationError, match=field): + provenance.validate_provenance(tampered, named, resolver, policy) + + +def test_named_input_roles_are_closed_complete_and_unique() -> None: + stamp, named, resolver, policy = _contract() + missing_change = tuple(item for item in named.inputs if item.role is not provenance.ProvenanceInputRole.CHANGE_INPUT) + with pytest.raises(provenance.ProvenanceValidationError, match="missing.*change_input"): + provenance.validate_provenance(stamp, replace(named, inputs=missing_change), resolver, policy) + duplicate = named.inputs + (named.inputs[0],) + with pytest.raises(provenance.ProvenanceValidationError, match="duplicate"): + provenance.validate_provenance(stamp, replace(named, inputs=duplicate), resolver, policy) + invalid = replace(named.inputs[0], role=cast(provenance.ProvenanceInputRole, "model")) + with pytest.raises(provenance.ProvenanceValidationError, match="unknown.*role"): + provenance.validate_provenance(stamp, replace(named, inputs=(invalid, *named.inputs[1:])), resolver, policy) + + +def test_manifest_and_trust_policy_identities_are_rederived() -> None: + stamp, named, resolver, policy = _contract() + with pytest.raises(provenance.ProvenanceValidationError, match="manifest"): + provenance.validate_provenance(stamp, replace(named, manifest_digest="0" * 64), resolver, policy) + with pytest.raises(provenance.ProvenanceValidationError, match="policy"): + provenance.validate_provenance(stamp, named, resolver, replace(policy, policy_digest="0" * 64)) + + +def test_substituted_named_input_is_resolved_then_refused() -> None: + stamp, named, resolver, policy = _contract() + target = named.inputs[0] + replacement_ref, replacement_resolver = stored_artifact(b"substitute", artifact_id=target.content.artifact_id) + replacement = replace(target, content=replacement_ref) + replaced_inputs = (replacement, *named.inputs[1:]) + substituted = provenance.NamedInputs( + inputs=replaced_inputs, + manifest_digest=provenance.derive_named_inputs_digest(replaced_inputs), + ) + artifacts = dict(resolver.artifacts) + artifacts[target.content.artifact_id] = replacement_resolver.artifacts[target.content.artifact_id] + counting = CountingResolver(InMemoryContentResolver(artifacts)) + + with pytest.raises(provenance.ProvenanceValidationError, match="named identity"): + provenance.validate_provenance(stamp, substituted, counting, policy) + assert target.content.artifact_id in counting.resolved + + +def test_unreadable_named_input_is_never_treated_as_empty() -> None: + stamp, named, _, policy = _contract() + assert all(item.content.byte_length > 0 for item in named.inputs) + with pytest.raises(canonical.ArtifactResolutionError, match="not found"): + provenance.validate_provenance(stamp, named, InMemoryContentResolver({}), policy) + + +def test_named_inputs_use_receipt_trust_and_size_boundary() -> None: + stamp, named, resolver, policy = _contract() + with pytest.raises(canonical.ArtifactValidationError, match="trusted"): + provenance.validate_provenance( + stamp, + named, + resolver, + replace( + policy, + trusted_store_identities=("other-store",), + policy_digest=provenance.derive_trust_policy_digest(("other-store",), 1024), + ), + ) + tiny = replace( + policy, + maximum_input_bytes=1, + policy_digest=provenance.derive_trust_policy_digest(policy.trusted_store_identities, 1), + ) + with pytest.raises(canonical.ArtifactValidationError, match="maximum"): + provenance.validate_provenance(stamp, named, resolver, tiny) + + +@pytest.mark.parametrize("value", ["", 7, None]) +def test_textual_and_digest_contracts_are_runtime_checked(value: object) -> None: + stamp, named, _, policy = _contract() + with pytest.raises((ValueError, canonical.DigestFormatError)): + replace(named.inputs[0], identity=cast(str, value)) + with pytest.raises(ValueError): + replace(policy, trusted_store_identities=(cast(str, value),)) + with pytest.raises(canonical.DigestFormatError): + replace(stamp, population_digest=cast(str, value)) + + +def test_model_and_proof_identities_are_decision_only() -> None: + assert not any("model" in item.name or "proof" in item.name for item in fields(provenance.ProvenanceStamp)) + decision_fields = {item.name for item in fields(provenance.DecisionProvenance)} + assert {"model_identity", "model_result_digest", "proof_record_digest", "proof_producer_digest"} <= decision_fields + digest = canonical.content_digest("identity") + identity = provenance.ModelCallIdentity( + provider="provider", + model="model", + program_digest=digest, + prompt_digest=digest, + configuration_digest=digest, + adapter_revision="adapter-v1", + ) + decision = provenance.DecisionProvenance(identity, digest, digest, digest, digest, digest) + assert decision.model_identity == identity + + +def test_named_input_manifest_is_order_independent_but_role_sensitive() -> None: + _, named, _, _ = _contract() + assert provenance.derive_named_inputs_digest(named.inputs) == provenance.derive_named_inputs_digest(tuple(reversed(named.inputs))) + swapped = replace(named.inputs[0], role=named.inputs[1].role) + assert provenance.derive_named_inputs_digest((swapped, *named.inputs[1:])) != named.manifest_digest