llm-ingestion-pipeline-security/src/llm_ingestion_guard/disposition.py

"""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 .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 (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) — 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)."""