llm-ingestion-pipeline-security/tests/test_disposition.py

"""Tests for disposition — the Report -> gate-decision policy (BRIEF §4.6/§4.7)."""
from __future__ import annotations

import pytest

from llm_ingestion_guard.report import Finding, Report, Severity, Source
from llm_ingestion_guard.disposition import (
    DEFAULT_ACTION_MAP,
    Disposition,
    DispositionResult,
    Policy,
    Provenance,
    Risk,
    Trust,
    decide,
    guard,
    PRESET_TRUSTED_SOURCE,
    PRESET_USER_UPLOAD,
)


# --- helpers ---------------------------------------------------------------

def _finding(label="lexicon:override", severity=Severity.HIGH, detector="lexicon",
             source=Source.INPUT):
    return Finding(label=label, severity=severity, source=source, detector=detector)


def _report(*findings):
    report = Report()
    for finding in findings:
        report.add(finding)
    return report


TRUSTED = Policy(trust=Trust.TRUSTED)
UNTRUSTED = Policy(trust=Trust.UNTRUSTED)


# --- any-tier exceptions (§4.7): CRITICAL + invisible carriers -------------

def test_critical_fails_secure_even_in_trusted_prose():
    # trust cannot rescue a CRITICAL finding — it blocks in any tier.
    report = _report(_finding(label="lexicon:identity-redef", severity=Severity.CRITICAL))
    result = decide(report, TRUSTED, provenance=Provenance.PROSE)
    assert result.disposition is Disposition.FAIL_SECURE
    assert result.max_severity is Severity.CRITICAL


@pytest.mark.parametrize("carrier_label,severity", [
    ("sanitize:zero-width", Severity.HIGH),
    ("sanitize:bidi-override", Severity.HIGH),
    ("sanitize:unicode-tag", Severity.CRITICAL),
    # M3: the same invariant must hold for the OUTPUT-gate carrier labels, so a
    # carrier surfacing in model output blocks in any tier too.
    ("lexicon:unicode-tags-present", Severity.HIGH),
    ("output:zero-width-present", Severity.HIGH),
    ("output:bidi-present", Severity.HIGH),
])
def test_invisible_carrier_fails_secure_in_any_tier(carrier_label, severity):
    # zero-width/bidi are HIGH (would only WARN in trusted prose by severity
    # alone); the carrier rule overrides and fails secure regardless of tier.
    report = _report(_finding(label=carrier_label, severity=severity, detector="sanitize"))
    result = decide(report, TRUSTED, provenance=Provenance.PROSE)
    assert result.disposition is Disposition.FAIL_SECURE
    assert any("carrier" in reason for reason in result.reasons)


# --- HIGH severity across the matrix --------------------------------------

def test_high_trusted_prose_warns():
    # HIGH in authored prose from a trusted source: legit security vocabulary
    # dominates -> WARN, not block.
    report = _report(_finding(severity=Severity.HIGH))
    assert decide(report, TRUSTED, provenance=Provenance.PROSE).disposition is Disposition.WARN


@pytest.mark.parametrize("provenance", [Provenance.CODE_FENCE, Provenance.LOCALIZED])
def test_high_trusted_lowtrust_provenance_fails_secure(provenance):
    # the same HIGH hit inside a code fence / localized string hard-fails.
    report = _report(_finding(severity=Severity.HIGH))
    assert decide(report, TRUSTED, provenance=provenance).disposition is Disposition.FAIL_SECURE


def test_high_untrusted_fails_secure():
    report = _report(_finding(severity=Severity.HIGH))
    assert decide(report, UNTRUSTED, provenance=Provenance.PROSE).disposition is Disposition.FAIL_SECURE


# --- MEDIUM severity ------------------------------------------------------

def test_medium_trusted_prose_warns():
    report = _report(_finding(severity=Severity.MEDIUM, label="lexicon:config"))
    assert decide(report, TRUSTED, provenance=Provenance.PROSE).disposition is Disposition.WARN


def test_medium_trusted_codefence_quarantines():
    report = _report(_finding(severity=Severity.MEDIUM, label="lexicon:config"))
    assert decide(report, TRUSTED, provenance=Provenance.CODE_FENCE).disposition is Disposition.QUARANTINE_REVIEW


