Spaces:
Running
Running
File size: 2,980 Bytes
14c1acb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 | """Small, dependency-free logging safeguards for credential-bearing HTTP logs."""
from __future__ import annotations
import logging
import re
from urllib.parse import urlsplit, urlunsplit
_BEARER_RE = re.compile(r"(\bBearer\s+)[^\s,;\"']+", re.IGNORECASE)
_SECRET_ASSIGNMENT_RE = re.compile(
r"(\b(?:api[_-]?key|apikey|access[_-]?token|refresh[_-]?token|"
r"client[_-]?secret|bot[_-]?token|session(?:[_-]?string)?|"
r"authorization)\b\s*[:=]\s*)[^\s,;&\"']+",
re.IGNORECASE,
)
_URL_RE = re.compile(r"https?://[^\s<>\"']+", re.IGNORECASE)
class SecretRedactionFilter(logging.Filter):
"""Redact credentials from formatted log messages and exception traces."""
def filter(self, record: logging.LogRecord) -> bool:
try:
message = record.getMessage()
if record.exc_info:
try:
traceback_text = logging.Formatter().formatException(record.exc_info)
message = f"{message}\n{traceback_text}"
except Exception:
pass
record.exc_info = None
record.msg = sanitize_log_text(message)
record.args = ()
if record.stack_info:
record.stack_info = sanitize_log_text(record.stack_info)
except Exception:
# Logging must never break application execution.
pass
return True
def safe_url_for_log(url: str) -> str:
"""Return a URL with userinfo, query, and fragment removed."""
try:
parts = urlsplit(str(url))
if parts.scheme and parts.hostname:
hostname = parts.hostname
# IPv6 literals require brackets when reconstructed as a netloc.
host = f"[{hostname}]" if ":" in hostname else hostname
port = parts.port # ValueError for an invalid port; handled below.
if port is not None:
host = f"{host}:{port}"
return urlunsplit((parts.scheme, host, parts.path, "", ""))
except Exception:
pass
return "[REDACTED_URL]"
def sanitize_log_text(text: object) -> str:
"""Remove credential-bearing URL, bearer, and assignment values from text."""
value = str(text)
value = _URL_RE.sub(lambda match: safe_url_for_log(match.group(0)), value)
value = _BEARER_RE.sub(r"\1[REDACTED]", value)
value = _SECRET_ASSIGNMENT_RE.sub(r"\1[REDACTED]", value)
return value
def install_logging_safety() -> None:
"""Install defense-in-depth redaction and suppress verbose HTTP request logs."""
redactor = SecretRedactionFilter()
root = logging.getLogger()
for handler in root.handlers:
handler.addFilter(redactor)
# httpx/httpcore INFO request lines include complete URLs. Since Gemini puts
# its key in the query string, do not emit those automatic request lines.
logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("httpcore").setLevel(logging.WARNING)
|