From d4fcde615b9da30f5f7042e3fb556361cad77d22 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sat, 31 Oct 2026 17:52:25 +0100 Subject: [PATCH] feat(detection-core): validate differential proof attachment --- .kiro/specs/detection-core/tasks.md | 2 +- src/openultrasast/core/_proof_apply.py | 142 ++++++++ src/openultrasast/core/_proof_attestation.py | 56 +++ src/openultrasast/core/proof.py | 345 ++++++++++++++++++ tests/test_core_proof.py | 351 +++++++++++++++++++ 5 files changed, 895 insertions(+), 1 deletion(-) create mode 100644 src/openultrasast/core/_proof_apply.py create mode 100644 src/openultrasast/core/_proof_attestation.py create mode 100644 src/openultrasast/core/proof.py create mode 100644 tests/test_core_proof.py diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index d7741a2..4b800c2 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -202,7 +202,7 @@ - _Requirements: 2.2, 2.3, 7.1, 7.2, 7.3, 7.4, 14.4_ - _Boundary: Verdict and Proof_ - _Depends: 2.5, 3.1_ -- [ ] 7.4 Validate differential proof attachment without enabling rollout +- [x] 7.4 Validate differential proof attachment without enabling rollout - Re-derive reviewed producer, sandbox, oracle, affected/safe side, control, witness, and verification identities. - Resolve all proof artifacts under the supplied producer, reviewer, store, and byte-limit trust policy. - Upgrade basis only for demonstrated affected-only behavior tied to the same family, packet, location, and change/control. diff --git a/src/openultrasast/core/_proof_apply.py b/src/openultrasast/core/_proof_apply.py new file mode 100644 index 0000000..da20095 --- /dev/null +++ b/src/openultrasast/core/_proof_apply.py @@ -0,0 +1,142 @@ +"""Differential-proof application against one exact defect claim.""" + +from __future__ import annotations + +from dataclasses import replace + +from openultrasast.core import canonical, provenance +from openultrasast.core._proof_attestation import proof_attestation_bytes, validate_proof_artifacts +from openultrasast.core.evidence import EvidenceBasis, EvidenceRecord, derive_packet_digest +from openultrasast.core.proof import ( + DifferentialProofRecord, + ProofAttachment, + ProofOutcome, + ProofTrustPolicy, + ProofValidationError, + ProofVerificationReceipt, + derive_producer_identity_digest, +) +from openultrasast.core.verdict_v2 import DefectClaim + + +def _validate_claim_binding(claim: DefectClaim, attachment: ProofAttachment) -> None: + try: + claim.__post_init__() + attachment.__post_init__() + attachment.proof.__post_init__() + attachment.proof.producer.__post_init__() + attachment.proof.affected.__post_init__() + attachment.proof.safe.__post_init__() + attachment.proof.control.__post_init__() + attachment.verification.__post_init__() + except (TypeError, ValueError) as exc: + raise ProofValidationError(f"proof attachment identity validation failed: {exc}") from exc + record = claim.record + if record.decision.basis is EvidenceBasis.EXECUTION_BACKED: + raise ProofValidationError("an execution-backed claim cannot replace its proof attachment") + if attachment.family_id != record.packet.family_id or attachment.packet_digest != derive_packet_digest(record.packet): + raise ProofValidationError("proof family or packet does not match the defect claim") + if attachment.location != claim.location: + raise ProofValidationError("proof attachment does not match the exact claim location") + + +def _validate_change_and_outcome(record: EvidenceRecord, proof: DifferentialProofRecord) -> None: + change = record.packet.provenance.change_input_digest + if proof.affected.input_digest != change or proof.control.affected_change_input_digest != change: + raise ProofValidationError("affected proof input does not match the evidence change input") + if proof.safe.input_digest != proof.control.safe_input_digest: + raise ProofValidationError("safe proof input does not match the reviewed control") + if proof.affected.input_digest == proof.safe.input_digest: + raise ProofValidationError("affected and safe proof inputs must be distinct") + if proof.outcome is not ProofOutcome.DEMONSTRATED or not proof.affected.observed or proof.safe.observed: + raise ProofValidationError("only demonstrated affected-only behavior can upgrade evidence basis") + + +def _validate_trust( + proof: DifferentialProofRecord, + verification: ProofVerificationReceipt, + policy: ProofTrustPolicy, +) -> str: + try: + policy.__post_init__() + except (TypeError, ValueError) as exc: + raise ProofValidationError(f"proof policy validation failed: {exc}") from exc + producer_digest = derive_producer_identity_digest(proof.producer) + if producer_digest not in policy.reviewed_producer_digests: + raise ProofValidationError("proof producer is not independently reviewed") + reviewer = proof.control.reviewer_identity + if reviewer not in policy.reviewed_control_reviewer_digests: + raise ProofValidationError("control reviewer is not admitted by the proof policy") + producer_ids = (proof.producer.producer_id, canonical.content_digest(proof.producer.producer_id), producer_digest) + if reviewer in producer_ids: + raise ProofValidationError("proof producer cannot supply a self-asserted safe control") + expected = ( + producer_digest, + proof.record_digest, + proof.affected.result_digest, + proof.safe.result_digest, + proof.witness.content_digest, + ) + actual = ( + verification.producer_identity_digest, + verification.proof_record_digest, + verification.affected_result_digest, + verification.safe_result_digest, + verification.witness_content_digest, + ) + if actual != expected: + raise ProofValidationError("proof verification identities do not match the proof record") + return producer_digest + + +def _resolve_artifacts( + proof: DifferentialProofRecord, + verification: ProofVerificationReceipt, + policy: ProofTrustPolicy, + resolver: canonical.ContentResolver, +) -> None: + try: + expected = proof_attestation_bytes( + verification.producer_identity_digest, + verification.proof_record_digest, + verification.affected_result_digest, + verification.safe_result_digest, + verification.witness_content_digest, + ) + validate_proof_artifacts( + proof.control.control_manifest, + proof.witness, + verification.attestation, + expected, + resolver, + policy.trusted_store_identities, + policy.maximum_artifact_bytes, + ) + except (canonical.ArtifactResolutionError, canonical.ArtifactValidationError, ValueError) as exc: + raise ProofValidationError(f"proof artifact validation failed: {exc}") from exc + + +def attach_differential_proof( + claim: DefectClaim, + attachment: ProofAttachment, + policy: ProofTrustPolicy, + resolver: canonical.ContentResolver, +) -> DefectClaim: + if not isinstance(claim, DefectClaim) or not isinstance(attachment, ProofAttachment): + raise ProofValidationError("proof application requires a DefectClaim and ProofAttachment") + if not isinstance(policy, ProofTrustPolicy): + raise ProofValidationError("proof application requires a ProofTrustPolicy") + _validate_claim_binding(claim, attachment) + _validate_change_and_outcome(claim.record, attachment.proof) + producer_digest = _validate_trust(attachment.proof, attachment.verification, policy) + _resolve_artifacts(attachment.proof, attachment.verification, policy, resolver) + decision = replace( + claim.record.decision, + basis=EvidenceBasis.EXECUTION_BACKED, + provenance=provenance.DecisionProvenance(None, None, None, attachment.proof.record_digest, producer_digest, None), + ) + record = EvidenceRecord(claim.record.packet, decision, "") + return DefectClaim(record, EvidenceBasis.EXECUTION_BACKED, claim.location, claim.change_anchor) + + +__all__ = ["attach_differential_proof"] diff --git a/src/openultrasast/core/_proof_attestation.py b/src/openultrasast/core/_proof_attestation.py new file mode 100644 index 0000000..5f5989b --- /dev/null +++ b/src/openultrasast/core/_proof_attestation.py @@ -0,0 +1,56 @@ +"""Canonical proof-attestation bytes and bounded artifact resolution.""" + +from __future__ import annotations + +from collections.abc import Collection + +from openultrasast.core import canonical + +PROOF_ATTESTATION_SCHEMA = "openultrasast.detection-core.proof-attestation.v2" + + +def proof_attestation_bytes( + producer_identity_digest: str, + proof_record_digest: str, + affected_result_digest: str, + safe_result_digest: str, + witness_content_digest: str, +) -> bytes: + payload: dict[str, canonical.JsonValue] = { + "schema_version": PROOF_ATTESTATION_SCHEMA, + "producer_identity_digest": producer_identity_digest, + "proof_record_digest": proof_record_digest, + "affected_result_digest": affected_result_digest, + "safe_result_digest": safe_result_digest, + "witness_content_digest": witness_content_digest, + } + return canonical.canonical_record_bytes(payload, known_schema_versions=(PROOF_ATTESTATION_SCHEMA,)) + + +def validate_proof_artifacts( + control: canonical.ContentRef, + witness: canonical.ContentRef, + attestation: canonical.ContentRef, + expected_attestation: bytes, + resolver: canonical.ContentResolver, + trusted_stores: Collection[str], + maximum_bytes: int, +) -> None: + for reference in (control, witness): + canonical.resolve_content( + reference, + resolver, + trusted_store_identities=trusted_stores, + maximum_bytes=maximum_bytes, + ) + resolved = canonical.resolve_content( + attestation, + resolver, + trusted_store_identities=trusted_stores, + maximum_bytes=maximum_bytes, + ) + if resolved != expected_attestation: + raise canonical.ArtifactValidationError("proof attestation does not bind the exact verification identities") + + +__all__ = ["PROOF_ATTESTATION_SCHEMA", "proof_attestation_bytes", "validate_proof_artifacts"] diff --git a/src/openultrasast/core/proof.py b/src/openultrasast/core/proof.py new file mode 100644 index 0000000..f996c86 --- /dev/null +++ b/src/openultrasast/core/proof.py @@ -0,0 +1,345 @@ +"""Canonical differential-proof validation for execution-backed evidence.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from enum import Enum + +from openultrasast.core import canonical +from openultrasast.core.semantics import SourceLocation +from openultrasast.core.verdict_v2 import DefectClaim + +DIFFERENTIAL_PROOF_SCHEMA = "openultrasast.detection-core.differential-proof.v2" +_DIGEST = re.compile(r"[0-9a-f]{64}") + + +class ProofValidationError(ValueError): + """A proof attachment cannot reproduce execution-backed evidence.""" + + +class ProofOutcome(Enum): + DEMONSTRATED = "demonstrated" + INCONCLUSIVE = "inconclusive" + FAILED = "failed" + + +def _digest(value: object, field: str) -> None: + if not isinstance(value, str) or _DIGEST.fullmatch(value) is None: + raise ProofValidationError(f"{field} must be a full lowercase SHA-256 digest") + + +def _text(value: object, field: str) -> None: + if not isinstance(value, str) or not value: + raise ProofValidationError(f"{field} must be nonempty text") + + +def _positive(value: object, field: str) -> None: + if isinstance(value, bool) or not isinstance(value, int) or value <= 0: + raise ProofValidationError(f"{field} must be a positive integer") + + +def _ref_value(value: canonical.ContentRef) -> canonical.JsonValue: + return { + "artifact_id": value.artifact_id, + "content_digest": value.content_digest, + "media_type": value.media_type, + "byte_length": value.byte_length, + "retention_policy_digest": value.retention_policy_digest, + "storage_receipt_digest": value.storage_receipt_digest, + } + + +def _location_value(value: SourceLocation) -> canonical.JsonValue: + return {"path": value.path, "line": value.line, "column": value.column, "symbol": value.symbol} + + +@dataclass(frozen=True) +class ProofProducerIdentity: + producer_id: str + verifier_revision: str + sandbox_profile_digest: str + oracle_contract_digest: str + review_digest: str + + def __post_init__(self) -> None: + _text(self.producer_id, "producer_id") + _text(self.verifier_revision, "verifier_revision") + for value, field in ( + (self.sandbox_profile_digest, "sandbox_profile_digest"), + (self.oracle_contract_digest, "oracle_contract_digest"), + (self.review_digest, "review_digest"), + ): + _digest(value, field) + + +def _producer_value(value: ProofProducerIdentity) -> canonical.JsonValue: + return { + "producer_id": value.producer_id, + "verifier_revision": value.verifier_revision, + "sandbox_profile_digest": value.sandbox_profile_digest, + "oracle_contract_digest": value.oracle_contract_digest, + "review_digest": value.review_digest, + } + + +def derive_producer_identity_digest(value: ProofProducerIdentity) -> str: + value.__post_init__() + return canonical.content_digest(_producer_value(value)) + + +@dataclass(frozen=True) +class ProofSideResult: + input_digest: str + observed: bool + observation_digest: str + run_count: int + result_digest: str + + def __post_init__(self) -> None: + _digest(self.input_digest, "proof side input_digest") + if not isinstance(self.observed, bool): + raise ProofValidationError("proof side observed must be boolean") + _digest(self.observation_digest, "proof side observation_digest") + _positive(self.run_count, "proof side run_count") + expected = derive_proof_side_result_digest(self) + if not self.result_digest: + object.__setattr__(self, "result_digest", expected) + elif self.result_digest != expected: + raise ProofValidationError("proof side result digest cannot be re-derived") + + +def _side_value(value: ProofSideResult) -> dict[str, canonical.JsonValue]: + return { + "input_digest": value.input_digest, + "observed": value.observed, + "observation_digest": value.observation_digest, + "run_count": value.run_count, + } + + +def derive_proof_side_result_digest(value: ProofSideResult) -> str: + return canonical.content_digest(_side_value(value)) + + +@dataclass(frozen=True) +class ControlBinding: + affected_change_input_digest: str + safe_input_digest: str + control_manifest: canonical.ContentRef + reviewer_identity: str + binding_digest: str + + def __post_init__(self) -> None: + _digest(self.affected_change_input_digest, "affected_change_input_digest") + _digest(self.safe_input_digest, "safe_input_digest") + if not isinstance(self.control_manifest, canonical.ContentRef): + raise ProofValidationError("control_manifest must be a ContentRef") + _digest(self.reviewer_identity, "control reviewer_identity") + expected = derive_control_binding_digest(self) + if not self.binding_digest: + object.__setattr__(self, "binding_digest", expected) + elif self.binding_digest != expected: + raise ProofValidationError("control binding digest cannot be re-derived") + + +def _control_value(value: ControlBinding) -> dict[str, canonical.JsonValue]: + return { + "affected_change_input_digest": value.affected_change_input_digest, + "safe_input_digest": value.safe_input_digest, + "control_manifest": _ref_value(value.control_manifest), + "reviewer_identity": value.reviewer_identity, + } + + +def derive_control_binding_digest(value: ControlBinding) -> str: + return canonical.content_digest(_control_value(value)) + + +@dataclass(frozen=True) +class DifferentialProofRecord: + schema_version: str + producer: ProofProducerIdentity + affected: ProofSideResult + safe: ProofSideResult + control: ControlBinding + witness: canonical.ContentRef + outcome: ProofOutcome + record_digest: str + + def __post_init__(self) -> None: + if self.schema_version != DIFFERENTIAL_PROOF_SCHEMA: + raise ProofValidationError("unknown differential proof schema") + if not isinstance(self.producer, ProofProducerIdentity): + raise ProofValidationError("producer must use ProofProducerIdentity") + if not isinstance(self.affected, ProofSideResult) or not isinstance(self.safe, ProofSideResult): + raise ProofValidationError("proof sides must use ProofSideResult") + if not isinstance(self.control, ControlBinding) or not isinstance(self.witness, canonical.ContentRef): + raise ProofValidationError("proof control and witness have invalid types") + if not isinstance(self.outcome, ProofOutcome): + raise ProofValidationError("proof outcome is invalid") + expected = derive_differential_proof_record_digest(self) + if not self.record_digest: + object.__setattr__(self, "record_digest", expected) + elif self.record_digest != expected: + raise ProofValidationError("differential proof digest cannot be re-derived") + + +def _proof_value(value: DifferentialProofRecord) -> canonical.JsonValue: + return { + "schema_version": value.schema_version, + "producer": _producer_value(value.producer), + "affected": {**_side_value(value.affected), "result_digest": value.affected.result_digest}, + "safe": {**_side_value(value.safe), "result_digest": value.safe.result_digest}, + "control": {**_control_value(value.control), "binding_digest": value.control.binding_digest}, + "witness": _ref_value(value.witness), + "outcome": value.outcome.value, + } + + +def derive_differential_proof_record_digest(value: DifferentialProofRecord) -> str: + return canonical.content_digest(_proof_value(value)) + + +@dataclass(frozen=True) +class ProofVerificationReceipt: + producer_identity_digest: str + proof_record_digest: str + affected_result_digest: str + safe_result_digest: str + witness_content_digest: str + attestation: canonical.ContentRef + receipt_digest: str + + def __post_init__(self) -> None: + for value, field in ( + (self.producer_identity_digest, "producer_identity_digest"), + (self.proof_record_digest, "proof_record_digest"), + (self.affected_result_digest, "affected_result_digest"), + (self.safe_result_digest, "safe_result_digest"), + (self.witness_content_digest, "witness_content_digest"), + ): + _digest(value, field) + if not isinstance(self.attestation, canonical.ContentRef): + raise ProofValidationError("attestation must be a ContentRef") + expected = derive_proof_verification_receipt_digest(self) + if not self.receipt_digest: + object.__setattr__(self, "receipt_digest", expected) + elif self.receipt_digest != expected: + raise ProofValidationError("proof verification receipt digest cannot be re-derived") + + +def _verification_value(value: ProofVerificationReceipt) -> canonical.JsonValue: + return { + "producer_identity_digest": value.producer_identity_digest, + "proof_record_digest": value.proof_record_digest, + "affected_result_digest": value.affected_result_digest, + "safe_result_digest": value.safe_result_digest, + "witness_content_digest": value.witness_content_digest, + "attestation": _ref_value(value.attestation), + } + + +def derive_proof_verification_receipt_digest(value: ProofVerificationReceipt) -> str: + return canonical.content_digest(_verification_value(value)) + + +@dataclass(frozen=True) +class ProofTrustPolicy: + reviewed_producer_digests: tuple[str, ...] + reviewed_control_reviewer_digests: tuple[str, ...] + trusted_store_identities: tuple[str, ...] + maximum_artifact_bytes: int + policy_digest: str + + def __post_init__(self) -> None: + for values, field in ( + (self.reviewed_producer_digests, "reviewed producer"), + (self.reviewed_control_reviewer_digests, "reviewed control reviewer"), + ): + if not values: + raise ProofValidationError(f"{field} identities must be nonempty") + for value in values: + _digest(value, field) + canonical.require_unique_identities(values, kind=field) + if not self.trusted_store_identities or not all(isinstance(value, str) and value for value in self.trusted_store_identities): + raise ProofValidationError("trusted store identities must be nonempty text") + canonical.require_unique_identities(self.trusted_store_identities, kind="trusted store") + _positive(self.maximum_artifact_bytes, "maximum_artifact_bytes") + expected = derive_proof_trust_policy_digest(self) + if not self.policy_digest: + object.__setattr__(self, "policy_digest", expected) + elif self.policy_digest != expected: + raise ProofValidationError("proof trust policy digest cannot be re-derived") + + +def derive_proof_trust_policy_digest(value: ProofTrustPolicy) -> str: + payload: canonical.JsonValue = { + "reviewed_producer_digests": tuple(sorted(value.reviewed_producer_digests)), + "reviewed_control_reviewer_digests": tuple(sorted(value.reviewed_control_reviewer_digests)), + "trusted_store_identities": tuple(sorted(value.trusted_store_identities)), + "maximum_artifact_bytes": value.maximum_artifact_bytes, + } + return canonical.content_digest(payload) + + +@dataclass(frozen=True) +class ProofAttachment: + proof_id: str + family_id: str + packet_digest: str + location: SourceLocation + proof: DifferentialProofRecord + verification: ProofVerificationReceipt + + def __post_init__(self) -> None: + _text(self.family_id, "proof family_id") + _digest(self.packet_digest, "proof packet_digest") + if not isinstance(self.location, SourceLocation): + raise ProofValidationError("proof location must be a SourceLocation") + if not isinstance(self.proof, DifferentialProofRecord) or not isinstance(self.verification, ProofVerificationReceipt): + raise ProofValidationError("proof attachment has invalid proof or verification") + expected = derive_proof_attachment_id(self) + if not self.proof_id: + object.__setattr__(self, "proof_id", expected) + elif self.proof_id != expected: + raise ProofValidationError("proof attachment identity cannot be re-derived") + + +def derive_proof_attachment_id(value: ProofAttachment) -> str: + payload: canonical.JsonValue = { + "family_id": value.family_id, + "packet_digest": value.packet_digest, + "location": _location_value(value.location), + "proof_record_digest": value.proof.record_digest, + "verification_receipt_digest": value.verification.receipt_digest, + } + return canonical.content_digest(payload) + + +def attach_differential_proof( + claim: DefectClaim, + attachment: ProofAttachment, + policy: ProofTrustPolicy, + resolver: canonical.ContentResolver, +) -> DefectClaim: + """Validate and attach affected-only proof without granting rollout permission.""" + from openultrasast.core._proof_apply import attach_differential_proof as apply_proof + + return apply_proof(claim, attachment, policy, resolver) + + +__all__ = [ + "ControlBinding", + "DIFFERENTIAL_PROOF_SCHEMA", + "DifferentialProofRecord", + "ProofAttachment", + "ProofOutcome", + "ProofProducerIdentity", + "ProofSideResult", + "ProofTrustPolicy", + "ProofValidationError", + "ProofVerificationReceipt", + "attach_differential_proof", + "derive_producer_identity_digest", +] diff --git a/tests/test_core_proof.py b/tests/test_core_proof.py new file mode 100644 index 0000000..bfa7539 --- /dev/null +++ b/tests/test_core_proof.py @@ -0,0 +1,351 @@ +"""Differential proof attachment and execution-backed basis tests.""" + +from __future__ import annotations + +from dataclasses import dataclass, replace + +import pytest +from detection_core_fixtures import InMemoryContentResolver, stored_artifact + +from openultrasast.core import canonical, provenance +from openultrasast.core._proof_attestation import proof_attestation_bytes +from openultrasast.core.evidence import ( + EVIDENCE_PACKET_SCHEMA, + AnalyzedScope, + Decision, + EvidenceBasis, + EvidencePacket, + EvidenceRecord, + NeutralOutcome, + SourceRegion, + derive_packet_digest, +) +from openultrasast.core.proof import ( + DIFFERENTIAL_PROOF_SCHEMA, + ControlBinding, + DifferentialProofRecord, + ProofAttachment, + ProofOutcome, + ProofProducerIdentity, + ProofSideResult, + ProofTrustPolicy, + ProofValidationError, + ProofVerificationReceipt, + attach_differential_proof, + derive_producer_identity_digest, +) +from openultrasast.core.semantics import ( + EvidenceProvenance, + FieldKind, + FieldValue, + ProvenanceKind, + RelationKind, + RelationStatus, + SemanticNode, + SourceLocation, + make_observation, + make_relation, +) +from openultrasast.core.verdict_v2 import DefectClaim + + +def _digest(value: str) -> str: + return canonical.content_digest(value) + + +def _record() -> EvidenceRecord: + location = SourceLocation("src/app.py", 12, 4, "call") + relation = make_relation( + kind=RelationKind.CALL, + role="flow", + source=SemanticNode("source", "source", SourceLocation("src/app.py", 10, 2, "source")), + target=SemanticNode("target", "target", SourceLocation("src/app.py", 11, 2, "target")), + location=location, + status=RelationStatus.COMPLETE, + conditions=(), + reasons=(), + provenance=EvidenceProvenance(ProvenanceKind.PRODUCT_CODE, _digest("source"), None, None, None), + ) + scope = AnalyzedScope("base", "head", _digest("change"), (SourceRegion("src/app.py", 1, 20),)) + stamp = provenance.ProvenanceStamp( + _digest("analyzer"), + _digest("family-contract"), + _digest("facts"), + _digest("acquisition"), + _digest("configuration"), + _digest("population"), + _digest("change-input"), + ) + packet = EvidencePacket( + EVIDENCE_PACKET_SCHEMA, + "reference-flow", + stamp.family_contract_digest, + _digest("object"), + "python", + scope, + (relation,), + (), + (FieldValue("name", FieldKind.TEXT, "read"),), + (), + stamp, + ) + decision = Decision( + "bad", + NeutralOutcome.VIOLATION, + EvidenceBasis.DETERMINISTIC, + scope, + (), + derive_packet_digest(packet), + provenance.DecisionProvenance(None, None, None, None, None, None), + ) + return EvidenceRecord(packet, decision, "") + + +def _claim(record: EvidenceRecord) -> DefectClaim: + return DefectClaim(record, record.decision.basis, record.packet.transitions[0].location, record.packet.scope.regions[0]) + + +@dataclass(frozen=True) +class ProofCase: + record: EvidenceRecord + attachment: ProofAttachment + policy: ProofTrustPolicy + resolver: InMemoryContentResolver + + +def _case() -> ProofCase: + record = _record() + reviewer = _digest("control-reviewer") + control_ref, control_resolver = stored_artifact(b'{"control":"safe"}', artifact_id="control") + witness_ref, witness_resolver = stored_artifact(b'{"affected":true,"safe":false}', artifact_id="witness") + producer = ProofProducerIdentity( + "search-with-proof", + "verifier-v4", + _digest("sandbox-profile"), + _digest("oracle-contract"), + _digest("producer-review"), + ) + affected = ProofSideResult(record.packet.provenance.change_input_digest, True, _digest("affected-observation"), 2, "") + safe_input = _digest("safe-control-input") + safe = ProofSideResult(safe_input, False, _digest("safe-observation"), 2, "") + control = ControlBinding(affected.input_digest, safe_input, control_ref, reviewer, "") + proof = DifferentialProofRecord( + DIFFERENTIAL_PROOF_SCHEMA, + producer, + affected, + safe, + control, + witness_ref, + ProofOutcome.DEMONSTRATED, + "", + ) + attestation_ref, attestation_resolver = stored_artifact( + proof_attestation_bytes( + derive_producer_identity_digest(producer), + proof.record_digest, + affected.result_digest, + safe.result_digest, + witness_ref.content_digest, + ), + artifact_id="attestation", + ) + verification = ProofVerificationReceipt( + derive_producer_identity_digest(producer), + proof.record_digest, + affected.result_digest, + safe.result_digest, + witness_ref.content_digest, + attestation_ref, + "", + ) + attachment = ProofAttachment( + "", + record.packet.family_id, + derive_packet_digest(record.packet), + record.packet.transitions[0].location, + proof, + verification, + ) + policy = ProofTrustPolicy( + (derive_producer_identity_digest(producer),), + (reviewer,), + ("trusted-store",), + 4096, + "", + ) + artifacts = { + **control_resolver.artifacts, + **witness_resolver.artifacts, + **attestation_resolver.artifacts, + } + return ProofCase(record, attachment, policy, InMemoryContentResolver(artifacts)) + + +def test_demonstrated_affected_only_proof_upgrades_basis_without_rollout_authority() -> None: + case = _case() + claim = _claim(case.record) + derived = attach_differential_proof(claim, case.attachment, case.policy, case.resolver) + assert derived is not claim and derived.record.packet is case.record.packet + assert derived.location == claim.location and derived.change_anchor == claim.change_anchor + assert derived.record.decision.outcome is NeutralOutcome.VIOLATION + assert derived.record.decision.family_code == claim.record.decision.family_code + assert derived.record.decision.bounded_scope is claim.record.decision.bounded_scope + assert derived.record.decision.reasons == claim.record.decision.reasons + assert derived.record.decision.basis is EvidenceBasis.EXECUTION_BACKED + assert derived.record.decision.provenance.proof_record_digest == case.attachment.proof.record_digest + assert derived.record.decision.provenance.proof_producer_digest == derive_producer_identity_digest(case.attachment.proof.producer) + assert not hasattr(derived, "rollout_authorized") + + +@pytest.mark.parametrize( + ("outcome", "affected", "safe"), + ( + (ProofOutcome.INCONCLUSIVE, True, False), + (ProofOutcome.FAILED, True, False), + (ProofOutcome.DEMONSTRATED, True, True), + (ProofOutcome.DEMONSTRATED, False, False), + ), +) +def test_only_demonstrated_affected_only_behavior_is_eligible(outcome: ProofOutcome, affected: bool, safe: bool) -> None: + case = _case() + affected_side = replace(case.attachment.proof.affected, observed=affected, result_digest="") + safe_side = replace(case.attachment.proof.safe, observed=safe, result_digest="") + proof = replace(case.attachment.proof, affected=affected_side, safe=safe_side, outcome=outcome, record_digest="") + verification = replace( + case.attachment.verification, + proof_record_digest=proof.record_digest, + affected_result_digest=affected_side.result_digest, + safe_result_digest=safe_side.result_digest, + receipt_digest="", + ) + attachment = replace(case.attachment, proof=proof, verification=verification, proof_id="") + with pytest.raises(ProofValidationError, match="affected-only"): + attach_differential_proof(_claim(case.record), attachment, case.policy, case.resolver) + + +@pytest.mark.parametrize("field", ("family", "packet", "location", "affected", "safe")) +def test_attachment_must_match_family_packet_location_and_change_control(field: str) -> None: + case = _case() + attachment = case.attachment + if field == "family": + attachment = replace(attachment, family_id="other-family", proof_id="") + elif field == "packet": + attachment = replace(attachment, packet_digest=_digest("other-packet"), proof_id="") + elif field == "location": + attachment = replace(attachment, location=SourceLocation("src/other.py", 1, None, None), proof_id="") + elif field == "affected": + side = replace(attachment.proof.affected, input_digest=_digest("other-change"), result_digest="") + proof = replace(attachment.proof, affected=side, record_digest="") + receipt = replace( + attachment.verification, + proof_record_digest=proof.record_digest, + affected_result_digest=side.result_digest, + receipt_digest="", + ) + attachment = replace(attachment, proof=proof, verification=receipt, proof_id="") + else: + control = replace(attachment.proof.control, safe_input_digest=_digest("other-safe"), binding_digest="") + proof = replace(attachment.proof, control=control, record_digest="") + receipt = replace(attachment.verification, proof_record_digest=proof.record_digest, receipt_digest="") + attachment = replace(attachment, proof=proof, verification=receipt, proof_id="") + with pytest.raises(ProofValidationError): + attach_differential_proof(_claim(case.record), attachment, case.policy, case.resolver) + + +def test_unreviewed_or_self_asserted_producer_and_control_are_refused() -> None: + case = _case() + with pytest.raises(ProofValidationError, match="producer"): + attach_differential_proof( + _claim(case.record), + case.attachment, + replace(case.policy, reviewed_producer_digests=(_digest("unknown-producer"),), policy_digest=""), + case.resolver, + ) + self_asserted = replace( + case.attachment.proof.control, + reviewer_identity=derive_producer_identity_digest(case.attachment.proof.producer), + binding_digest="", + ) + proof = replace(case.attachment.proof, control=self_asserted, record_digest="") + receipt = replace(case.attachment.verification, proof_record_digest=proof.record_digest, receipt_digest="") + attachment = replace(case.attachment, proof=proof, verification=receipt, proof_id="") + policy = replace( + case.policy, + reviewed_control_reviewer_digests=(self_asserted.reviewer_identity,), + policy_digest="", + ) + with pytest.raises(ProofValidationError, match="self-asserted"): + attach_differential_proof(_claim(case.record), attachment, policy, case.resolver) + + +def test_all_artifacts_must_resolve_under_store_and_byte_policy() -> None: + case = _case() + untrusted = replace(case.policy, trusted_store_identities=("other-store",), policy_digest="") + with pytest.raises(ProofValidationError, match="artifact"): + attach_differential_proof(_claim(case.record), case.attachment, untrusted, case.resolver) + too_small = replace(case.policy, maximum_artifact_bytes=1, policy_digest="") + with pytest.raises(ProofValidationError, match="artifact"): + attach_differential_proof(_claim(case.record), case.attachment, too_small, case.resolver) + + +def test_verification_identity_substitution_is_refused() -> None: + case = _case() + object.__setattr__(case.attachment.verification, "witness_content_digest", _digest("substituted-witness")) + with pytest.raises(ProofValidationError, match="verification"): + attach_differential_proof(_claim(case.record), case.attachment, case.policy, case.resolver) + + +def test_rehashed_proof_cannot_reuse_a_trusted_attestation() -> None: + case = _case() + affected = replace( + case.attachment.proof.affected, + observation_digest=_digest("substituted-observation"), + run_count=case.attachment.proof.affected.run_count + 1, + result_digest="", + ) + proof = replace(case.attachment.proof, affected=affected, record_digest="") + verification = replace( + case.attachment.verification, + proof_record_digest=proof.record_digest, + affected_result_digest=affected.result_digest, + receipt_digest="", + ) + attachment = replace(case.attachment, proof=proof, verification=verification, proof_id="") + with pytest.raises(ProofValidationError, match="attestation"): + attach_differential_proof(_claim(case.record), attachment, case.policy, case.resolver) + + +def test_proof_at_another_complete_packet_location_cannot_upgrade_claim() -> None: + case = _case() + location_b = SourceLocation("src/app.py", 18, 4, "other") + observation = make_observation( + role="context", + location=location_b, + fields=(), + status=RelationStatus.COMPLETE, + provenance=EvidenceProvenance(ProvenanceKind.PRODUCT_CODE, _digest("other-source"), None, None, None), + ) + packet = replace(case.record.packet, observations=(observation,)) + decision = replace(case.record.decision, packet_digest=derive_packet_digest(packet)) + record = EvidenceRecord(packet, decision, "") + claim = DefectClaim(record, record.decision.basis, record.packet.transitions[0].location, record.packet.scope.regions[0]) + attachment = replace( + case.attachment, + packet_digest=derive_packet_digest(packet), + location=location_b, + proof_id="", + ) + with pytest.raises(ProofValidationError, match="claim location"): + attach_differential_proof(claim, attachment, case.policy, case.resolver) + + +def test_raw_non_claim_record_cannot_gain_execution_backed_basis() -> None: + case = _case() + decision = replace( + case.record.decision, + family_code="good", + outcome=NeutralOutcome.SATISFIED, + ) + record = EvidenceRecord(case.record.packet, decision, "") + with pytest.raises(ProofValidationError, match="DefectClaim"): + attach_differential_proof(record, case.attachment, case.policy, case.resolver) # type: ignore[arg-type]