bot-bottle/bot_bottle/gateway/egress_dlp_config.py

"""Inspection and DLP configuration parsing for egress routes.

A route's optional `inspect:` object names which outbound/inbound detectors run
and what the proxy does when an outbound detector matches a token
(`outbound_on_match`). This module owns parsing and validating that block,
kept apart from the request-time scan/decision flow in `egress_addon_core`
so each half reads top-to-bottom without scrolling past the other.

Stdlib-only; ships flat into the gateway image alongside
`egress_addon_core.py` — see `Dockerfile.gateway`."""

from __future__ import annotations

import typing

OUTBOUND_DETECTOR_NAMES = frozenset({"token_patterns", "known_secrets", "entropy"})
INBOUND_DETECTOR_NAMES = frozenset({"naive_injection_detection"})

# Per-route policy for what the proxy does when an outbound DLP detector
# matches a token (PRD 0062).
ON_MATCH_BLOCK = "block"          # hard 403, never overridable
ON_MATCH_REDACT = "redact"        # scrub the matched value, forward the request
ON_MATCH_SUPERVISE = "supervise"  # queue for operator approval, hold the request
OUTBOUND_ON_MATCH_VALUES = (ON_MATCH_BLOCK, ON_MATCH_REDACT, ON_MATCH_SUPERVISE)
# Unset resolves to supervise (fall back to block when supervise is not wired).
DEFAULT_OUTBOUND_ON_MATCH = ON_MATCH_SUPERVISE


def parse_inspect_block(
    idx: int,
    host: str,
    inspect: dict[str, object],
) -> tuple[tuple[str, ...] | None, tuple[str, ...] | None, str]:
    """Parse DLP settings from an inspected route."""
    label = f"route[{idx}] ({host})"
    dlp = inspect

    def _parse_detector_field(
        field: str,
        valid_names: frozenset[str],
    ) -> tuple[str, ...] | None:
        val = dlp.get(field)
        if val is None:
            return None
        if val is False:
            return ()
        if not isinstance(val, list):
            raise ValueError(
                f"{label}: inspect.{field} must be false, a list, or omitted"
            )
        items = typing.cast(list[object], val)
        names: list[str] = []
        for j, item in enumerate(items):
            if not isinstance(item, str):
                raise ValueError(
                    f"{label}: inspect.{field}[{j}] must be a string"
                )
            if item not in valid_names:
                raise ValueError(
                    f"{label}: inspect.{field}[{j}] {item!r} is not a valid "
                    f"detector name; valid names: {', '.join(sorted(valid_names))}"
                )
            names.append(item)
        return tuple(names)

    outbound = _parse_detector_field("outbound_detectors", OUTBOUND_DETECTOR_NAMES)
    inbound = _parse_detector_field("inbound_detectors", INBOUND_DETECTOR_NAMES)

    on_match = ""
    on_match_raw = dlp.get("outbound_on_match")
    if on_match_raw is not None:
        if not isinstance(on_match_raw, str) or on_match_raw not in OUTBOUND_ON_MATCH_VALUES:
            raise ValueError(
                f"{label}: inspect.outbound_on_match must be one of "
                f"{', '.join(OUTBOUND_ON_MATCH_VALUES)} (got {on_match_raw!r})"
            )
        on_match = on_match_raw

    return outbound, inbound, on_match