meida_backend/packages/shared/logging.py

"""Structured JSON logging with request/trace propagation and redaction.

NFR-SEC-03 / 架构文档 §11.2: 手机号、UnionID、图片地址、诊断结论、令牌
一律脱敏或不记录. Redaction runs as a structlog processor so it applies to
every log line regardless of who wrote it — there is no opt-in to forget.
"""

from __future__ import annotations

import logging
import re
import sys
from collections.abc import Iterable, MutableMapping
from typing import Any

import structlog

from packages.shared.context import context_snapshot

REDACTED = "***"

#: Field names whose values must never be logged in clear text.
SENSITIVE_KEYS: frozenset[str] = frozenset(
    {
        # credentials
        "authorization",
        "access_token",
        "refresh_token",
        "token",
        "id_token",
        "password",
        "secret",
        "api_key",
        "apikey",
        "jwt",
        "purchase_token",
        "refresh_token_hash",
        # personal identifiers
        "phone",
        "phone_number",
        "mobile",
        "id_card",
        "union_id",
        "unionid",
        "open_id",
        "openid",
        "device_fp",
        "client_ip",
        "nickname",
        # face / image material (架构文档 §11.2, CR-02)
        "face_image",
        "face_image_key",
        "face_key",
        "avatar_key",
        "image_key",
        "image_url",
        "photo_url",
        "cos_key",
        "presigned_url",
        # diagnosis output is sensitive derived data
        "diagnosis",
        "dimensions",
        "derived_features",
    }
)

_PHONE_RE = re.compile(r"(?<!\d)(1[3-9]\d)\d{4}(\d{4})(?!\d)")
_SK_RE = re.compile(r"\bsk-[A-Za-z0-9]{16,}\b")
_BEARER_RE = re.compile(r"(?i)\bbearer\s+[A-Za-z0-9._\-]{8,}")


def _scrub_text(value: str) -> str:
    value = _SK_RE.sub("sk-" + REDACTED, value)
    value = _BEARER_RE.sub("Bearer " + REDACTED, value)
    return _PHONE_RE.sub(r"\1****\2", value)


def _redact_value(value: Any, sensitive_keys: frozenset[str], depth: int = 0) -> Any:
    if depth > 6:
        return value
    if isinstance(value, str):
        return _scrub_text(value)
    if isinstance(value, dict):
        return {
            k: (
                REDACTED
                if isinstance(k, str) and k.lower() in sensitive_keys
                else _redact_value(v, sensitive_keys, depth + 1)
            )
            for k, v in value.items()
        }
    if isinstance(value, (list, tuple)):
        return type(value)(_redact_value(v, sensitive_keys, depth + 1) for v in value)
    return value


def make_redaction_processor(
    extra_keys: Iterable[str] = (),
) -> Any:
    sensitive = SENSITIVE_KEYS | {k.lower() for k in extra_keys}

    def processor(
        _logger: Any, _name: str, event_dict: MutableMapping[str, Any]
    ) -> MutableMapping[str, Any]:
        for key in list(event_dict):
            if isinstance(key, str) and key.lower() in sensitive:
                event_dict[key] = REDACTED
            else:
                event_dict[key] = _redact_value(event_dict[key], sensitive)
        return event_dict

    return processor


def _context_processor(
    _logger: Any, _name: str, event_dict: MutableMapping[str, Any]
) -> MutableMapping[str, Any]:
    for key, value in context_snapshot().items():
        event_dict.setdefault(key, value)
    return event_dict


def configure_logging(
    *,
    level: str = "INFO",
    json_output: bool = True,
    extra_redact_keys: Iterable[str] = (),
) -> None:
    """Idempotently configure stdlib logging + structlog."""
    logging.basicConfig(
        format="%(message)s",
        stream=sys.stdout,
        level=getattr(logging, level.upper(), logging.INFO),
        force=True,
    )
    for noisy in ("uvicorn.access", "httpx", "httpcore"):
        logging.getLogger(noisy).setLevel(logging.WARNING)

    renderer: Any = (
        structlog.processors.JSONRenderer(ensure_ascii=False)
        if json_output
        else structlog.dev.ConsoleRenderer(colors=False)
    )

    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,
            structlog.processors.add_log_level,
            structlog.processors.TimeStamper(fmt="iso", utc=True),
            _context_processor,
            make_redaction_processor(extra_redact_keys),
            structlog.processors.StackInfoRenderer(),
            structlog.processors.format_exc_info,
            renderer,
        ],
        wrapper_class=structlog.make_filtering_bound_logger(
            getattr(logging, level.upper(), logging.INFO)
        ),
        logger_factory=structlog.PrintLoggerFactory(file=sys.stdout),
        cache_logger_on_first_use=True,
    )


def get_logger(name: str | None = None) -> Any:
    return structlog.get_logger(name)