104 lines
3.5 KiB
Python
104 lines
3.5 KiB
Python
"""Recursive sanitization for MAF roundtable events, payloads, and logs.
|
|
|
|
Strips credentials, tokens, cookies, passwords, secrets, and API keys from
|
|
nested data structures before they reach the event store, context builder,
|
|
or log output. Applied uniformly to tool arguments, tool results, fact
|
|
ledger entries, and event payloads.
|
|
|
|
Design:
|
|
- Pure functions, no I/O.
|
|
- Recursive: handles nested dicts, lists, and strings.
|
|
- Key-based: redacts values whose key matches a sensitive pattern.
|
|
- Value-based: redacts string values that look like tokens/keys.
|
|
- Never raises: malformed input passes through unchanged (best-effort).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import Any
|
|
|
|
_SENSITIVE_KEYS = frozenset({
|
|
"token", "access_token", "refresh_token", "id_token",
|
|
"authorization", "auth", "auth_token", "authtoken",
|
|
"cookie", "cookies", "set-cookie",
|
|
"password", "passwd", "pwd",
|
|
"secret", "secret_key", "client_secret",
|
|
"api_key", "apikey", "api-key",
|
|
"private_key", "private-key",
|
|
"credential", "credentials",
|
|
"session_id", "sessionid",
|
|
"x-api-key", "x-auth-token",
|
|
})
|
|
|
|
_KEY_PATTERN = re.compile(
|
|
r"(token|authorization|cookie|password|passwd|secret|api[_-]?key|private[_-]?key|credential|bearer)",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
_VALUE_PATTERNS = [
|
|
re.compile(r"^Bearer\s+\S+", re.IGNORECASE),
|
|
re.compile(r"^(sk|pk|ak|rk)[_-][A-Za-z0-9]{16,}$"),
|
|
re.compile(r"^eyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}"),
|
|
re.compile(r"^[A-Za-z0-9+/]{40,}={0,2}$"),
|
|
]
|
|
|
|
_INLINE_SECRET_PATTERN = re.compile(
|
|
r"(token|password|passwd|secret|api[_-]?key|authorization|cookie)\s*[=:]\s*\S+",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
_REDACTED = "***REDACTED***"
|
|
_MAX_DEPTH = 12
|
|
|
|
|
|
def sanitize(data: Any, *, _depth: int = 0) -> Any:
|
|
"""Recursively sanitize sensitive fields from a data structure."""
|
|
if _depth > _MAX_DEPTH:
|
|
return data
|
|
if isinstance(data, dict):
|
|
return _sanitize_dict(data, _depth)
|
|
if isinstance(data, (list, tuple)):
|
|
return [sanitize(item, _depth=_depth + 1) for item in data]
|
|
if isinstance(data, str):
|
|
return _sanitize_value_str(data)
|
|
return data
|
|
|
|
|
|
def sanitize_payload(payload: dict[str, Any] | None) -> dict[str, Any] | None:
|
|
"""Sanitize an event payload dict. Returns None if input is None."""
|
|
if payload is None:
|
|
return None
|
|
result = sanitize(payload)
|
|
return result if isinstance(result, dict) else payload
|
|
|
|
|
|
def _sanitize_dict(d: dict[str, Any], depth: int) -> dict[str, Any]:
|
|
out: dict[str, Any] = {}
|
|
for key, value in d.items():
|
|
key_lower = str(key).lower().replace("-", "_")
|
|
if key_lower in _SENSITIVE_KEYS or _KEY_PATTERN.search(key_lower):
|
|
out[key] = _REDACTED
|
|
elif isinstance(value, dict):
|
|
out[key] = _sanitize_dict(value, depth + 1)
|
|
elif isinstance(value, (list, tuple)):
|
|
out[key] = [sanitize(item, _depth=depth + 1) for item in value]
|
|
elif isinstance(value, str):
|
|
out[key] = _sanitize_value_str(value)
|
|
else:
|
|
out[key] = value
|
|
return out
|
|
|
|
|
|
def _sanitize_value_str(s: str) -> str:
|
|
# Never skip sanitization based on length — long strings (error traces,
|
|
# log dumps, seat-run errors) may embed credentials deep inside.
|
|
for pat in _VALUE_PATTERNS:
|
|
if pat.match(s):
|
|
return _REDACTED
|
|
if _INLINE_SECRET_PATTERN.search(s):
|
|
return _INLINE_SECRET_PATTERN.sub(
|
|
lambda m: m.group(0).split("=")[0].split(":")[0] + "=***REDACTED***", s
|
|
)
|
|
return s
|