"""Policy-based redaction without summarising or size-truncating visible data.""" from __future__ import annotations import re from dataclasses import dataclass from typing import Any _SECRET_KEYS = frozenset( { "api_key", "openai_api_key", "authorization", "access_token", "refresh_token", "token", "secret", "password", "cookie", "set-cookie", "base_url", "openai_api_base", } ) _PRIVACY_KEYS = frozenset({"email", "phone", "mobile", "id_card", "identity_number", "身份证", "手机号"}) _DROP_KEYS = frozenset( { "traceback", "stack", "stack_trace", "stacktrace", "exc_info", "exception_tb", "system_prompt", "internal_prompt", "hidden_prompt", "soul", "soul_md", "chain_of_thought", "thinking", "thinking_delta", } ) _SECRET_PATTERNS = ( re.compile(r"sk-[A-Za-z0-9_\-]{3,}"), re.compile(r"(?:Bearer\s+)[A-Za-z0-9\-._~+/]+=*", re.IGNORECASE), re.compile(r"(?:api[_-]?key\s*[:=]\s*)[^\s,;]+", re.IGNORECASE), re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----[\s\S]*?-----END [A-Z ]*PRIVATE KEY-----"), ) _PRIVACY_PATTERNS = ( re.compile(r"(? dict[str, str]: return {"path": self.path, "reason": self.reason, "action": self.action} @dataclass(frozen=True) class SanitizationResult: value: Any redactions: tuple[RedactionRecord, ...] = () def redact_text(text: str, *, path: str = "$", include_privacy: bool = True) -> SanitizationResult: current = text reasons: list[RedactionRecord] = [] for pattern in _SECRET_PATTERNS: replaced = pattern.sub("[REDACTED]", current) if replaced != current: reasons.append(RedactionRecord(path=path, reason="secret")) current = replaced if include_privacy: for pattern in _PRIVACY_PATTERNS: replaced = pattern.sub("[REDACTED]", current) if replaced != current: reasons.append(RedactionRecord(path=path, reason="privacy")) current = replaced return SanitizationResult(current, tuple(reasons)) def sanitize_for_user(value: Any, *, allow_public_reasoning: bool = False) -> SanitizationResult: records: list[RedactionRecord] = [] def walk(item: Any, path: str, *, public_reasoning: bool) -> Any: if isinstance(item, dict): explicit_public = bool(item.get("public_reasoning")) or str(item.get("reasoning_visibility") or "").lower() == "public" out: dict[Any, Any] = {} for key, child in item.items(): key_text = str(key) lowered = key_text.lower() child_path = f"{path}.{key_text}" if lowered in _DROP_KEYS: records.append(RedactionRecord(path=child_path, reason="internal_or_stack", action="removed")) continue if lowered in {"reasoning", "reasoning_delta"} and not (allow_public_reasoning and explicit_public): records.append(RedactionRecord(path=child_path, reason="hidden_reasoning", action="removed")) continue if lowered in _SECRET_KEYS: out[key] = "[REDACTED]" records.append(RedactionRecord(path=child_path, reason="secret")) continue if lowered in _PRIVACY_KEYS: out[key] = "[REDACTED]" records.append(RedactionRecord(path=child_path, reason="privacy")) continue out[key] = walk(child, child_path, public_reasoning=public_reasoning or explicit_public) return out if isinstance(item, list): return [walk(child, f"{path}[{index}]", public_reasoning=public_reasoning) for index, child in enumerate(item)] if isinstance(item, tuple): return [walk(child, f"{path}[{index}]", public_reasoning=public_reasoning) for index, child in enumerate(item)] if isinstance(item, str): result = redact_text(item, path=path) records.extend(result.redactions) return result.value return item return SanitizationResult(walk(value, "$", public_reasoning=False), tuple(records)) def redact_mapping(value: Any) -> Any: """Compatibility helper for model/tool adapters.""" return sanitize_for_user(value).value __all__ = ["RedactionRecord", "SanitizationResult", "redact_mapping", "redact_text", "sanitize_for_user"]