def test_medium_untrusted_quarantines():
    report = _report(_finding(severity=Severity.MEDIUM, label="lexicon:config"))
    assert decide(report, UNTRUSTED, provenance=Provenance.PROSE).disposition is Disposition.QUARANTINE_REVIEW


# --- LOW / clean ----------------------------------------------------------

def test_low_trusted_warns():
    report = _report(_finding(severity=Severity.LOW, label="lexicon:soft"))
    assert decide(report, TRUSTED, provenance=Provenance.PROSE).disposition is Disposition.WARN


def test_low_plain_untrusted_warns_without_quarantine_default():
    # plain untrusted (no upload preset) does not force quarantine on a LOW hit.
    report = _report(_finding(severity=Severity.LOW, label="lexicon:soft"))
    assert decide(report, UNTRUSTED, provenance=Provenance.PROSE).disposition is Disposition.WARN


def test_clean_report_warns():
    assert decide(_report(), TRUSTED).disposition is Disposition.WARN
    assert decide(_report(), UNTRUSTED).disposition is Disposition.WARN


def test_clean_report_max_severity_none():
    assert decide(_report(), TRUSTED).max_severity is None


# --- Overlay A: compound forced-fallback (§4.6) ---------------------------

def test_transform_failure_with_findings_fails_secure():
    # a scan hit plus a transform failure is a probable forced-fallback attack.
    report = _report(_finding(severity=Severity.MEDIUM, label="lexicon:config"))
    result = decide(report, TRUSTED, provenance=Provenance.PROSE, transform_failed=True)
    assert result.disposition is Disposition.FAIL_SECURE
    assert any("forced-fallback" in reason for reason in result.reasons)


def test_transform_failure_without_findings_does_not_fail_secure():
    # a transform failure alone (clean scan) is not the forced-fallback signal.
    result = decide(_report(), TRUSTED, transform_failed=True)
    assert result.disposition is Disposition.WARN


# --- Overlay B: compound escalation (§4.6) --------------------------------

def test_two_medium_findings_escalate_in_trusted_prose():
    # each MEDIUM alone -> WARN in trusted prose; two together escalate one tier.
    report = _report(
        _finding(severity=Severity.MEDIUM, label="lexicon:config"),
        _finding(severity=Severity.MEDIUM, label="entropy:base64", detector="entropy"),
    )
    result = decide(report, TRUSTED, provenance=Provenance.PROSE)
    assert result.disposition is Disposition.QUARANTINE_REVIEW
    assert any("compound" in reason for reason in result.reasons)


def test_single_medium_does_not_escalate():
    report = _report(_finding(severity=Severity.MEDIUM, label="lexicon:config"))
    assert decide(report, TRUSTED, provenance=Provenance.PROSE).disposition is Disposition.WARN


def test_two_low_findings_do_not_escalate():
    # LOW findings are trivial signals; they do not compound into escalation.
    report = _report(
        _finding(severity=Severity.LOW, label="a", detector="lexicon"),
        _finding(severity=Severity.LOW, label="b", detector="entropy"),
    )
    assert decide(report, TRUSTED, provenance=Provenance.PROSE).disposition is Disposition.WARN


# --- Overlay C: fail-closed guard -----------------------------------------

def test_guard_fails_secure_on_scanner_exception():
    def boom():
        raise RuntimeError("detector unavailable")
    result = guard(boom, TRUSTED)
    assert result.disposition is Disposition.FAIL_SECURE
    assert any("fail-closed" in reason for reason in result.reasons)
    assert result.max_severity is None


def test_guard_fails_secure_when_decide_itself_raises():
    # m6 — total fail-closed: even if decide raises (e.g. a scan_fn that returns
    # a non-Report), guard yields FAIL_SECURE, never a leaked exception / persist.
    result = guard(lambda: None, TRUSTED)  # None has no .max_severity() -> decide raises
    assert result.disposition is Disposition.FAIL_SECURE
    assert any("fail-closed" in reason for reason in result.reasons)


def test_guard_passes_through_clean_scan():
    assert guard(lambda: _report(), TRUSTED).disposition is Disposition.WARN


