LegacyHUB/app/common/json_logger.py

"""Structured JSON logging helpers shared across LegacyHUB (obs-core).

Two responsibilities, both required by the platform contract
(``INTER_MODULE_CONTRACT`` §8, ``03_INTEGRATION_STANDARDS`` §6,
``02_MODULE_CONTRACT`` §1.6):

- ``redact_secrets`` / ``mask_sensitive`` recursively replace secret-bearing
  values with ``***`` so API keys, tokens, passwords and authorization headers
  never reach stdout logs **or** persisted event payloads.
- ``build_event(component, event, level, **fields)`` emits a structured JSONL
  line with a stable ``component``/``event`` field contract.

The module is intentionally free of app-specific imports so it can be lifted
into a shared observability core unchanged.
"""

from __future__ import annotations

from collections.abc import Mapping
from typing import Any

import structlog

MASK = "***"

# Exact (case-insensitive) field names whose value must never be exposed.
SENSITIVE_KEYS: frozenset[str] = frozenset(
    {
        "api_key",
        "x_api_key",
        "x-api-key",
        "apikey",
        "api_key_hash",
        "token",
        "access_token",
        "refresh_token",
        "secret",
        "secret_key",
        "minio_secret_key",
        "qdrant_api_key",
        "kms_key_id",
        "password",
        "passwd",
        "authorization",
        "private_key",
        "client_secret",
    }
)

# Substrings that mark a field name as sensitive regardless of prefix. Kept
# narrow on purpose so neutral keys like ``object_key`` / ``idempotency_key`` /
# ``storage_key`` are never masked.
SENSITIVE_SUBSTRINGS: tuple[str, ...] = ("password", "secret", "authorization")


def is_sensitive_key(key: str) -> bool:
    """Return True if a field name should have its value masked."""
    lowered = key.lower()
    if lowered in SENSITIVE_KEYS:
        return True
    if lowered == "token" or lowered.endswith("_token"):
        return True
    return any(sub in lowered for sub in SENSITIVE_SUBSTRINGS)


def _mask(value: Any, *, key: str | None = None) -> Any:
    if key is not None and is_sensitive_key(key):
        return MASK if value is not None else None
    if isinstance(value, Mapping):
        return {k: _mask(v, key=str(k)) for k, v in value.items()}
    if isinstance(value, (list, tuple)):
        return [_mask(v) for v in value]
    return value


def redact_secrets(data: Any) -> Any:
    """Return a copy of ``data`` with all secret-bearing values masked.

    Use before persisting any structure that may carry credentials (event
    payloads, manifests) so DB rows never store raw secrets.
    """
    return _mask(data)


def mask_sensitive(
    _logger: Any, _method_name: str, event_dict: dict[str, Any]
) -> dict[str, Any]:
    """structlog processor that masks secret-bearing keys in the event dict."""
    return _mask(event_dict)


def build_event(component: str, event: str, level: str = "info", **fields: Any) -> dict[str, Any]:
    """Emit a structured JSONL log line and return the masked field map.

    ``component`` identifies the subsystem (e.g. ``"knowledge_ingest"``),
    ``event`` is a stable dotted event name. Returns the masked payload so
    callers/tests can assert on it without re-parsing stdout.
    """
    log = structlog.get_logger(component)
    emit = getattr(log, level.lower(), None) or log.info
    emit(event, component=component, **fields)
    return redact_secrets({"component": component, "event": event, "level": level, **fields})


__all__ = [
    "MASK",
    "SENSITIVE_KEYS",
    "build_event",
    "is_sensitive_key",
    "mask_sensitive",
    "redact_secrets",
]