app/observability/sanitize.py

"""Attribute and log redaction for the observability boundary.

Every value that leaves the process boundary -- span attribute, metric
dimension, structured log field, exported observation -- passes through here
first. The global constraint from the plan preamble is absolute:

    "Never export prompts, full responses, PDF text, chunks, annotation
     contents, filesystem paths, URL query strings, base64, credentials,
     API headers, or raw document IDs."

Two complementary defences:
  * key-based -> a key naming forbidden content (``prompt``, ``chunk``,
    ``api_key``, ``document_id`` ...) has its *value* replaced with the
    ``[redacted]`` marker regardless of what the value is;
  * value-based -> a string value that *looks* like a secret, base64 blob,
    filesystem path, or URL query string is redacted/stripped even under an
    innocuous key, and oversized strings are truncated.

This module emits nothing; it only transforms values.
"""

from __future__ import annotations

import re
from typing import Any

REDACTED = "[redacted]"

# Oversized-string bounds. Attribute strings are metadata, not content, so a
# generous-but-bounded cap is enough to keep an accidental blob out of export.
MAX_ATTR_STRING = 256
MAX_EXCEPTION_CHARS = 500
_TRUNCATION_SUFFIX = "[truncated]"
SAFE_EXCEPTION_MESSAGE = "Exception details redacted."


# --- Forbidden keys -------------------------------------------------

# Word-level tokens: a key is split on ``. _ -`` and matched token-by-token so
# that "context_input_chars" (token {context,input,chars}) is NOT caught by the
# "text" token, while "prompt_text" (token {prompt,text}) is. Never add generic
# tokens like "key"/"id"/"document" here -- they collide with legitimate
# measures (document_diversity, requested_top_k). Those are handled by the
# narrower phrase list below plus value-based detection.
_FORBIDDEN_KEY_TOKENS: frozenset[str] = frozenset(
    {
        "prompt",
        "prompts",
        "response",
        "responses",
        "completion",
        "completions",
        "chunk",
        "chunks",
        "content",
        "contents",
        "text",
        "body",
        "message",
        "messages",
        "pdf",
        "annotation",
        "annotations",
        "note",
        "notes",
        "token",
        "tokens",
        "cookie",
        "cookies",
        "header",
        "headers",
        "auth",
        "password",
        "passwd",
        "secret",
        "secrets",
        "credential",
        "credentials",
        "bearer",
    }
)

# Substring phrases matched against the key with all separators removed, for
# compound names that survive token-splitting (``api_key`` -> "apikey",
# ``document_id`` -> "documentid"). Each phrase is chosen so it cannot appear
# inside a legitimate measure name.
_FORBIDDEN_KEY_PHRASES: tuple[str, ...] = (
    "apikey",
    "password",
    "secret",
    "authorization",
    "credential",
    "bearer",
    "accesstoken",
    "documentid",
    "docid",
)

_TOKEN_SPLIT = re.compile(r"[._\-\s]+")


def is_forbidden_key(key: str) -> bool:
    """True if ``key`` names content/credentials that must never be exported."""
    lowered = str(key).lower()
    tokens = {tok for tok in _TOKEN_SPLIT.split(lowered) if tok}
    if tokens & _FORBIDDEN_KEY_TOKENS:
        return True
    collapsed = re.sub(r"[^a-z0-9]", "", lowered)
    return any(phrase in collapsed for phrase in _FORBIDDEN_KEY_PHRASES)


# --- Value-based string detection -------------------------------------------------

# A URL: strip its query string and fragment, keep scheme/host/path.
_URL_RE = re.compile(r"^[a-z][a-z0-9+.\-]*://", re.IGNORECASE)

# Filesystem paths -- redacted whole. Unix absolute path of >=2 segments, a
# Windows drive/UNC path, or an explicit user-home reference.
_UNIX_PATH_RE = re.compile(r"(?:^|[\s'\"(=])(?:/[\w.\-]+){2,}")
_WIN_PATH_RE = re.compile(r"[A-Za-z]:[\\/]|\\\\[\w.\-]+\\")
# Exception messages often prefix a path with prose, so the broad detector
# above is insufficient for replacing the entire Windows path in-place.
_WIN_EXCEPTION_PATH_RE = re.compile(r"(?:[A-Za-z]:[\\/]|\\\\)[^\s'\"()=]*")