def test_guard_disposes_findings_like_decide():
    report = _report(_finding(severity=Severity.CRITICAL, label="lexicon:override"))
    assert guard(lambda: report, TRUSTED).disposition is Disposition.FAIL_SECURE


# --- Presets --------------------------------------------------------------

def test_user_upload_preset_holds_medium_for_review():
    report = _report(_finding(severity=Severity.MEDIUM, label="lexicon:config"))
    assert decide(report, PRESET_USER_UPLOAD).disposition is Disposition.QUARANTINE_REVIEW


def test_user_upload_floor_does_not_fire_on_a_lone_low_finding():
    # 0.3.1: the floor fires at MEDIUM+, not on ANY finding. "Any finding ->
    # review" rested on the premise that findings are the exception; that premise
    # broke the moment every ordinary markdown link became a (LOW) finding, and
    # the floor then quarantined documents whose only sin was having a link.
    report = _report(_finding(severity=Severity.LOW, label="active:markdown-link"))
    assert decide(report, PRESET_USER_UPLOAD).disposition is Disposition.WARN


def test_quarantine_floor_still_lifts_a_semi_trusted_policy():
    # The floor is not dead weight: a caller-defined TRUSTED policy that opts into
    # quarantine_default still lifts a MEDIUM finding that trust alone would WARN.
    semi_trusted = Policy(trust=Trust.TRUSTED, quarantine_default=True)
    report = _report(_finding(severity=Severity.MEDIUM, label="lexicon:config"))
    assert decide(report, semi_trusted).disposition is Disposition.QUARANTINE_REVIEW


def test_user_upload_preset_hard_fails_on_critical():
    report = _report(_finding(severity=Severity.CRITICAL, label="lexicon:override"))
    assert decide(report, PRESET_USER_UPLOAD).disposition is Disposition.FAIL_SECURE


def test_user_upload_preset_clean_report_warns():
    # no finding -> no quarantine floor.
    assert decide(_report(), PRESET_USER_UPLOAD).disposition is Disposition.WARN


def test_trusted_source_preset_warns_on_high_prose():
    report = _report(_finding(severity=Severity.HIGH))
    assert decide(report, PRESET_TRUSTED_SOURCE, provenance=Provenance.PROSE).disposition is Disposition.WARN


def test_presets_are_policies():
    assert isinstance(PRESET_TRUSTED_SOURCE, Policy)
    assert PRESET_TRUSTED_SOURCE.trust is Trust.TRUSTED
    assert PRESET_USER_UPLOAD.trust is Trust.UNTRUSTED
    assert PRESET_USER_UPLOAD.quarantine_default is True


# --- DispositionResult shape ----------------------------------------------

def test_result_is_dataclass_with_auditable_reasons():
    report = _report(_finding(severity=Severity.CRITICAL, label="lexicon:override"))
    result = decide(report, TRUSTED)
    assert isinstance(result, DispositionResult)
    assert result.reasons  # non-empty, auditable trail


def test_floor_and_escalation_compose_to_fail_secure():
    # upload preset + two MEDIUM untrusted: base QUARANTINE (low-trust) then
    # compound escalation -> FAIL_SECURE.
    report = _report(
        _finding(severity=Severity.MEDIUM, label="lexicon:config"),
        _finding(severity=Severity.MEDIUM, label="entropy:base64", detector="entropy"),
    )
    assert decide(report, PRESET_USER_UPLOAD).disposition is Disposition.FAIL_SECURE


# --- 0.5.0 axis separation: assessment (how dangerous) vs disposition (what to
# --- do). PLAN-v1.md:294/:380 — change DISPOSITION, never the grading.
#
# `Disposition` is three *actions*, but BRIEF principle 4 says the library
# reports and the *pipeline* decides. `decide` therefore returned an action it
# cannot enforce, and a consumer wanting a different action had to re-derive it
# from that action — the assessment which produced it was already gone. That is
# why a consumer ends up pinning the *grading*: the action is all they get. The
# assessment axis hands them the input instead of the verdict.


