"""Structured logging configuration utilities.""" from __future__ import annotations import json import logging from datetime import datetime, timezone from typing import Any from aegis_gitea_mcp.request_context import get_request_id # Context keys whose values must never be written to logs verbatim. SENSITIVE_CONTEXT_KEYS = frozenset( {"token", "authorization", "cookie", "password", "secret", "api_key"} ) class JsonLogFormatter(logging.Formatter): """Format log records as JSON documents.""" def format(self, record: logging.LogRecord) -> str: """Serialize a log record to JSON.""" payload: dict[str, Any] = { "timestamp": datetime.now(timezone.utc).isoformat(), "level": record.levelname, "logger": record.name, "message": record.getMessage(), "request_id": get_request_id(), } context = getattr(record, "context", None) if isinstance(context, dict) and context: payload["context"] = context if record.exc_info: # Security decision: include only exception type to avoid stack leakage. exception_type = record.exc_info[0] if exception_type is not None: payload["exception_type"] = str(exception_type.__name__) return json.dumps(payload, separators=(",", ":"), ensure_ascii=True) def configure_logging(level: str) -> None: """Configure application-wide structured JSON logging. Args: level: Logging level string. """ logger = logging.getLogger() logger.setLevel(level.upper()) for handler in list(logger.handlers): logger.removeHandler(handler) stream_handler = logging.StreamHandler() stream_handler.setFormatter(JsonLogFormatter()) logger.addHandler(stream_handler) def sanitize_context(context: dict[str, Any]) -> dict[str, Any]: """Redact sensitive values from a logging context mapping. Keys whose lower-cased name is in ``SENSITIVE_CONTEXT_KEYS`` have their value replaced with ``"***"`` so credentials never reach the log sink. Args: context: Arbitrary key/value pairs intended for structured logging. Returns: A new mapping with sensitive values masked. """ return { key: ("***" if key.lower() in SENSITIVE_CONTEXT_KEYS else value) for key, value in context.items() } def log_event(logger: logging.Logger, level: int, event: str, **context: Any) -> None: """Emit a structured log event with a sanitized context payload. Args: logger: Logger to emit on. level: Standard ``logging`` level (e.g. ``logging.DEBUG``). event: Stable, machine-friendly event name (e.g. ``get_issue.start``). **context: Extra structured fields; sensitive keys are masked. """ logger.log(level, event, extra={"context": sanitize_context(context)}) def log_nullable_field(logger: logging.Logger, event: str, field_name: str, value: Any) -> None: """Log whether a parsed response field is ``None`` and its runtime type. Makes null/nullable field failures (such as a ``null`` ``labels`` array) obvious in logs without dumping the field's full contents. Args: logger: Logger to emit on. event: Stable event name for the field check. field_name: Name of the field being inspected. value: The parsed value to characterize. """ log_event( logger, logging.DEBUG, event, field=field_name, is_none=value is None, value_type=(type(value).__name__ if value is not None else None), )