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