185 lines
6.6 KiB
Python
185 lines
6.6 KiB
Python
"""Recursive redaction and leak checks for evidence values."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from collections.abc import Mapping
|
|
from typing import Any, Iterable, Pattern
|
|
|
|
|
|
_REDACTED = "[REDACTED]"
|
|
_REDACTED_NAME = "[REDACTED_CREDENTIAL_NAME]"
|
|
DEFAULT_CREDENTIAL_NAME_PATTERNS = (
|
|
r"(?<![A-Za-z0-9])OPENAI_API_KEY(?![A-Za-z0-9])",
|
|
r"(?<![A-Za-z0-9])ANTHROPIC_API_KEY(?![A-Za-z0-9])",
|
|
r"(?<![A-Za-z0-9])MORPH_API_KEY(?![A-Za-z0-9])",
|
|
r"(?<![A-Za-z0-9])GITHUB_TOKEN(?![A-Za-z0-9])",
|
|
r"(?<![A-Za-z0-9])GH_TOKEN(?![A-Za-z0-9])",
|
|
r"(?<![A-Za-z0-9])SSH_AUTH_SOCK(?![A-Za-z0-9])",
|
|
r"(?<![A-Za-z0-9])AWS_(?:ACCESS_KEY_ID|SECRET_ACCESS_KEY)(?![A-Za-z0-9])",
|
|
r"(?<![A-Za-z0-9])(?:API_KEY|ACCESS_TOKEN|REFRESH_TOKEN|PRIVATE_KEY|PASSWORD|SECRET)(?![A-Za-z0-9])",
|
|
)
|
|
# This expression is deliberately applied to mapping keys only. Thus ordinary
|
|
# fields such as "status" and "provider" remain untouched.
|
|
_DEFAULT_KEY_PATTERN = re.compile(
|
|
r"^(?:[A-Z][A-Z0-9]*(?:_|-))*"
|
|
r"(?:API[_-]?KEY|TOKEN|SECRET|PASSWORD|PRIVATE[_-]?KEY|CREDENTIAL|SSH[_-]?AUTH[_-]?SOCK)$",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
|
|
class RedactionError(ValueError):
|
|
"""A value cannot be safely included in a committed summary."""
|
|
|
|
|
|
def _patterns(patterns: Iterable[str | Pattern[str]] | None) -> tuple[Pattern[str], ...]:
|
|
selected = (*DEFAULT_CREDENTIAL_NAME_PATTERNS, *(patterns or ()))
|
|
result: list[Pattern[str]] = []
|
|
for pattern in selected:
|
|
result.append(re.compile(pattern) if isinstance(pattern, str) else pattern)
|
|
return tuple(result)
|
|
|
|
|
|
def _is_credential_name(key: object, patterns: tuple[Pattern[str], ...]) -> bool:
|
|
if not isinstance(key, str):
|
|
return False
|
|
if _DEFAULT_KEY_PATTERN.fullmatch(key):
|
|
return True
|
|
return any(pattern.fullmatch(key) is not None for pattern in patterns)
|
|
|
|
|
|
def _redact_text(
|
|
value: str,
|
|
canaries: tuple[str, ...],
|
|
credential_patterns: tuple[Pattern[str], ...],
|
|
) -> str:
|
|
# Protect our own marker while applying user-supplied patterns. A pattern
|
|
# such as ``CREDENTIAL`` must not recursively redact the marker itself.
|
|
marker_sentinel = "\x00NONO_REDACTED_CREDENTIAL_NAME\x00"
|
|
result = value.replace(_REDACTED_NAME, marker_sentinel)
|
|
# Longer values first prevents a short canary from obscuring a longer one in
|
|
# the common case where one test fixture intentionally nests both.
|
|
for canary in sorted((item for item in canaries if item), key=len, reverse=True):
|
|
result = result.replace(canary, _REDACTED)
|
|
for pattern in credential_patterns:
|
|
result = pattern.sub(_REDACTED_NAME, result)
|
|
return result.replace(marker_sentinel, _REDACTED_NAME)
|
|
|
|
|
|
def redact(
|
|
value: Any,
|
|
*,
|
|
canaries: Iterable[str] = (),
|
|
credential_name_patterns: Iterable[str | Pattern[str]] | None = None,
|
|
) -> Any:
|
|
"""Return a recursively redacted copy of JSON-like evidence.
|
|
|
|
Credential patterns match mapping keys in full, and match credential-name
|
|
tokens in free text. This distinction protects safe field names from an
|
|
over-broad substring replacement.
|
|
"""
|
|
canary_values = tuple(str(item) for item in canaries if str(item))
|
|
patterns = _patterns(credential_name_patterns)
|
|
|
|
def visit(item: Any) -> Any:
|
|
if isinstance(item, Mapping):
|
|
result: dict[Any, Any] = {}
|
|
for key, child in item.items():
|
|
if _is_credential_name(key, patterns):
|
|
result[_REDACTED_NAME] = _REDACTED
|
|
else:
|
|
safe_key = visit(key) if isinstance(key, (str, int, float, bool)) else str(key)
|
|
result[safe_key] = visit(child)
|
|
return result
|
|
if isinstance(item, list):
|
|
return [visit(child) for child in item]
|
|
if isinstance(item, tuple):
|
|
return [visit(child) for child in item]
|
|
if isinstance(item, set):
|
|
return [visit(child) for child in sorted(item, key=str)]
|
|
if isinstance(item, BaseException):
|
|
return _redact_text(str(item), canary_values, patterns)
|
|
if isinstance(item, str):
|
|
return _redact_text(item, canary_values, patterns)
|
|
return item
|
|
|
|
return visit(value)
|
|
|
|
|
|
def scan_for_leaks(
|
|
value: Any,
|
|
*,
|
|
canaries: Iterable[str] = (),
|
|
credential_name_patterns: Iterable[str | Pattern[str]] | None = None,
|
|
) -> list[str]:
|
|
"""Return JSON-style paths where an unredacted canary/name is present."""
|
|
canary_values = tuple(str(item) for item in canaries if str(item))
|
|
patterns = _patterns(credential_name_patterns)
|
|
leaks: list[str] = []
|
|
|
|
def inspect(item: Any, path: str) -> None:
|
|
if isinstance(item, Mapping):
|
|
for key, child in item.items():
|
|
key_path = f"{path}.{key}" if path else str(key)
|
|
if _is_credential_name(key, patterns):
|
|
leaks.append(key_path)
|
|
if isinstance(key, str) and any(
|
|
canary and canary in key for canary in canary_values
|
|
):
|
|
leaks.append(key_path)
|
|
inspect(child, key_path)
|
|
return
|
|
if isinstance(item, (list, tuple, set)):
|
|
for index, child in enumerate(item):
|
|
inspect(child, f"{path}[{index}]")
|
|
return
|
|
if isinstance(item, BaseException):
|
|
inspect(str(item), path)
|
|
return
|
|
if isinstance(item, str):
|
|
if any(canary and canary in item for canary in canary_values):
|
|
leaks.append(path or "$")
|
|
# Do not rediscover the marker emitted by this module when a
|
|
# generic configured pattern includes the word CREDENTIAL.
|
|
scan_text = item.replace(_REDACTED_NAME, "")
|
|
if any(pattern.search(scan_text) is not None for pattern in patterns):
|
|
leaks.append(path or "$")
|
|
|
|
inspect(value, "")
|
|
return leaks
|
|
|
|
|
|
def assert_no_leaks(
|
|
value: Any,
|
|
*,
|
|
canaries: Iterable[str] = (),
|
|
credential_name_patterns: Iterable[str | Pattern[str]] | None = None,
|
|
) -> None:
|
|
leaks = scan_for_leaks(
|
|
value,
|
|
canaries=canaries,
|
|
credential_name_patterns=credential_name_patterns,
|
|
)
|
|
if leaks:
|
|
raise RedactionError("unredacted evidence at " + ", ".join(leaks))
|
|
|
|
|
|
# Explicit aliases make call sites read naturally and preserve a stable harness API.
|
|
redact_recursive = redact
|
|
redact_evidence = redact
|
|
leak_scan = scan_for_leaks
|
|
assert_clean = assert_no_leaks
|
|
|
|
|
|
__all__ = [
|
|
"DEFAULT_CREDENTIAL_NAME_PATTERNS",
|
|
"RedactionError",
|
|
"assert_clean",
|
|
"assert_no_leaks",
|
|
"leak_scan",
|
|
"redact",
|
|
"redact_evidence",
|
|
"redact_recursive",
|
|
"scan_for_leaks",
|
|
]
|