Spaces:
Running
Running
| """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) | |