diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index d088c91..f30c756 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -170,7 +170,7 @@ - Prevent model output from creating semantic relations, satisfaction over missing evidence, or execution-backed basis. - Done when accepted derivation leaves the evidence packet byte-identical and every incomplete or contaminated path remains indeterminate. - _Requirements: 2.4, 2.5, 9.3, 9.4, 9.5, 13.1_ -- [ ] 6.3 (P) Evaluate independently reviewed equal-semantics reversed-surface probes +- [x] 6.3 (P) Evaluate independently reviewed equal-semantics reversed-surface probes - Derive semantic projections from records and family contracts rather than trusting caller summaries. - Resolve distinct surface artifacts and independently reviewed semantic-equivalence evidence. - Reject unequal semantics, digest-only review, non-finite tolerances, and fabricated passed summaries. diff --git a/src/openultrasast/core/__init__.py b/src/openultrasast/core/__init__.py index f915df5..7053174 100644 --- a/src/openultrasast/core/__init__.py +++ b/src/openultrasast/core/__init__.py @@ -18,12 +18,14 @@ from openultrasast.core.coverage import Coverage, InstrumentFailure, ResolutionO from openultrasast.core.factpack import FactPackError, admit_fact_pack, verify_fact_pack from openultrasast.core.probes import ( FittedComponentBlocked, - ProbeResult, ShortcutDetected, assert_no_shortcut, require_admission_evidence, reversed_pair_probe, ) +from openultrasast.core.probes import ( + LegacyProbeResult as ProbeResult, +) from openultrasast.core.provenance import ProvenanceMismatch, stamp, verify_stamp from openultrasast.core.trace import Guard, Hop, TraceRecord, Witness from openultrasast.core.verdict import BoundedSilence, Judgment, ProvenClaim, as_clean diff --git a/src/openultrasast/core/_probe_eval.py b/src/openultrasast/core/_probe_eval.py new file mode 100644 index 0000000..3492fb2 --- /dev/null +++ b/src/openultrasast/core/_probe_eval.py @@ -0,0 +1,172 @@ +"""Validation and evaluation helpers for canonical reversed-surface probes.""" + +from __future__ import annotations + +from decimal import Decimal + +from openultrasast.core import canonical +from openultrasast.core.factpack import ( + ReviewIndependencePolicy, + derive_independence_attestation_digest, + derive_review_independence_policy_digest, +) +from openultrasast.core.family import FamilyContract, family_contract_digest +from openultrasast.core.probes import ( + ProbeObservation, + ProbeResult, + ProbeValidationError, + ReversedPairProbe, + SemanticProjection, + _probe_digest, + derive_equivalence_subject_digest, + project_semantics, +) + + +def _validate_policy(policy: ReviewIndependencePolicy) -> None: + if not isinstance(policy, ReviewIndependencePolicy): + raise ProbeValidationError("review policy has the wrong type") + policy.__post_init__() + expected = derive_review_independence_policy_digest( + policy.reviewed_reviewer_digests, + policy.forbidden_identity_pairs, + policy.trusted_store_identities, + policy.maximum_evidence_bytes, + ) + if policy.policy_digest != expected: + raise ProbeValidationError("review policy digest cannot be re-derived") + + +def _resolve_inputs( + probe: ReversedPairProbe, + policy: ReviewIndependencePolicy, + resolver: canonical.ContentResolver, +) -> None: + surfaces = tuple( + canonical.resolve_content( + reference, + resolver, + trusted_store_identities=policy.trusted_store_identities, + maximum_bytes=policy.maximum_evidence_bytes, + ) + for reference in (probe.forward_surface, probe.reversed_surface) + ) + if not all(surfaces): + raise ProbeValidationError("surface artifacts must be nonempty") + if surfaces[0] == surfaces[1]: + raise ProbeValidationError("probe requires distinct surface artifact bytes") + disclosure = canonical.resolve_content( + probe.equivalence.independence.conflict_disclosure, + resolver, + trusted_store_identities=policy.trusted_store_identities, + maximum_bytes=policy.maximum_evidence_bytes, + ) + if not disclosure: + raise ProbeValidationError("equivalence conflict disclosure must be nonempty") + + +def _validate_review( + probe: ReversedPairProbe, + forward_digest: str, + reversed_digest: str, + policy: ReviewIndependencePolicy, +) -> None: + review = probe.equivalence + review.__post_init__() + attestation = review.independence + attestation.__post_init__() + if (review.forward_projection_digest, review.reversed_projection_digest) != (forward_digest, reversed_digest): + raise ProbeValidationError("review projection digests do not match derived semantics") + expected_subject = derive_equivalence_subject_digest( + forward_digest, + reversed_digest, + probe.family_contract_digest, + probe.forward_surface, + probe.reversed_surface, + ) + if attestation.subject_digest != expected_subject: + raise ProbeValidationError("equivalence attestation subject does not match canonical probe inputs") + if attestation.attestation_digest != derive_independence_attestation_digest(attestation): + raise ProbeValidationError("equivalence attestation digest cannot be re-derived") + if attestation.reviewer_identity not in policy.reviewed_reviewer_digests: + raise ProbeValidationError("equivalence reviewer is not trusted") + if attestation.producer_identity == attestation.reviewer_identity: + raise ProbeValidationError("equivalence review must be independent") + if (attestation.producer_identity, attestation.reviewer_identity) in policy.forbidden_identity_pairs: + raise ProbeValidationError("equivalence producer/reviewer pair is forbidden") + + +def _validate_inputs( + probe: ReversedPairProbe, + observation: ProbeObservation, + contract: FamilyContract, + policy: ReviewIndependencePolicy, + resolver: canonical.ContentResolver, +) -> tuple[SemanticProjection, SemanticProjection]: + if not isinstance(probe, ReversedPairProbe) or not isinstance(observation, ProbeObservation): + raise ProbeValidationError("probe and observation must use v2 probe types") + probe.__post_init__() + observation.__post_init__() + _validate_policy(policy) + expected_contract = family_contract_digest(contract) + if probe.family_contract_digest != expected_contract: + raise ProbeValidationError("probe family contract does not match supplied contract") + if probe.probe_id != _probe_digest(probe) or observation.probe_digest != probe.probe_id: + raise ProbeValidationError("observation probe identity does not match canonical probe") + decisions = {item.family_code for item in contract.decisions} + if observation.forward_decision not in decisions or observation.reversed_decision not in decisions: + raise ProbeValidationError("observation decision is not declared by the family") + forward = project_semantics(probe.forward_record, contract) + reversed_projection = project_semantics(probe.reversed_record, contract) + _validate_review(probe, forward.projection_digest, reversed_projection.projection_digest, policy) + _resolve_inputs(probe, policy, resolver) + return forward, reversed_projection + + +def _result_reasons( + probe: ReversedPairProbe, + observation: ProbeObservation, + expected_decision: str, +) -> tuple[str, ...]: + reasons: list[str] = [] + if observation.forward_decision != observation.reversed_decision: + reasons.append("decision_changed_under_reversed_surface") + if expected_decision not in (observation.forward_decision, observation.reversed_decision) or ( + observation.forward_decision == observation.reversed_decision != expected_decision + ): + reasons.append("component_decision_mismatch") + if ( + observation.forward_score is not None + and observation.reversed_score is not None + and abs(Decimal(str(observation.forward_score)) - Decimal(str(observation.reversed_score))) + > Decimal(str(probe.maximum_score_delta)) + ): + reasons.append("score_delta_exceeds_tolerance") + return tuple(reasons) + + +def evaluate_probe( + probe: ReversedPairProbe, + observation: ProbeObservation, + contract: FamilyContract, + review_policy: ReviewIndependencePolicy, + resolver: canonical.ContentResolver, +) -> ProbeResult: + """Re-derive semantic equivalence and evaluate one frozen component observation.""" + try: + forward, reversed_projection = _validate_inputs(probe, observation, contract, review_policy, resolver) + except ProbeValidationError: + raise + except (TypeError, ValueError) as exc: + raise ProbeValidationError(f"probe input cannot be validated: {exc}") from exc + semantic_equal = forward == reversed_projection + if not semantic_equal: + invalid_reasons = ["semantic_projection_mismatch"] + if forward.family_decision != reversed_projection.family_decision: + invalid_reasons.append("family_decision_mismatch") + return ProbeResult(probe.probe_id, observation.component_digest, False, False, tuple(invalid_reasons), "") + reasons = _result_reasons(probe, observation, forward.family_decision) + return ProbeResult(probe.probe_id, observation.component_digest, True, not reasons, reasons, "") + + +__all__ = ["evaluate_probe"] diff --git a/src/openultrasast/core/probes.py b/src/openultrasast/core/probes.py index 5732dff..8a170a5 100644 --- a/src/openultrasast/core/probes.py +++ b/src/openultrasast/core/probes.py @@ -1,56 +1,408 @@ -"""Anti-shortcut probes for fitted components (Req 8, 9). - -A fitted prompt, threshold, or model must not separate a fix from its reversed fix -using shape features (diff size, deletion count, line counts): that is pool-specific -shortcut learning, the 2026-10-14 deletion-count 0.90 finding. A component making any -detection claim must also carry per-repository holdout evidence and a GO admission on -an untouched population. -""" +"""Canonical reversed-surface probes for fitted-component shortcut resistance.""" from __future__ import annotations +import math +import re from collections.abc import Callable, Mapping -from dataclasses import dataclass +from dataclasses import dataclass, replace from typing import Any +from openultrasast.core import canonical +from openultrasast.core.evidence import EvidenceRecord +from openultrasast.core.factpack import IndependenceAttestation, ReviewIndependencePolicy +from openultrasast.core.family import FamilyContract, family_contract_digest, validate_record +from openultrasast.core.semantics import FieldValue, RelationKind, RelationStatus + +_DIGEST = re.compile(r"[0-9a-f]{64}") + + +class ProbeValidationError(ValueError): + """A probe input or claimed result cannot be reproduced.""" + class ShortcutDetected(ValueError): - """A component separates a fix from its reversed fix by its decision features.""" + """A legacy component separates a fix from its reversed surface.""" class FittedComponentBlocked(ValueError): - """A fitted component claims detection without holdout + admission evidence.""" + """A legacy fitted component lacks its required release evidence.""" @dataclass(frozen=True) -class ProbeResult: - """Outcome of one reversed-pair probe.""" +class LegacyProbeResult: + """Frozen v1 shortcut result retained until the migration cutover.""" separated: bool fix_score: float reversed_score: float -def reversed_pair_probe( - score_fn: Callable[[Mapping[str, Any]], float], fix_change: Mapping[str, Any], reversed_change: Mapping[str, Any] +def _digest(value: object, field: str) -> None: + if not isinstance(value, str) or _DIGEST.fullmatch(value) is None: + raise ProbeValidationError(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 ProbeValidationError(f"{field} must be nonempty text") + + +def _field_value(field: FieldValue) -> canonical.JsonValue: + return {"name": field.name, "kind": field.kind.value, "value": field.value} + + +@dataclass(frozen=True) +class RoleProjection: + role: str + fields: tuple[FieldValue, ...] + relation_kinds: tuple[RelationKind, ...] + statuses: tuple[RelationStatus, ...] + + def __post_init__(self) -> None: + _text(self.role, "projection role") + if not isinstance(self.fields, tuple) or not all(isinstance(item, FieldValue) for item in self.fields): + raise ProbeValidationError("projection fields must be FieldValue values") + if not isinstance(self.relation_kinds, tuple) or not all(isinstance(item, RelationKind) for item in self.relation_kinds): + raise ProbeValidationError("projection relation kinds are invalid") + if not isinstance(self.statuses, tuple) or not all(isinstance(item, RelationStatus) for item in self.statuses): + raise ProbeValidationError("projection statuses are invalid") + + +@dataclass(frozen=True) +class SemanticProjection: + family_contract_digest: str + security_object_fields: tuple[FieldValue, ...] + roles: tuple[RoleProjection, ...] + completeness_reasons: tuple[str, ...] + family_decision: str + projection_digest: str + + def __post_init__(self) -> None: + _digest(self.family_contract_digest, "projection family contract") + _text(self.family_decision, "projection family decision") + if not isinstance(self.security_object_fields, tuple) or not all( + isinstance(item, FieldValue) for item in self.security_object_fields + ): + raise ProbeValidationError("projection security-object fields are invalid") + if not isinstance(self.roles, tuple) or not all(isinstance(item, RoleProjection) for item in self.roles): + raise ProbeValidationError("projection roles are invalid") + if not isinstance(self.completeness_reasons, tuple) or not all( + isinstance(item, str) and item for item in self.completeness_reasons + ): + raise ProbeValidationError("projection completeness reasons are invalid") + expected = _projection_digest(self) + if not self.projection_digest: + object.__setattr__(self, "projection_digest", expected) + elif self.projection_digest != expected: + raise ProbeValidationError("projection digest cannot be re-derived") + + +@dataclass(frozen=True) +class SemanticEquivalenceReview: + forward_projection_digest: str + reversed_projection_digest: str + independence: IndependenceAttestation + review_digest: str + + def __post_init__(self) -> None: + _digest(self.forward_projection_digest, "forward projection") + _digest(self.reversed_projection_digest, "reversed projection") + if not isinstance(self.independence, IndependenceAttestation): + raise ProbeValidationError("equivalence review requires an independence attestation") + expected = _review_digest(self) + if not self.review_digest: + object.__setattr__(self, "review_digest", expected) + elif self.review_digest != expected: + raise ProbeValidationError("equivalence review digest cannot be re-derived") + + +@dataclass(frozen=True) +class ReversedPairProbe: + probe_id: str + family_contract_digest: str + forward_record: EvidenceRecord + reversed_record: EvidenceRecord + forward_surface: canonical.ContentRef + reversed_surface: canonical.ContentRef + equivalence: SemanticEquivalenceReview + maximum_score_delta: float + + def __post_init__(self) -> None: + _digest(self.family_contract_digest, "probe family contract") + if not isinstance(self.forward_record, EvidenceRecord) or not isinstance(self.reversed_record, EvidenceRecord): + raise ProbeValidationError("probe records must be EvidenceRecord values") + if not isinstance(self.forward_surface, canonical.ContentRef) or not isinstance(self.reversed_surface, canonical.ContentRef): + raise ProbeValidationError("probe surfaces must be ContentRef values") + if not isinstance(self.equivalence, SemanticEquivalenceReview): + raise ProbeValidationError("probe equivalence review is invalid") + if isinstance(self.maximum_score_delta, bool) or not isinstance(self.maximum_score_delta, (int, float)): + raise ProbeValidationError("maximum score delta must be a finite non-negative number") + if not math.isfinite(self.maximum_score_delta) or self.maximum_score_delta < 0: + raise ProbeValidationError("maximum score delta must be a finite non-negative number") + expected = _probe_digest(self) + if not self.probe_id: + object.__setattr__(self, "probe_id", expected) + elif self.probe_id != expected: + raise ProbeValidationError("probe identity cannot be re-derived") + + +@dataclass(frozen=True) +class ProbeObservation: + probe_digest: str + component_digest: str + forward_decision: str + reversed_decision: str + forward_score: float | None + reversed_score: float | None + + def __post_init__(self) -> None: + _digest(self.probe_digest, "observation probe") + _digest(self.component_digest, "observation component") + _text(self.forward_decision, "forward decision") + _text(self.reversed_decision, "reversed decision") + scores = (self.forward_score, self.reversed_score) + if (scores[0] is None) != (scores[1] is None): + raise ProbeValidationError("probe scores must both be present or both be absent") + if any( + value is not None and (isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value)) + for value in scores + ): + raise ProbeValidationError("probe scores must be finite numbers") + + +@dataclass(frozen=True) +class ProbeEvaluationInput: + probe: ReversedPairProbe + observation: ProbeObservation + + def __post_init__(self) -> None: + if not isinstance(self.probe, ReversedPairProbe) or not isinstance(self.observation, ProbeObservation): + raise ProbeValidationError("probe evaluation input has invalid members") + + +@dataclass(frozen=True) +class ProbeResult: + probe_digest: str + component_digest: str + valid: bool + passed: bool + reasons: tuple[str, ...] + result_digest: str + + def __post_init__(self) -> None: + _digest(self.probe_digest, "result probe") + _digest(self.component_digest, "result component") + if not isinstance(self.valid, bool) or not isinstance(self.passed, bool): + raise ProbeValidationError("probe result flags must be booleans") + if self.passed and not self.valid: + raise ProbeValidationError("an invalid probe cannot pass") + if not isinstance(self.reasons, tuple) or not all(isinstance(item, str) and item for item in self.reasons): + raise ProbeValidationError("probe result reasons are invalid") + if self.passed and self.reasons: + raise ProbeValidationError("a passing probe cannot carry failure reasons") + if not self.passed and not self.reasons: + raise ProbeValidationError("a non-passing probe requires reasons") + expected = _result_digest(self) + if not self.result_digest: + object.__setattr__(self, "result_digest", expected) + elif self.result_digest != expected: + raise ProbeValidationError("probe result digest cannot be re-derived") + + +def _projection_value(projection: SemanticProjection) -> canonical.JsonValue: + return { + "family_contract_digest": projection.family_contract_digest, + "security_object_fields": tuple(_field_value(item) for item in projection.security_object_fields), + "roles": tuple( + { + "role": role.role, + "fields": tuple(_field_value(item) for item in role.fields), + "relation_kinds": tuple(item.value for item in role.relation_kinds), + "statuses": tuple(item.value for item in role.statuses), + } + for role in projection.roles + ), + "completeness_reasons": projection.completeness_reasons, + "family_decision": projection.family_decision, + } + + +def _projection_digest(projection: SemanticProjection) -> str: + return canonical.content_digest(_projection_value(projection)) + + +def _content_ref_value(reference: canonical.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 _attestation_value(attestation: IndependenceAttestation) -> canonical.JsonValue: + return { + "subject_digest": attestation.subject_digest, + "producer_identity": attestation.producer_identity, + "reviewer_identity": attestation.reviewer_identity, + "conflict_disclosure": _content_ref_value(attestation.conflict_disclosure), + "attestation_digest": attestation.attestation_digest, + } + + +def _review_digest(review: SemanticEquivalenceReview) -> str: + return canonical.content_digest( + { + "forward_projection_digest": review.forward_projection_digest, + "reversed_projection_digest": review.reversed_projection_digest, + "independence": _attestation_value(review.independence), + } + ) + + +def _probe_digest(probe: ReversedPairProbe) -> str: + return canonical.content_digest( + { + "family_contract_digest": probe.family_contract_digest, + "forward_record_id": probe.forward_record.record_id, + "reversed_record_id": probe.reversed_record.record_id, + "forward_surface": _content_ref_value(probe.forward_surface), + "reversed_surface": _content_ref_value(probe.reversed_surface), + "equivalence_review_digest": probe.equivalence.review_digest, + "maximum_score_delta": probe.maximum_score_delta, + } + ) + + +def _result_digest(result: ProbeResult) -> str: + return canonical.content_digest( + { + "probe_digest": result.probe_digest, + "component_digest": result.component_digest, + "valid": result.valid, + "passed": result.passed, + "reasons": result.reasons, + } + ) + + +def _sort_fields(values: tuple[FieldValue, ...]) -> tuple[FieldValue, ...]: + return tuple(sorted(values, key=lambda item: canonical.canonical_bytes(_field_value(item)))) + + +def project_semantics(record: EvidenceRecord, contract: FamilyContract) -> SemanticProjection: + """Derive only family-declared security semantics, excluding presentation shape.""" + for relation in record.packet.transitions: + relation.__post_init__() + for observation in record.packet.observations: + observation.__post_init__() + for field in observation.fields: + field.__post_init__() + for field in record.packet.family_fields: + field.__post_init__() + record.packet.__post_init__() + record.decision.__post_init__() + record.__post_init__() + validate_record(contract, record) + rule = next(item for item in contract.decisions if item.family_code == record.decision.family_code) + actual_fields = {item.name: item for item in record.packet.family_fields} + object_fields = tuple(actual_fields[item.name] for item in contract.security_object.fields if item.name in actual_fields) + roles: list[RoleProjection] = [] + for role in rule.required_roles: + observations = tuple(item for item in record.packet.observations if item.role == role) + relations = tuple(item for item in record.packet.transitions if item.role == role) + fields = _sort_fields(tuple(field for item in observations for field in item.fields)) + kinds = tuple(sorted((item.kind for item in relations), key=lambda item: item.value)) + relation_statuses = tuple(item.status for item in relations) + observation_statuses = tuple(item.status for item in observations) + statuses = tuple(sorted((*relation_statuses, *observation_statuses), key=lambda item: item.value)) + roles.append(RoleProjection(role, fields, kinds, statuses)) + provisional = SemanticProjection( + family_contract_digest(contract), + object_fields, + tuple(roles), + tuple(sorted(record.packet.completeness_reasons)), + record.decision.family_code, + "", + ) + return replace(provisional, projection_digest=_projection_digest(provisional)) + + +def derive_equivalence_subject_digest( + forward_projection_digest: str, + reversed_projection_digest: str, + contract_digest: str, + forward_surface: canonical.ContentRef, + reversed_surface: canonical.ContentRef, +) -> str: + for value, field in ( + (forward_projection_digest, "forward projection"), + (reversed_projection_digest, "reversed projection"), + (contract_digest, "family contract"), + ): + _digest(value, field) + return canonical.content_digest( + { + "forward_projection_digest": forward_projection_digest, + "reversed_projection_digest": reversed_projection_digest, + "family_contract_digest": contract_digest, + "forward_surface": _content_ref_value(forward_surface), + "reversed_surface": _content_ref_value(reversed_surface), + } + ) + + +def evaluate_probe( + probe: ReversedPairProbe, + observation: ProbeObservation, + contract: FamilyContract, + review_policy: ReviewIndependencePolicy, + resolver: canonical.ContentResolver, ) -> ProbeResult: - """Score a fix and its reversed fix; separated=True means the features shortcut.""" + from openultrasast.core._probe_eval import evaluate_probe as evaluate + + return evaluate(probe, observation, contract, review_policy, resolver) + + +def validate_probe_result( + evaluation: ProbeEvaluationInput, + claimed: ProbeResult, + contract: FamilyContract, + review_policy: ReviewIndependencePolicy, + resolver: canonical.ContentResolver, +) -> None: + actual = evaluate_probe(evaluation.probe, evaluation.observation, contract, review_policy, resolver) + if claimed != actual: + raise ProbeValidationError("claimed probe result does not match canonical evaluation") + + +def reversed_pair_probe( + score_fn: Callable[[Mapping[str, Any]], float], + fix_change: Mapping[str, Any], + reversed_change: Mapping[str, Any], +) -> LegacyProbeResult: + """Run the frozen v1 shape probe without granting it v2 authority.""" fix_score = float(score_fn(fix_change)) reversed_score = float(score_fn(reversed_change)) - return ProbeResult(separated=fix_score != reversed_score, fix_score=fix_score, reversed_score=reversed_score) + return LegacyProbeResult(fix_score != reversed_score, fix_score, reversed_score) def assert_no_shortcut( - score_fn: Callable[[Mapping[str, Any]], float], fix_change: Mapping[str, Any], reversed_change: Mapping[str, Any] + score_fn: Callable[[Mapping[str, Any]], float], + fix_change: Mapping[str, Any], + reversed_change: Mapping[str, Any], ) -> None: - """Raise ShortcutDetected when the component separates fix from reversed fix.""" + """Preserve the v1 exception behavior during migration.""" result = reversed_pair_probe(score_fn, fix_change, reversed_change) if result.separated: raise ShortcutDetected(f"shape shortcut: fix={result.fix_score} vs reversed={result.reversed_score}") def require_admission_evidence(*, holdout_passed: bool, admission_decision: str) -> None: - """Block any fitted-component claim without holdout and an admission GO (Req 8.2).""" + """Preserve the v1 fitted-component release guard during migration.""" if not holdout_passed: raise FittedComponentBlocked("fitted component lacks per-repository holdout evidence") if admission_decision != "GO": @@ -59,9 +411,21 @@ def require_admission_evidence(*, holdout_passed: bool, admission_decision: str) __all__ = [ "FittedComponentBlocked", + "LegacyProbeResult", + "ProbeEvaluationInput", + "ProbeObservation", "ProbeResult", + "ProbeValidationError", + "ReversedPairProbe", + "RoleProjection", + "SemanticEquivalenceReview", + "SemanticProjection", "ShortcutDetected", "assert_no_shortcut", + "derive_equivalence_subject_digest", + "evaluate_probe", + "project_semantics", "require_admission_evidence", "reversed_pair_probe", + "validate_probe_result", ] diff --git a/tests/test_core_probes.py b/tests/test_core_probes.py index a1ed25f..b6cc01a 100644 --- a/tests/test_core_probes.py +++ b/tests/test_core_probes.py @@ -1,52 +1,407 @@ -"""Anti-shortcut probe tests for detection-core tasks 5.1-5.2 (Req 8, 9).""" +"""Independently reviewed equal-semantics reversed-surface probes.""" -import sys -from pathlib import Path - -sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +from __future__ import annotations from collections.abc import Mapping +from dataclasses import dataclass, replace from typing import Any -from openultrasast.core.probes import ( # noqa: E402 - FittedComponentBlocked, - ShortcutDetected, - assert_no_shortcut, - require_admission_evidence, - reversed_pair_probe, +import pytest +from detection_core_fixtures import InMemoryContentResolver, stored_artifact + +import openultrasast.core as public_core +from openultrasast.core import canonical, evidence, probes, provenance, semantics +from openultrasast.core.factpack import ( + IndependenceAttestation, + ReviewIndependencePolicy, + derive_independence_attestation_digest, + derive_review_independence_policy_digest, +) +from openultrasast.core.family import ( + DecisionRule, + FamilyContract, + FieldSpec, + ObjectSpec, + RelationOriginPolicy, + ResidualQuestion, + RoleSpec, + derive_security_object_id, + family_contract_digest, ) -FIX: dict[str, Any] = {"id": "fix", "files": [{"path": "a.py", "head_lines": [[1, 10]]}], "seconds": 1.0} -REVERSED: dict[str, Any] = {"id": "reversed", "files": [{"path": "a.py", "head_lines": [[1, 50]]}], "seconds": 1.0} + +@dataclass(frozen=True) +class ProbeCase: + probe: probes.ReversedPairProbe + observation: probes.ProbeObservation + contract: FamilyContract + policy: ReviewIndependencePolicy + resolver: InMemoryContentResolver -def test_reversed_pair_probe_detects_shape_shortcut() -> None: - # deletion-count style component: score = lines removed -> separates fix from reversed +def _digest(value: str) -> str: + return canonical.content_digest(value) + + +def _contract() -> FamilyContract: + return FamilyContract( + "probe-family", + "1", + evidence.EVIDENCE_PACKET_SCHEMA, + ("python",), + ObjectSpec( + "resource", + (FieldSpec("operation", semantics.FieldKind.TEXT, True),), + ("operation",), + ), + ( + RoleSpec( + "flow", + (), + (semantics.RelationKind.CALL,), + RelationOriginPolicy.PRODUCT_OR_REVIEWED_FACT, + ), + ), + ( + DecisionRule( + "unsafe", + evidence.NeutralOutcome.VIOLATION, + ("flow",), + (semantics.RelationStatus.COMPLETE,), + ), + DecisionRule( + "safe", + evidence.NeutralOutcome.SATISFIED, + ("flow",), + (semantics.RelationStatus.COMPLETE,), + ), + ), + (ResidualQuestion("classify", ("flow",), ("unsafe", "safe")),), + ("missing",), + ) + + +def _record(contract: FamilyContract, path: str, *, operation: str = "read", decision: str = "unsafe") -> evidence.EvidenceRecord: + location = semantics.SourceLocation(path, 7, 2, "run") + product = semantics.EvidenceProvenance( + semantics.ProvenanceKind.PRODUCT_CODE, + _digest(f"source:{path}"), + None, + None, + None, + ) + relation = semantics.make_relation( + kind=semantics.RelationKind.CALL, + role="flow", + source=semantics.SemanticNode(f"source:{path}", "source", location), + target=semantics.SemanticNode(f"target:{path}", "target", location), + location=location, + status=semantics.RelationStatus.COMPLETE, + conditions=("reachable",), + reasons=(), + provenance=product, + ) + fields = (semantics.FieldValue("operation", semantics.FieldKind.TEXT, operation),) + scope = evidence.AnalyzedScope("base", "head", _digest(f"change:{path}"), (evidence.SourceRegion(path, 1, 20),)) + stamp = provenance.ProvenanceStamp( + _digest("analyzer"), + family_contract_digest(contract), + _digest("facts"), + _digest("acquisition"), + _digest("configuration"), + _digest("population"), + _digest(f"input:{path}"), + ) + packet = evidence.EvidencePacket( + evidence.EVIDENCE_PACKET_SCHEMA, + contract.family_id, + family_contract_digest(contract), + derive_security_object_id(contract.security_object, fields), + "python", + scope, + (relation,), + (), + fields, + (), + stamp, + ) + outcome = evidence.NeutralOutcome.VIOLATION if decision == "unsafe" else evidence.NeutralOutcome.SATISFIED + derived = evidence.Decision( + decision, + outcome, + evidence.EvidenceBasis.DETERMINISTIC, + scope, + (), + evidence.derive_packet_digest(packet), + provenance.DecisionProvenance(None, None, None, None, None, None), + ) + return evidence.EvidenceRecord(packet, derived, "") + + +def _policy(reviewer: str) -> ReviewIndependencePolicy: + reviewers = (reviewer,) + stores = ("trusted-store",) + digest = derive_review_independence_policy_digest(reviewers, (), stores, 4096) + return ReviewIndependencePolicy(reviewers, (), stores, 4096, digest) + + +def _review( + forward: probes.SemanticProjection, + reversed_projection: probes.SemanticProjection, + contract: FamilyContract, + forward_surface: canonical.ContentRef, + reversed_surface: canonical.ContentRef, + disclosure: canonical.ContentRef, + *, + producer: str, + reviewer: str, +) -> probes.SemanticEquivalenceReview: + subject = probes.derive_equivalence_subject_digest( + forward.projection_digest, + reversed_projection.projection_digest, + family_contract_digest(contract), + forward_surface, + reversed_surface, + ) + provisional = IndependenceAttestation(subject, producer, reviewer, disclosure, "0" * 64) + attestation = replace( + provisional, + attestation_digest=derive_independence_attestation_digest(provisional), + ) + return probes.SemanticEquivalenceReview( + forward.projection_digest, + reversed_projection.projection_digest, + attestation, + "", + ) + + +def _case( + *, + reversed_operation: str = "read", + reversed_decision: str = "unsafe", + forward_surface_bytes: bytes = b"- old\n+ new\n", + reversed_surface_bytes: bytes = b"+ old\n- new\n", + same_reviewer: bool = False, +) -> ProbeCase: + contract = _contract() + forward_record = _record(contract, "src/forward.py") + reversed_record = _record( + contract, + "src/reversed.py", + operation=reversed_operation, + decision=reversed_decision, + ) + forward_surface, forward_resolver = stored_artifact(forward_surface_bytes, artifact_id="forward-surface") + reversed_surface, reversed_resolver = stored_artifact(reversed_surface_bytes, artifact_id="reversed-surface") + disclosure, disclosure_resolver = stored_artifact(b'{"conflicts":[]}', artifact_id="review-disclosure") + producer = _digest("fixture-producer") + reviewer = producer if same_reviewer else _digest("independent-reviewer") + forward_projection = probes.project_semantics(forward_record, contract) + reversed_projection = probes.project_semantics(reversed_record, contract) + review = _review( + forward_projection, + reversed_projection, + contract, + forward_surface, + reversed_surface, + disclosure, + producer=producer, + reviewer=reviewer, + ) + probe = probes.ReversedPairProbe( + "", + family_contract_digest(contract), + forward_record, + reversed_record, + forward_surface, + reversed_surface, + review, + 0.05, + ) + observation = probes.ProbeObservation( + probe.probe_id, + _digest("relation-component"), + "unsafe", + "unsafe", + 0.50, + 0.52, + ) + artifacts = { + **forward_resolver.artifacts, + **reversed_resolver.artifacts, + **disclosure_resolver.artifacts, + } + return ProbeCase(probe, observation, contract, _policy(reviewer), InMemoryContentResolver(artifacts)) + + +def test_projection_excludes_paths_and_surface_presentation() -> None: + case = _case() + forward = probes.project_semantics(case.probe.forward_record, case.contract) + reversed_projection = probes.project_semantics(case.probe.reversed_record, case.contract) + + assert forward == reversed_projection + assert forward.family_decision == "unsafe" + assert forward.roles[0].relation_kinds == (semantics.RelationKind.CALL,) + assert forward.security_object_fields[0].value == "read" + + +def test_relation_component_passes_equal_semantics_within_registered_tolerance() -> None: + case = _case() + + result = probes.evaluate_probe(case.probe, case.observation, case.contract, case.policy, case.resolver) + repeated = probes.evaluate_probe(case.probe, case.observation, case.contract, case.policy, case.resolver) + + assert result.valid is True + assert result.passed is True + assert result.reasons == () + assert repeated == result + + +def test_surface_shape_score_gap_is_detected() -> None: + case = _case() + shaped = replace(case.observation, component_digest=_digest("surface-component"), reversed_score=0.90) + + result = probes.evaluate_probe(case.probe, shaped, case.contract, case.policy, case.resolver) + + assert result.valid is True + assert result.passed is False + assert result.reasons == ("score_delta_exceeds_tolerance",) + + +def test_score_delta_at_registered_tolerance_boundary_passes() -> None: + case = _case() + boundary = replace(case.observation, forward_score=0.50, reversed_score=0.55) + + result = probes.evaluate_probe(case.probe, boundary, case.contract, case.policy, case.resolver) + + assert result.valid is True + assert result.passed is True + + +def test_score_delta_just_above_registered_tolerance_boundary_fails() -> None: + case = _case() + above = replace(case.observation, forward_score=0.50, reversed_score=0.550000000001) + + result = probes.evaluate_probe(case.probe, above, case.contract, case.policy, case.resolver) + + assert result.valid is True + assert result.passed is False + assert result.reasons == ("score_delta_exceeds_tolerance",) + + +def test_unequal_derived_semantics_make_probe_invalid() -> None: + case = _case(reversed_operation="write") + + result = probes.evaluate_probe(case.probe, case.observation, case.contract, case.policy, case.resolver) + + assert result.valid is False + assert result.passed is False + assert "semantic_projection_mismatch" in result.reasons + + +def test_different_declared_family_decisions_make_probe_invalid() -> None: + case = _case(reversed_decision="safe") + + result = probes.evaluate_probe(case.probe, case.observation, case.contract, case.policy, case.resolver) + + assert result.valid is False + assert result.reasons == ("semantic_projection_mismatch", "family_decision_mismatch") + + +def test_constant_but_wrong_component_decision_does_not_pass() -> None: + case = _case() + wrong = replace(case.observation, forward_decision="safe", reversed_decision="safe") + + result = probes.evaluate_probe(case.probe, wrong, case.contract, case.policy, case.resolver) + + assert result.valid is True + assert result.passed is False + assert result.reasons == ("component_decision_mismatch",) + + +def test_digest_only_equivalence_claim_is_refused() -> None: + case = _case() + fake = _digest("caller-claimed-projection") + attestation = case.probe.equivalence.independence + subject = probes.derive_equivalence_subject_digest( + fake, + fake, + family_contract_digest(case.contract), + case.probe.forward_surface, + case.probe.reversed_surface, + ) + provisional = replace(attestation, subject_digest=subject, attestation_digest="0" * 64) + signed = replace(provisional, attestation_digest=derive_independence_attestation_digest(provisional)) + review = probes.SemanticEquivalenceReview(fake, fake, signed, "") + substituted = replace(case.probe, equivalence=review, probe_id="") + observation = replace(case.observation, probe_digest=substituted.probe_id) + + with pytest.raises(probes.ProbeValidationError, match="projection"): + probes.evaluate_probe(substituted, observation, case.contract, case.policy, case.resolver) + + +@pytest.mark.parametrize("tolerance", [float("nan"), float("inf"), -0.01]) +def test_tolerance_must_be_finite_and_nonnegative(tolerance: float) -> None: + case = _case() + + with pytest.raises(probes.ProbeValidationError, match="finite non-negative"): + replace(case.probe, maximum_score_delta=tolerance, probe_id="") + + +def test_surface_artifacts_must_resolve_to_distinct_bytes() -> None: + case = _case(reversed_surface_bytes=b"- old\n+ new\n") + + with pytest.raises(probes.ProbeValidationError, match="distinct surface"): + probes.evaluate_probe(case.probe, case.observation, case.contract, case.policy, case.resolver) + + +def test_equivalence_review_must_be_independent() -> None: + case = _case(same_reviewer=True) + + with pytest.raises(probes.ProbeValidationError, match="independent"): + probes.evaluate_probe(case.probe, case.observation, case.contract, case.policy, case.resolver) + + +def test_fabricated_passed_summary_is_refused_by_re_evaluation() -> None: + case = _case() + shaped = replace(case.observation, component_digest=_digest("surface-component"), reversed_score=0.90) + forged = probes.ProbeResult(case.probe.probe_id, shaped.component_digest, True, True, (), "") + + with pytest.raises(probes.ProbeValidationError, match="claimed probe result"): + probes.validate_probe_result( + probes.ProbeEvaluationInput(case.probe, shaped), + forged, + case.contract, + case.policy, + case.resolver, + ) + + +def test_observation_for_another_probe_is_refused() -> None: + case = _case() + substituted = replace(case.observation, probe_digest=_digest("another-probe")) + + with pytest.raises(probes.ProbeValidationError, match="probe identity"): + probes.evaluate_probe(case.probe, substituted, case.contract, case.policy, case.resolver) + + +def test_legacy_reversed_pair_probe_remains_compatible() -> None: + forward: dict[str, Any] = {"lines": 10} + reversed_change: dict[str, Any] = {"lines": 50} + def shape_score(change: Mapping[str, Any]) -> float: - return float(50.0 - change["files"][0]["head_lines"][0][1]) + return float(change["lines"]) - result = reversed_pair_probe(shape_score, FIX, REVERSED) + result = probes.reversed_pair_probe(shape_score, forward, reversed_change) assert result.separated is True - try: - assert_no_shortcut(shape_score, FIX, REVERSED) - raise AssertionError("shape shortcut must be detected") - except ShortcutDetected: - pass + with pytest.raises(probes.ShortcutDetected): + probes.assert_no_shortcut(shape_score, forward, reversed_change) + + assert public_core.ProbeResult is probes.LegacyProbeResult -def test_reversed_pair_probe_passes_relation_based_component() -> None: - def relation_score(change: Mapping[str, Any]) -> float: - return float(bool(change.get("has_guard"))) - - assert reversed_pair_probe(relation_score, {**FIX, "has_guard": True}, {**REVERSED, "has_guard": True}).separated is False - assert_no_shortcut(relation_score, {**FIX, "has_guard": True}, {**REVERSED, "has_guard": True}) - - -def test_fitted_component_requires_holdout_and_admission_go() -> None: - require_admission_evidence(holdout_passed=True, admission_decision="GO") - for holdout, decision in ((False, "GO"), (True, "NO-GO"), (False, "NO-GO")): - try: - require_admission_evidence(holdout_passed=holdout, admission_decision=decision) - raise AssertionError(f"holdout={holdout} decision={decision} must block the claim") - except FittedComponentBlocked: - pass +def test_legacy_admission_guard_remains_compatible() -> None: + probes.require_admission_evidence(holdout_passed=True, admission_decision="GO") + with pytest.raises(probes.FittedComponentBlocked): + probes.require_admission_evidence(holdout_passed=False, admission_decision="GO")