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'ах корня — прикладные логгеры тоже закрыты."""
    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 https://gendsgn.ru/hook?token=%s", _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