137 lines
4.8 KiB
Python
137 lines
4.8 KiB
Python
"""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"(?<![\w.+-])[\w.+-]+@[\w.-]+\.[A-Za-z]{2,}(?![\w.-])"),
|
|
re.compile(r"(?<!\d)1[3-9]\d{9}(?!\d)"),
|
|
re.compile(r"(?<!\d)\d{17}[\dXx](?!\d)"),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RedactionRecord:
|
|
path: str
|
|
reason: str
|
|
action: str = "replaced"
|
|
|
|
def as_dict(self) -> 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"]
|