Developer
102 lines
4.5 KiB
Python
102 lines
4.5 KiB
Python
"""Logging helpers for the backend."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
from typing import Any
|
|
from urllib.parse import urlsplit
|
|
|
|
from pythonjsonlogger import jsonlogger
|
|
|
|
|
|
def _json_formatter() -> logging.Formatter:
|
|
"""Return a JSON formatter that preserves timestamp, level, logger, and message."""
|
|
return jsonlogger.JsonFormatter(
|
|
fmt="%(timestamp)s %(level)s %(name)s %(message)s",
|
|
rename_fields={"asctime": "timestamp", "levelname": "level"},
|
|
)
|
|
|
|
|
|
def _text_formatter() -> logging.Formatter:
|
|
"""Return the legacy plain-text formatter."""
|
|
return logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s")
|
|
|
|
|
|
def configure_logging(level_name: str | None = None, log_format: str | None = None) -> int:
|
|
"""Configure root logging once and return the numeric log level."""
|
|
resolved_name = (level_name or os.getenv("LOG_LEVEL", "INFO")).upper()
|
|
level = getattr(logging, resolved_name, logging.INFO)
|
|
use_json = (log_format or os.getenv("LOG_FORMAT", "text")).lower() == "json"
|
|
|
|
root = logging.getLogger()
|
|
if not root.handlers:
|
|
handler = logging.StreamHandler()
|
|
handler.setFormatter(_json_formatter() if use_json else _text_formatter())
|
|
logging.basicConfig(level=level, handlers=[handler])
|
|
else:
|
|
# Reconfigure existing handler formatters if the format changed.
|
|
formatter = _json_formatter() if use_json else _text_formatter()
|
|
for handler in root.handlers:
|
|
handler.setFormatter(formatter)
|
|
|
|
root.setLevel(level)
|
|
logging.getLogger("media_library_viewer_api").setLevel(level)
|
|
logging.getLogger("uvicorn").setLevel(level)
|
|
logging.getLogger("uvicorn.error").setLevel(level)
|
|
logging.getLogger("uvicorn.access").setLevel(level)
|
|
logging.getLogger("paramiko").setLevel(logging.WARNING)
|
|
logging.getLogger("urllib3").setLevel(logging.WARNING)
|
|
return level
|
|
|
|
|
|
def _sanitize_url(url: str | None) -> str:
|
|
if not url:
|
|
return "<unset>"
|
|
parsed = urlsplit(url.strip())
|
|
if not parsed.scheme or not parsed.netloc:
|
|
return url.rstrip("/")
|
|
return f"{parsed.scheme}://{parsed.netloc}"
|
|
|
|
|
|
def describe_settings(settings: object) -> dict[str, str]:
|
|
"""Return a secret-safe summary of the current backend settings."""
|
|
return {
|
|
"log_level": str(getattr(settings, "log_level", "INFO") or "INFO").upper(),
|
|
"log_format": str(getattr(settings, "log_format", "text") or "text").lower(),
|
|
"auth_enabled": str(bool(getattr(settings, "auth_enabled", True))),
|
|
"oidc_issuer_url": _sanitize_url(getattr(settings, "oidc_issuer_url", "")),
|
|
"oidc_audience": getattr(settings, "oidc_audience", "") or "<unset>",
|
|
"oidc_jwks_url": _sanitize_url(getattr(settings, "oidc_jwks_url", "")),
|
|
"ssh_host": getattr(settings, "ssh_host", "") or "<unset>",
|
|
"ssh_username": getattr(settings, "ssh_username", "") or "<unset>",
|
|
"ssh_port": str(getattr(settings, "ssh_port", 22) or 22),
|
|
"ssh_key_directory": getattr(settings, "ssh_key_directory", "") or "<unset>",
|
|
"ssh_key_name": getattr(settings, "ssh_key_name", "") or "<unset>",
|
|
"ssh_password": "set" if getattr(settings, "ssh_password", "") else "missing",
|
|
"smtp_host": _sanitize_url(getattr(settings, "smtp_host", "")),
|
|
"smtp_port": str(getattr(settings, "smtp_port", 587) or 587),
|
|
"smtp_username": "set" if getattr(settings, "smtp_username", "") else "missing",
|
|
"smtp_from_address": getattr(settings, "smtp_from_address", "") or "<unset>",
|
|
"smtp_from_name": getattr(settings, "smtp_from_name", "") or "<unset>",
|
|
"smtp_use_tls": str(bool(getattr(settings, "smtp_use_tls", True))),
|
|
"smtp_use_ssl": str(bool(getattr(settings, "smtp_use_ssl", False))),
|
|
"remote_media_root": getattr(settings, "remote_media_root", "") or "<unset>",
|
|
"remote_path_prefix": getattr(settings, "remote_path_prefix", "") or "<unset>",
|
|
}
|
|
|
|
|
|
def sanitize_log_extra(extra: dict[str, Any] | None) -> dict[str, Any]:
|
|
"""Remove obvious secret keys from log extra fields before emission."""
|
|
if not extra:
|
|
return {}
|
|
blocked_suffixes = ("_password", "_secret", "_token", "_key", "private_key", "passphrase")
|
|
sanitized: dict[str, Any] = {}
|
|
for key, value in extra.items():
|
|
lower_key = key.lower()
|
|
if any(lower_key.endswith(suffix) for suffix in blocked_suffixes):
|
|
sanitized[key] = "<redacted>" if value else ""
|
|
else:
|
|
sanitized[key] = value
|
|
return sanitized
|