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, Mapping, Optional

from .calibration import DEFAULT_ACTION_MAP as _DEFAULT_ACTION_MAP_VALUES
from .calibration import DISPOSITION_RANK, RISK_RANK
from .report import Report, Severity, severity_rank


class Risk(str, Enum):
    """How dangerous the artifact is *given its source context* — the assessment.

    The other half of the 0.5.0 axis separation. :class:`Disposition` names an
    **action**; ``Risk`` names the **judgement that action was derived from**.
    Through 0.4.0 only the action was returned, which had two costs: a consumer
    who wanted different behaviour had to reinterpret an action whose reasoning
    was already gone, and ``NONE`` vs ``LOW`` — a clean document versus one
    carrying only low-severity findings — were indistinguishable, since both
    render as ``WARN``.

    Risk is **trust-aware**, exactly as BRIEF §4.7 describes the domain: the
    same finding genuinely *is* a different judgement in authored prose than in
    a code fence, not merely a different action taken on one shared judgement.
    """

    NONE = "none"          # no findings at all
    LOW = "low"            # findings present, none dispositive in this context
    ELEVATED = "elevated"  # suspicious enough to hold for a human
    SEVERE = "severe"      # treat as a real payload


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.

    ``action_map`` overrides how an assessed :class:`Risk` becomes a
    :class:`Disposition`. ``None`` — the default — means
    :data:`DEFAULT_ACTION_MAP`, which reproduces every disposition 0.4.0
    rendered. It is deliberately ``None`` rather than the map itself so an
    untouched ``Policy`` stays hashable exactly as before.

    This is the seam a consumer needs: wanting *hold for review* where we
    render *fail secure* is now a policy statement, not a reason to pin our
    grading.
    """

    trust: Trust
    quarantine_default: bool = False  # any finding -> at least QUARANTINE_REVIEW
    action_map: Optional[Mapping[Risk, Disposition]] = None


@dataclass(frozen=True)
class DispositionResult:
    """The decision plus an auditable trail of which rules fired.

    ``assessment`` is the risk the rules actually established; ``disposition``
    is that risk mapped through the policy's action map. Consumers that gate on
    the assessment are insulated from a later recalibration of the mapping.
    """

    disposition: Disposition
    reasons: tuple[str, ...]
    max_severity: Optional[Severity]
    assessment: Risk
    # Deliberately *required*, with no default. ``Risk.NONE`` would be the
    # natural-looking default and is precisely the wrong one: a construction
    # site that forgot the field would report a clean assessment, so the axis
    # would fail open. Fail-loud beats fail-silent for a gate (BRIEF §4.7).


# 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 ranks rebuilt from calibration's value-keyed source of truth
# (calibration is a leaf module and cannot import these enums without a cycle).
# Higher = more severe.
_DISPOSITION_RANK = {d: DISPOSITION_RANK[d.value] for d in Disposition}
_RISK_RANK = {r: RISK_RANK[r.value] for r in Risk}

DEFAULT_ACTION_MAP: Mapping[Risk, Disposition] = {
    Risk(risk_value): Disposition(disposition_value)
    for risk_value, disposition_value in _DEFAULT_ACTION_MAP_VALUES.items()
}
"""The default :class:`Risk` -> :class:`Disposition` mapping.

Reproduces every disposition 0.4.0 rendered, so the axis separation is additive:
a caller that never reads ``assessment`` sees no behavioural change.
"""


def _more_severe(a: Risk, b: Risk) -> Risk:
    return a if _RISK_RANK[a] >= _RISK_RANK[b] else b


def _escalate(risk: Risk) -> Risk:
    """Escalate one tier on the assessment axis.

    ``NONE`` is unreachable here — the only caller is the compound overlay,
    which needs two MEDIUM+ findings and so implies at least ``LOW`` — but it
    escalates rather than being a no-op, so the function is total.
    """
    if risk is Risk.SEVERE:
        return Risk.SEVERE
    return Risk(next(
        r for r in Risk if _RISK_RANK[r] == _RISK_RANK[risk] + 1
    ))


def _action(risk: Risk, policy: Policy) -> Disposition:
    """Map an assessed ``risk`` to an action under ``policy``.

    An action map that omits a risk level falls back to the default rather than
    raising: a partial override is a likely way to use this, and a ``KeyError``
    from inside the gate would be turned into a fail-closed by :func:`guard`
    anyway — silently, and with a useless reason.
    """
    if policy.action_map is not None and risk in policy.action_map:
        return policy.action_map[risk]
    return DEFAULT_ACTION_MAP[risk]


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()

    def result(risk: Risk) -> DispositionResult:
        return DispositionResult(_action(risk, policy), tuple(reasons), max_sev, risk)

    # 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 result(Risk.SEVERE)

    # 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 result(Risk.SEVERE)
    if max_sev is Severity.CRITICAL:
        reasons.append("any-tier: CRITICAL finding")
        return result(Risk.SEVERE)

    # 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,
    )

    risk = _base_risk(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. Escalating the *assessment*
    # rather than the action is what keeps the overlay from being silently lost
    # under a custom action map.
    if _is_compound(report):
        escalated = _escalate(risk)
        if escalated is not risk:
            reasons.append("compound: >=2 findings at MEDIUM+ -> escalated one tier")
            risk = escalated

    return result(risk)


def _base_risk(
    report: Report,
    max_sev: Optional[Severity],
    low_trust: bool,
    policy: Policy,
    reasons: list[str],
) -> Risk:
    """Assess ``report`` before the overlays, on the risk axis.

    The reason strings still name the *disposition* each branch yields, because
    they are an audit trail consumers already read; the axis separation must not
    silently reword it. They are rendered through :func:`_action` so a policy
    that remaps an action gets a trail that matches what it actually did.
    """
    tier = "low" if low_trust else "high"
    if max_sev is None:
        risk = Risk.NONE
        reasons.append("clean: no findings")
    elif max_sev is Severity.HIGH:
        risk = Risk.SEVERE if low_trust else Risk.LOW
        reasons.append(f"HIGH under {tier}-trust -> {_action(risk, policy).value}")
    elif max_sev is Severity.MEDIUM:
        risk = Risk.ELEVATED if low_trust else Risk.LOW
        reasons.append(f"MEDIUM under {tier}-trust -> {_action(risk, policy).value}")
    else:  # LOW or INFO
        risk = Risk.LOW
        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(risk, Risk.ELEVATED)
        if floored is not risk:
            reasons.append("quarantine-floor: MEDIUM+ finding -> QUARANTINE_REVIEW")
            risk = floored

    return risk


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
        # Both axes are pinned to their most severe value, and deliberately NOT
        # routed through the policy's action map: an un-scannable artifact is
        # not a risk judgement a caller gets to remap. A policy that downgrades
        # SEVERE means "I accept this class of finding", never "I accept a
        # scanner that crashed on crafted input" (BRIEF §4.6, fail closed).
        return DispositionResult(
            Disposition.FAIL_SECURE,
            (f"fail-closed: scan/dispose error: {type(exc).__name__}",),
            None,
            Risk.SEVERE,
        )


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