Spaces:
Sleeping
Sleeping
| """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 | |