gendesign/tradein-mvp/backend/tests/test_3154_query_secret_scrub.py

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

Прод-факт: uvicorn access-log печатал полный путь вместе с `?secret=<64 hex>`
(секрет вебхука GlitchTip = `TRADEIN_INTERNAL_AUTH_SECRET`), лог уезжал в Loki и
лежал там открытым. Скруббер Alloy (#3115) ловит только форму `user:pass@host`.

Проверка ПО ЗНАЧЕНИЮ: строка гоняется через настоящий handler с фильтром, в
выводе должно быть `secret=***` и НЕ должно быть самого секрета.
"""

from __future__ import annotations

import os

os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

import io
import logging

import pytest

from app.core.log_scrub import QuerySecretFilter, install_query_secret_filter, scrub_query_secrets

_SECRET = "5fc281ce5d1bcf612417dffcae51d82fe5c3db2d7c4b4eac9a1b2c3d4e5f60718"

# Настоящая форма access-строки uvicorn (msg + args, путь лежит в args).
_ACCESS_MSG = '%s - "%s %s HTTP/%s" %d'


def _emit(logger_name: str, msg: str, *args: object) -> str:
    """Прогоняет запись через настоящий handler и возвращает текст строки."""
    stream = io.StringIO()
    handler = logging.StreamHandler(stream)
    handler.setFormatter(logging.Formatter("%(message)s"))
    logger = logging.getLogger(logger_name)
    logger.addHandler(handler)
    logger.setLevel(logging.INFO)
    previous_propagate = logger.propagate
    logger.propagate = False
    try:
        logger.info(msg, *args)
    finally:
        logger.propagate = previous_propagate
        logger.removeHandler(handler)
    return stream.getvalue()


def test_uvicorn_access_line_masks_secret_query_param() -> None:
    install_query_secret_filter("uvicorn.access")
    line = _emit(
        "uvicorn.access",
        _ACCESS_MSG,
        "172.18.0.3:60322",
        "POST",
        f"/api/v1/trade-in/ops/glitchtip-webhook?secret={_SECRET}",
        "1.1",
        200,
    )
    assert "secret=***" in line, line
    assert _SECRET not in line, line
    # Остальная строка цела — иначе access-лог перестал бы годиться для разбора.
    assert "POST /api/v1/trade-in/ops/glitchtip-webhook?secret=*** HTTP/1.1" in line
    assert "172.18.0.3:60322" in line and "200" in line


def test_app_logger_via_root_handler_masks_secret() -> None:
    """Фильтр стоит и на handler'ах корня — прикладные логгеры тоже закрыты.

    Секрет + query-контекст (`?token=...`) передаются ОДНИМ %s-аргументом — как реально
    устроены вызовы в этом кодбейзе (см. body_preview в app/services/dadata.py), а не
    расщеплены между msg-шаблоном и голым значением в args. Фильтр (#3471) скрабит
    ЭЛЕМЕНТЫ `record.args` по отдельности, не трогая arity/`record.msg`-шаблон — поэтому
    секрет обязан быть самодостаточным (с контекстом) внутри одного аргумента.
    """
    stream = io.StringIO()
    handler = logging.StreamHandler(stream)
    handler.setFormatter(logging.Formatter("%(message)s"))
    handler.addFilter(QuerySecretFilter())
    root = logging.getLogger()
    root.addHandler(handler)
    try:
        logging.getLogger("app.some.module").warning(
            "retry callback %s", f"https://gendsgn.ru/hook?token={_SECRET}"
        )
    finally:
        root.removeHandler(handler)
    out = stream.getvalue()
    assert "token=***" in out, out
    assert _SECRET not in out, out


@pytest.mark.parametrize(
    "raw",
    [
        f"/hook?secret={_SECRET}",
        f"/hook?SECRET={_SECRET}",
        f"/hook?a=1&token={_SECRET}&b=2",
        f"/hook?api_key={_SECRET}",
        f"/hook?apiKey={_SECRET}",
        f'"GET /hook?access_token={_SECRET} HTTP/1.1"',
        # Имя с префиксом: чувствительное слово в КОНЦЕ имени параметра.
        f"/hook?client_secret={_SECRET}",
        f"/hook?webhook_secret={_SECRET}",
        f"/hook?refresh_token={_SECRET}",
        f"/hook?auth_token={_SECRET}",
    ],
)
def test_sensitive_param_names_are_masked(raw: str) -> None:
    scrubbed = scrub_query_secrets(raw)
    assert _SECRET not in scrubbed, scrubbed
    assert "***" in scrubbed


@pytest.mark.parametrize(
    "raw",
    [
        "GET /api/v1/trade-in/offers?limit=50&city=Екатеринбург",
        "https://metrics.gendsgn.ru/ingest/loki/api/v1/push",
        # Имя НЕ оканчивается на чувствительное слово — маскировать нечего.
        "/hook?secretary=anna",
        "/hook?tokens_page=2",
        # `token-info` в ПУТИ, а не в query: значения там нет вовсе.
        "GET /api/v1/token-info?limit=5",
    ],
)
def test_innocent_lines_untouched(raw: str) -> None:
    assert scrub_query_secrets(raw) == raw


def test_filter_installed_by_app_main() -> None:
    """Проводка: импорт приложения ставит фильтр на access-лог uvicorn."""
    import app.main  # noqa: F401  (импорт ради побочного эффекта установки фильтра)

    filters = logging.getLogger("uvicorn.access").filters
    assert any(isinstance(f, QuerySecretFilter) for f in filters), filters


def test_install_is_idempotent() -> None:
    install_query_secret_filter("uvicorn.access")
    install_query_secret_filter("uvicorn.access")
    filters = logging.getLogger("uvicorn.access").filters
    assert sum(isinstance(f, QuerySecretFilter) for f in filters) == 1, filters


def test_uvicorn_access_formatter_survives_secret_scrub() -> None:
    """Прод-баг: скраб секрета в access-логе ронял сам процесс логирования (docker logs).

    `uvicorn.logging.AccessFormatter.formatMessage()` распаковывает `record.args` НАПРЯМУЮ
    как 5-tuple (`client_addr, method, full_path, http_version, status_code = record.args`),
    в обход `record.getMessage()`. Старая реализация фильтра при найденном секрете обнуляла
    `record.args = ()` → `ValueError: not enough values to unpack (expected 5, got 0)` →
    `--- Logging error ---` в docker-логах ТОЛЬКО в момент, когда фильтр пытался защитить
    секрет (см. трассу через starlette/middleware/errors.py `_send` — это просто цепочка
    вложенных ASGI `send()`, ведущая к access-log вызову внутри uvicorn).
    """
    from uvicorn.logging import AccessFormatter

    stream = io.StringIO()
    handler = logging.StreamHandler(stream)
    handler.setFormatter(AccessFormatter("%(client_addr)s - %(request_line)s %(status_code)s"))
    logger = logging.getLogger("uvicorn.access.formatter_test")
    logger.addHandler(handler)
    logger.setLevel(logging.INFO)
    logger.propagate = False
    install_query_secret_filter("uvicorn.access.formatter_test")
    try:
        # НЕ должно бросить ValueError при обработке handler'ом (падение уходит в
        # handleError → stderr, а не наверх — поэтому проверяем результат, а не raises).
        logger.info(
            '%s - "%s %s HTTP/%s" %d',
            "172.18.0.3:60322",
            "POST",
            f"/api/v1/trade-in/ops/glitchtip-webhook?secret={_SECRET}",
            "1.1",
            200,
        )
    finally:
        logger.removeHandler(handler)
        logger.filters.clear()

    out = stream.getvalue()
    assert out, "форматтер не выдал ни строки — запись эмита провалилась (см. handleError)"
    assert _SECRET not in out, out
    assert "secret=***" in out, out
    assert "200" in out and "POST" in out