Open_Ultra_SAST/tests/test_core_probes.py

"""Independently reviewed equal-semantics reversed-surface probes."""

from __future__ import annotations

from dataclasses import dataclass, replace

import pytest
from detection_core_fixtures import InMemoryContentResolver, stored_artifact

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,
)


@dataclass(frozen=True)
class ProbeCase:
    probe: probes.ReversedPairProbe
    observation: probes.ProbeObservation
    contract: FamilyContract
    policy: ReviewIndependencePolicy
    resolver: InMemoryContentResolver


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)