feat(disposition): source-trust + provenance gate — WARN|QUARANTINE|FAIL_SECURE, compound + fail-closed (TDD) [skip-docs]

src/llm_ingestion_guard/disposition.py

@@ -0,0 +1,221 @@
+"""disposition — turn a Report (plus source-trust context) into a gate decision.
+
+This is the *caller-side* decision layer (design principle 4): detectors report
+findings; this module maps a :class:`~llm_ingestion_guard.report.Report` to one
+of three dispositions — ``WARN`` | ``QUARANTINE_REVIEW`` | ``FAIL_SECURE`` —
+under a source-trust :class:`Policy`. It imposes no blocking of its own; a
+pipeline decides whether to honour a ``FAIL_SECURE``.
+
+Two BRIEF principles drive the policy:
+
+* **§4.6 — fail-secure under compound signals.** A scan hit together with a
+  transform failure is treated as a probable forced-fallback attack and halts.
+  Several weaker findings escalate one tier. And when the scanner itself errors,
+  :func:`guard` fails *closed* — an un-scannable artifact is never persisted.
+* **§4.7 — disposition scales with trust.** The same finding does not carry the
+  same disposition for every source. Source-level trust (:class:`Trust`) and
+  intra-document provenance (:class:`Provenance`) both downgrade a region: a
+  spoofed ``<system>`` block inside a code fence is a likelier real payload than
+  the same string in authored prose. Invisible carriers and CRITICAL findings
+  are the exception — they have no legitimate place and block in any tier.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum
+from typing import Callable, Optional
+
+from .report import Report, Severity, severity_rank
+
+
+class Disposition(str, Enum):
+    """The gate decision, ordered by :data:`_DISPOSITION_RANK`."""
+
+    WARN = "warn"                            # non-blocking floor: proceed, logged
+    QUARANTINE_REVIEW = "quarantine_review"  # hold for human review
+    FAIL_SECURE = "fail_secure"              # halt; never persist
+
+
+class Trust(str, Enum):
+    """Source-level trust (BRIEF §4.7)."""
+
+    TRUSTED = "trusted"      # pinned, reputable, single-author (e.g. a changelog)
+    UNTRUSTED = "untrusted"  # community / scraped / user-generated
+
+
+class Provenance(str, Enum):
+    """Intra-document provenance of the scanned region (BRIEF §4.7)."""
+
+    PROSE = "prose"            # authored, en-locale prose — the highest-trust surface
+    CODE_FENCE = "code_fence"  # inside a code sample — a low-trust surface
+    LOCALIZED = "localized"    # a non-English string — a low-trust surface
+
+
+@dataclass(frozen=True)
+class Policy:
+    """A named source-trust policy. See the ``PRESET_*`` constants."""
+
+    trust: Trust
+    quarantine_default: bool = False  # any finding -> at least QUARANTINE_REVIEW
+
+
+@dataclass(frozen=True)
+class DispositionResult:
+    """The decision plus an auditable trail of which rules fired."""
+
+    disposition: Disposition
+    reasons: tuple[str, ...]
+    max_severity: Optional[Severity]
+
+
+# Invisible carriers have no legitimate place in a reference file: they block in
+# any tier regardless of trust or provenance (BRIEF §4.7).
+_CARRIER_LABELS = frozenset({
+    "sanitize:zero-width",
+    "sanitize:bidi-override",
+    "sanitize:unicode-tag",
+})
+
+_DISPOSITION_RANK = {
+    Disposition.WARN: 0,
+    Disposition.QUARANTINE_REVIEW: 1,
+    Disposition.FAIL_SECURE: 2,
+}
+
+
+def _more_severe(a: Disposition, b: Disposition) -> Disposition:
+    return a if _DISPOSITION_RANK[a] >= _DISPOSITION_RANK[b] else b
+
+
+def _escalate(disposition: Disposition) -> Disposition:
+    if disposition is Disposition.WARN:
+        return Disposition.QUARANTINE_REVIEW
+    return Disposition.FAIL_SECURE
+
+
+def _carrier_label(report: Report) -> Optional[str]:
+    return next((f.label for f in report.findings if f.label in _CARRIER_LABELS), None)
+
+
+def _is_compound(report: Report) -> bool:
+    """Two or more findings at MEDIUM+ — weaker signals that escalate together."""
+    strong = sum(
+        1 for f in report.findings
+        if severity_rank(f.severity) >= severity_rank(Severity.MEDIUM)
+    )
+    return strong >= 2
+
+
+def decide(
+    report: Report,
+    policy: Policy,
+    *,
+    provenance: Provenance = Provenance.PROSE,
+    transform_failed: bool = False,
+) -> DispositionResult:
+    """Map ``report`` to a :class:`Disposition` under ``policy``.
+
+    Rules apply in fail-secure precedence order; the most severe outcome wins.
+    """
+    reasons: list[str] = []
+    max_sev = report.max_severity()
+
+    # Overlay A — compound forced-fallback (§4.6): a scan hit plus a failed
+    # transform is a probable forced-fallback attack. Overrides everything.
+    if transform_failed and report.found:
+        reasons.append("compound-forced-fallback: transform failed with active findings")
+        return DispositionResult(Disposition.FAIL_SECURE, tuple(reasons), max_sev)
+
+    # Any-tier exceptions (§4.7): invisible carriers and CRITICAL findings block
+    # regardless of trust or provenance.
+    carrier = _carrier_label(report)
+    if carrier is not None:
+        reasons.append(f"any-tier: invisible carrier ({carrier})")
+        return DispositionResult(Disposition.FAIL_SECURE, tuple(reasons), max_sev)
+    if max_sev is Severity.CRITICAL:
+        reasons.append("any-tier: CRITICAL finding")
+        return DispositionResult(Disposition.FAIL_SECURE, tuple(reasons), max_sev)
+
+    # Effective low-trust: an untrusted source, or a low-trust region within an
+    # otherwise-trusted document (a code fence or a localized string).
+    low_trust = policy.trust is Trust.UNTRUSTED or provenance in (
+        Provenance.CODE_FENCE,
+        Provenance.LOCALIZED,
+    )
+
+    disposition = _base_disposition(report, max_sev, low_trust, policy, reasons)
+
+    # Overlay B — compound escalation (§4.6): several weaker signals escalate one
+    # tier even when each alone would only WARN.
+    if _is_compound(report):
+        escalated = _escalate(disposition)
+        if escalated is not disposition:
+            reasons.append("compound: >=2 findings at MEDIUM+ -> escalated one tier")
+            disposition = escalated
+
+    return DispositionResult(disposition, tuple(reasons), max_sev)
+
+
+def _base_disposition(
+    report: Report,
+    max_sev: Optional[Severity],
+    low_trust: bool,
+    policy: Policy,
+    reasons: list[str],
+) -> Disposition:
+    tier = "low" if low_trust else "high"
+    if max_sev is None:
+        disposition = Disposition.WARN
+        reasons.append("clean: no findings")
+    elif max_sev is Severity.HIGH:
+        disposition = Disposition.FAIL_SECURE if low_trust else Disposition.WARN
+        reasons.append(f"HIGH under {tier}-trust -> {disposition.value}")
+    elif max_sev is Severity.MEDIUM:
+        disposition = Disposition.QUARANTINE_REVIEW if low_trust else Disposition.WARN
+        reasons.append(f"MEDIUM under {tier}-trust -> {disposition.value}")
+    else:  # LOW or INFO
+        disposition = Disposition.WARN
+        reasons.append(f"{max_sev.value} -> WARN")
+
+    # quarantine_default floor (upload preset): any finding is held for review.
+    if policy.quarantine_default and report.found:
+        floored = _more_severe(disposition, Disposition.QUARANTINE_REVIEW)
+        if floored is not disposition:
+            reasons.append("quarantine-floor: untrusted upload, any finding -> QUARANTINE_REVIEW")
+            disposition = floored
+
+    return disposition
+
+
+def guard(
+    scan_fn: Callable[[], Report],
+    policy: Policy,
+    *,
+    provenance: Provenance = Provenance.PROSE,
+    transform_failed: bool = False,
+) -> DispositionResult:
+    """Run ``scan_fn`` and dispose the result, failing *closed* on error.
+
+    A scanner that raises (a missing detector dependency, a crash on crafted
+    input) must never yield an auto-persist: an un-scannable artifact is a BLOCK
+    (BRIEF §4.6, fail-closed).
+    """
+    try:
+        report = scan_fn()
+    except Exception as exc:  # noqa: BLE001 — fail closed on ANY scanner error
+        return DispositionResult(
+            Disposition.FAIL_SECURE,
+            (f"fail-closed: scanner error: {type(exc).__name__}",),
+            None,
+        )
+    return decide(report, policy, provenance=provenance, transform_failed=transform_failed)
+
+
+PRESET_TRUSTED_SOURCE = Policy(trust=Trust.TRUSTED)
+"""Pinned, reputable single-author source (a changelog): security vocabulary
+WARNs, while CRITICAL findings and invisible carriers still fail secure."""
+
+PRESET_USER_UPLOAD = Policy(trust=Trust.UNTRUSTED, quarantine_default=True)
+"""High-untrust user-upload / open-source inbox (the flagship consumer): any
+finding is held for QUARANTINE_REVIEW and CRITICAL hard-fails — over-blocking one
+upload is far cheaper than persisting a poisoned one (BRIEF §4.7)."""