# Secret / opaque-blob shapes.
_DATA_URI_RE = re.compile(r"^data:[^;,]*;base64,", re.IGNORECASE)
_BASE64_RE = re.compile(r"^[A-Za-z0-9+/]{40,}={0,2}$")
_OPAQUE_TOKEN_RE = re.compile(r"^[A-Za-z0-9_\-]{32,}$")  # hashes, raw doc IDs, keys
_BEARER_RE = re.compile(r"(?i)\bbearer\b")
_SK_KEY_RE = re.compile(r"sk-[A-Za-z0-9]{12,}")
_INLINE_SECRET_RE = re.compile(r"(?i)(api[_\-]?key|secret|password|token)\s*[=:]\s*\S+")


def _looks_secret_or_blob(value: str) -> bool:
    if _DATA_URI_RE.match(value):
        return True
    if _BASE64_RE.match(value):
        return True
    if _OPAQUE_TOKEN_RE.match(value):
        return True
    if _BEARER_RE.search(value):
        return True
    if _SK_KEY_RE.search(value):
        return True
    if _INLINE_SECRET_RE.search(value):
        return True
    return False


def _looks_like_path(value: str) -> bool:
    if _WIN_PATH_RE.search(value):
        return True
    if _UNIX_PATH_RE.search(value):
        return True
    return False


def _sanitize_string(value: str) -> str:
    # URLs first: a query string like "?token=secret" would otherwise trip the
    # inline-secret detector, and "/v1/traces" would trip the path detector.
    if _URL_RE.match(value):
        stripped = value.split("?", 1)[0].split("#", 1)[0]
        return _truncate(stripped)
    if _looks_secret_or_blob(value):
        return REDACTED
    if _looks_like_path(value):
        return REDACTED
    return _truncate(value)


def _truncate(value: str, limit: int = MAX_ATTR_STRING) -> str:
    if len(value) <= limit:
        return value
    return value[:limit] + _TRUNCATION_SUFFIX


def sanitize_attribute_value(value: Any) -> Any:
    """Sanitize one attribute value.

    * ``bool``/``int``/``float`` -> unchanged (a metric-safe scalar).
    * ``str`` -> credential/base64/path redaction, URL query stripping,
      oversize truncation.
    * ``list``/``tuple`` of scalars/strings -> each element sanitized.
    * ``None`` and mappings -> ``None`` (dropped by :func:`sanitize_attributes`,
      since a nested mapping can smuggle forbidden content).
    """
    if isinstance(value, bool):
        return value
    if isinstance(value, (int, float)):
        return value
    if isinstance(value, str):
        return _sanitize_string(value)
    if isinstance(value, (list, tuple)):
        cleaned = [sanitize_attribute_value(item) for item in value]
        return [item for item in cleaned if item is not None]
    return None


def sanitize_attributes(attributes: dict[str, Any]) -> dict[str, Any]:
    """Return a redacted copy of ``attributes`` (input is never mutated).

    Forbidden keys keep their name but get the ``[redacted]`` marker as value;
    allowed keys get value-based sanitization; values that sanitize to ``None``
    (``None`` itself, nested mappings) are dropped entirely.
    """
    out: dict[str, Any] = {}
    for key, value in attributes.items():
        skey = str(key)
        if is_forbidden_key(skey):
            out[skey] = REDACTED
            continue
        sanitized = sanitize_attribute_value(value)
        if sanitized is None:
            continue
        out[skey] = sanitized
    return out


def sanitize_exception_text(text: str) -> str:
    """Strip filesystem paths from an exception message and bound its length."""
    # Remove Windows and Unix absolute paths, leaving a marker in their place.
    cleaned = _WIN_EXCEPTION_PATH_RE.sub(REDACTED, text)
    cleaned = re.sub(r"(?:/[\w.\-]+){2,}", REDACTED, cleaned)
    cleaned = cleaned.strip()
    if _looks_secret_or_blob(cleaned):
        return REDACTED
    return _truncate(cleaned, MAX_EXCEPTION_CHARS)


def sanitize_exception(exc: BaseException) -> tuple[str, str]:
    """Return safe exception metadata without inspecting ``str(exc)``.

    Exception text is untrusted product data: it can contain prompts, paper
    text, annotation contents, or even raise while rendering itself. Keep only
    the stable exception class and a fixed summary suitable for export.
    """
    try:
        exception_type = type(exc).__name__[:120]
    except BaseException:  # Defensive: this helper must never replace product flow.
        exception_type = "Exception"
    return exception_type, SAFE_EXCEPTION_MESSAGE