def test_result_carries_an_assessment_distinct_from_the_disposition():
    # The payoff: a clean report and a LOW-severity report are BOTH `WARN`
    # today, and indistinguishable without re-reading the report. The assessment
    # axis separates them while leaving the action identical.
    clean = decide(_report(), UNTRUSTED)
    low = decide(_report(_finding(severity=Severity.LOW, label="active:markdown-link")),
                 UNTRUSTED)

    assert clean.disposition is Disposition.WARN
    assert low.disposition is Disposition.WARN          # action: identical
    assert clean.assessment is Risk.NONE
    assert low.assessment is Risk.LOW                   # assessment: distinct


def test_default_action_map_reproduces_todays_outcomes():
    # The default mapping is a no-op by construction: this is the contract that
    # keeps `PRESET_USER_UPLOAD` grading untouched (locked promise 1).
    assert DEFAULT_ACTION_MAP == {
        Risk.NONE: Disposition.WARN,
        Risk.LOW: Disposition.WARN,
        Risk.ELEVATED: Disposition.QUARANTINE_REVIEW,
        Risk.SEVERE: Disposition.FAIL_SECURE,
    }


@pytest.mark.parametrize("severity,policy,provenance,expected_risk", [
    (Severity.CRITICAL, TRUSTED, Provenance.PROSE, Risk.SEVERE),
    (Severity.HIGH, TRUSTED, Provenance.PROSE, Risk.LOW),
    (Severity.HIGH, TRUSTED, Provenance.CODE_FENCE, Risk.SEVERE),
    (Severity.HIGH, UNTRUSTED, Provenance.PROSE, Risk.SEVERE),
    (Severity.MEDIUM, TRUSTED, Provenance.PROSE, Risk.LOW),
    (Severity.MEDIUM, UNTRUSTED, Provenance.PROSE, Risk.ELEVATED),
    (Severity.LOW, UNTRUSTED, Provenance.PROSE, Risk.LOW),
])
def test_assessment_tracks_danger_given_trust(severity, policy, provenance, expected_risk):
    # The assessment is trust-aware, exactly as BRIEF §4.7 describes: the *same*
    # hit is a different assessment in prose vs a code fence — not merely a
    # different action on one shared assessment.
    result = decide(_report(_finding(severity=severity)), policy, provenance=provenance)
    assert result.assessment is expected_risk
    assert result.disposition is DEFAULT_ACTION_MAP[expected_risk]


def test_custom_action_map_changes_the_action_not_the_assessment():
    # THE point of the separation. A consumer that wants to hold for review
    # rather than block says so in the policy, and the assessment it was derived
    # from is unchanged — so they never have to pin our grading to get their
    # behaviour.
    report = _report(_finding(severity=Severity.HIGH))
    strict = Policy(trust=Trust.UNTRUSTED)
    lenient = Policy(trust=Trust.UNTRUSTED, action_map={
        **DEFAULT_ACTION_MAP,
        Risk.SEVERE: Disposition.QUARANTINE_REVIEW,
    })

    assert decide(report, strict).disposition is Disposition.FAIL_SECURE
    assert decide(report, lenient).disposition is Disposition.QUARANTINE_REVIEW
    assert decide(report, strict).assessment is Risk.SEVERE
    assert decide(report, lenient).assessment is Risk.SEVERE


def test_overlays_escalate_the_assessment_not_only_the_action():
    # Compound escalation and the quarantine floor are assessment-level moves;
    # if they only moved the action, a custom action_map would silently drop
    # them. Two MEDIUM findings untrusted: ELEVATED escalated to SEVERE.
    report = _report(
        _finding(severity=Severity.MEDIUM, label="lexicon:config"),
        _finding(severity=Severity.MEDIUM, label="entropy:base64", detector="entropy"),
    )
    result = decide(report, PRESET_USER_UPLOAD)
    assert result.assessment is Risk.SEVERE
    assert result.disposition is Disposition.FAIL_SECURE


def test_guard_fails_closed_with_a_severe_assessment():
    # The fail-closed path must not leave the assessment unset, or a consumer
    # mapping on the assessment alone would read a scanner crash as clean.
    def boom() -> Report:
        raise RuntimeError("detector exploded")

    result = guard(boom, PRESET_USER_UPLOAD)
    assert result.disposition is Disposition.FAIL_SECURE
    assert result.assessment is Risk.SEVERE