llm-ingestion-pipeline-secu.../tests/test_disposition.py
Kjell Tore Guttormsen 5397ba15a1 fix(security): harden 5 adversarial-review findings (M1/M2/M3 + m4/m6) via TDD
Pre-release hardening from an independent adversarial review; each fixed
test-first (failing test -> fix -> green). 214 tests pass.

- entropy (M1): decode-and-rescan now runs BEFORE false-positive suppression,
  so an SRI/media-prefixed injection blob is still decoded and lexicon-rescanned.
  Suppression gates only the entropy finding, never the decode.
- output/disposition (M3): the invisible-carrier invariant now holds on the
  persist gate. scan_output flags zero-width/BIDI presence and disposition
  treats those + lexicon:unicode-tags-present as any-tier carriers, so a carrier
  in model output fails secure even under a trusted policy.
- contract (M2): assert_credential_allowlist catches a bare <PROVIDER>_KEY
  (e.g. STRIPE_KEY) that the old regex silently missed (fail-open). Deliberately
  broad: also flags PARTITION_KEY/SORT_KEY as loud, allowlistable FPs -- fail-loud
  beats fail-silent for an isolation control.
- disposition (m6): guard runs decide inside its guarded block -> total
  fail-closed even on a malformed report.
- output (m4): egress placeholder suppression anchors word markers (example,
  todo, ...) to a word boundary, closing a fail-open where a real secret merely
  containing such a word was suppressed.

Docs: CHANGELOG Security subsection; README honest-limit for lexicon dedup (m5,
documented tradeoff, not fixed).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HyRCQMocjZ6SmSQ6JidJ2k
2026-07-05 10:45:05 +02:00

244 lines
10 KiB
Python

"""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 (
Disposition,
DispositionResult,
Policy,
Provenance,
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_quarantines_any_finding():
# a single LOW finding that would WARN under a plain policy -> QUARANTINE here.
report = _report(_finding(severity=Severity.LOW, label="lexicon:soft"))
assert decide(report, PRESET_USER_UPLOAD).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