gendesign/tradein-mvp/backend/app/core/log_scrub.py

"""Секреты из query-строки не попадают в лог процесса (#3154).

Прод-факт: uvicorn пишет в access-log ПОЛНЫЙ путь вместе с query, а лог уезжает
в Loki (ретенция 30 суток, доступ по входу в Grafana):

    INFO: 172.18.0.3:60322 - "POST /api/v1/trade-in/ops/glitchtip-webhook
                              ?secret=<64 hex> HTTP/1.1" 200 OK

Скруббер в Alloy (#3115) это не ловит: там одно выражение под форму
``scheme://user:pass@host`` (DSN postgres_exporter, #3114). Чиним в СВОЁМ
процессе — тогда секрета нет и в `docker logs`, до отправки куда-либо.

Фильтр вешается на логгер (`logging.Filter`), а не на форматтер: uvicorn.access
кладёт путь в ``record.args``, до форматирования он уже там. Поэтому берём
``record.getMessage()`` и, если что-то замаскировали, подменяем msg/args.
"""

from __future__ import annotations

import logging
import re

# Имена параметров, значение которых маскируем: имя ОКАНЧИВАЕТСЯ на чувствительное
# слово, поэтому перед альтернацией допускаем префикс (`client_secret`,
# `refresh_token`, `webhook_secret`). Значение — до следующего `&`, пробела или
# кавычки (access-строка uvicorn обрамляет запрос кавычками).
_SENSITIVE_QUERY = re.compile(
    r"([?&][\w.-]*(?:secret|token|api[-_]?key|apikey|access[-_]?token|password|signature|sig)=)"
    r"[^&\s\"'<>]+",
    re.IGNORECASE,
)


def scrub_query_secrets(text: str) -> str:
    """Заменяет значения чувствительных query-параметров на ``***``.

    Имя параметра и остальная строка сохраняются — иначе access-лог перестал бы
    годиться для диагностики.
    """
    return _SENSITIVE_QUERY.sub(r"\1***", text)


class QuerySecretFilter(logging.Filter):
    """Маскирует секреты в query-строке ЛЮБОЙ записи логгера, к которому привязан.

    Скрабит `record.msg` и КАЖДЫЙ элемент `record.args` по отдельности — НЕ схлопывает
    их в единую строку через `record.getMessage()` с последующим `args = ()`.

    Прод-баг (#3471): схлопывание ломало `uvicorn.access` — там `record.args` это
    структурный 5-tuple `(client_addr, method, full_path, http_version, status_code)`,
    который `uvicorn.logging.AccessFormatter.formatMessage()` распаковывает напрямую
    (`a, b, c, d, e = record.args`), в обход `record.getMessage()`. Как только фильтр
    находил секрет (например `?secret=` в webhook-пути) и обнулял `args`, форматтер падал
    с `ValueError: not enough values to unpack (expected 5, got 0)` — сама попытка
    заскрабить секрет ломала запись лога целиком (`--- Logging error ---` в докер-логах).
    """

    def filter(self, record: logging.LogRecord) -> bool:
        args = record.args
        if isinstance(args, tuple) and args:
            # Есть позиционные args — скрабим КАЖДЫЙ элемент отдельно, arity не трогаем.
            # `record.msg` (шаблон вида `"%s ..."`) не трогаем вовсе: в реальных вызовах
            # этого кодбейза секрет+query-контекст лежат САМОДОСТАТОЧНО внутри одного
            # аргумента (например body_preview в app/services/dadata.py), а не расщеплены
            # между текстом шаблона и голым значением — трогать msg тут не нужно и опасно
            # (шаблонный `%s` сам по себе мог бы ложно совпасть с чувствительным именем
            # параметра прямо перед ним).
            scrubbed_args = tuple(
                scrub_query_secrets(a) if isinstance(a, str) else a for a in args
            )
            if scrubbed_args != args:
                record.args = scrubbed_args
        elif isinstance(record.msg, str):
            # Args нет — вся запись уже готовым текстом в msg (f-string и т.п.).
            scrubbed_msg = scrub_query_secrets(record.msg)
            if scrubbed_msg != record.msg:
                record.msg = scrubbed_msg

        return True


# ── Секреты в произвольном тексте (тело ответа внешнего API) — #3471 ──────────
#
# DaData на HTTP 403 возвращает диагностику вида:
#   "Feature 'CLEAN' disabled for token '<действующий 40-символьный токен>'. See ..."
# body_preview из этого текста уходит в WARNING/ERROR лог (app/services/dadata.py) →
# docker logs → потенциально breadcrumb к любой последующей ошибке в GlitchTip.
# Маскируем ДО логирования. Не завязываемся на конкретную формулировку вендора —
# она может измениться (см. generic-слой ниже).

_TOKEN_QUOTED = re.compile(r"(\btoken\s*['\"])([^'\"]+)(['\"])", re.IGNORECASE)
# Любая hex/base64-подобная последовательность 24+ символов — ловит секрет независимо
# от контекста (Authorization/X-Secret значения, если когда-нибудь попадут в текст как есть).
_LONG_SECRET_LIKE = re.compile(r"[A-Za-z0-9+/_-]{24,}")


def _mask_value(value: str) -> str:
    """`8d4e…(40)` — первые 4 символа + длина в скобках, остальное скрыто."""
    if len(value) <= 4:
        return "***"
    return f"{value[:4]}…({len(value)})"


def scrub_body_secrets(text: str | None) -> str:
    """Маскирует токены/секреты в произвольном тексте (тело ответа внешнего API и т.п.).

    Двухслойно: 1) явный ``token '<value>'`` (текущая формулировка DaData на 403),
    2) generic — любая hex/base64-подобная последовательность 24+ символов, чтобы защита
    не зависела от того, как вендор сформулирует сообщение завтра.
    """
    if not text:
        return text or ""

    def _replace_quoted(m: re.Match[str]) -> str:
        return f"{m.group(1)}{_mask_value(m.group(2))}{m.group(3)}"

    scrubbed = _TOKEN_QUOTED.sub(_replace_quoted, text)
    scrubbed = _LONG_SECRET_LIKE.sub(lambda m: _mask_value(m.group(0)), scrubbed)
    return scrubbed


def install_query_secret_filter(*logger_names: str) -> None:
    """Вешает фильтр на access-лог uvicorn И на обработчики корневого логгера.

    Двумя местами, потому что uvicorn в своём log-config ставит `uvicorn.access`
    собственный handler с ``propagate = False`` — до корневого его записи не
    доходят. А фильтр на handler'ах корня закрывает всё остальное приложение
    (записи дочерних логгеров фильтры родителя не проходят, фильтры handler'а —
    проходят).

    Идемпотентно: повторный вызов не наплодит дублей.
    """
    targets: list[logging.Logger | logging.Handler] = [
        logging.getLogger(name) for name in logger_names or ("uvicorn.access",)
    ]
    targets.extend(logging.getLogger().handlers)
    for target in targets:
        if not any(isinstance(f, QuerySecretFilter) for f in target.filters):
            target.addFilter(QuerySecretFilter())