"""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 ```` 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 .calibration import DISPOSITION_RANK 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). Both the input-side # (``sanitize:*``, which strips) and the output-side presence labels are listed # so the invariant holds on the persist gate too — model output is never # sanitized, so its carrier signal arrives via ``output``/``lexicon`` instead. _CARRIER_LABELS = frozenset({ "sanitize:zero-width", "sanitize:bidi-override", "sanitize:unicode-tag", "output:zero-width-present", "output:bidi-present", "lexicon:unicode-tags-present", }) # Enum-keyed rank rebuilt from calibration's value-keyed source of truth # (calibration is a leaf module and cannot import the Disposition enum without a # cycle). Higher = more severe. _DISPOSITION_RANK = {d: DISPOSITION_RANK[d.value] for d in Disposition} 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: a finding at MEDIUM+ is held for review. # # Through 0.3.0 this floor fired on *any* finding, on the premise that a # finding is the exception. Adding the active-content detector broke that # premise — every ordinary markdown link became a finding — and the floor # then quarantined documents whose only sin was linking somewhere. Raising it # to MEDIUM+ restores the intent (hold what is actually suspicious) and is a # no-op for every detector that existed before 0.3.0: none of them emit LOW. if policy.quarantine_default and max_sev is not None and ( severity_rank(max_sev) >= severity_rank(Severity.MEDIUM) ): floored = _more_severe(disposition, Disposition.QUARANTINE_REVIEW) if floored is not disposition: reasons.append("quarantine-floor: MEDIUM+ 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) — or a disposition that raises (a malformed report) — must never yield an auto-persist: an un-scannable / un-disposable artifact is a BLOCK (BRIEF §4.6, fail-closed). ``decide`` runs inside the guarded block so the fail-closed guarantee is total. """ try: report = scan_fn() return decide(report, policy, provenance=provenance, transform_failed=transform_failed) except Exception as exc: # noqa: BLE001 — fail closed on ANY scan/dispose error return DispositionResult( Disposition.FAIL_SECURE, (f"fail-closed: scan/dispose error: {type(exc).__name__}",), None, ) 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)."""