gendesign/backend/tests/ops/test_3114_log_credential_scrub.py

"""Пароли не уезжают в Loki: скруббер учётных данных в Alloy (#3114).

Прод-факт. `postgres_exporter` при неудачном скрейпе печатает полный DSN:

    msg="error scraping dsn" dsn="postgresql://gendesign_reader:<ПАРОЛЬ>@tradein-postgres:5432/tradein?sslmode=disable"

Замер по Loki за сутки: 104 такие строки на инфра-хосте и 106 на продуктовом.
При ретенции 30 дней это порядка 6000 строк с паролями БД в хранилище, доступ
к которому даёт вход в Grafana.

ЧТО ПРОВЕРЕНО ЭКСПЕРИМЕНТОМ, А НЕ ПРЕДПОЛОЖЕНО. Первым решением напрашивалось
передать пароль отдельно от строки подключения (`DATA_SOURCE_URI` +
`DATA_SOURCE_USER` + `DATA_SOURCE_PASS_FILE`). Прогон на скретч-контейнерах с
настоящим файлом запросов показал, что это НЕ помогает: экспортер собирает
строку сам и логирует её целиком. Поэтому чистим на своей стороне, до отправки.

Скруббер — защита в глубину, а не замена причине: конкретно эта ошибка уходит
грантом `pg_monitor`. Но он ловит любой пароль в URL, включая те компоненты, о
которых мы ещё не знаем.

Тесты ниже берут выражение ИЗ КОНФИГА и применяют его к настоящей строке из
прода — проверяется фактическая маскировка, а не наличие нужных слов в файле.
"""

from __future__ import annotations

import re
from pathlib import Path

import pytest

REPO_ROOT = Path(__file__).resolve().parents[3]
CONFIGS = [
    REPO_ROOT / "ops" / "metrics" / "alloy" / "alloy-apps.alloy",
    REPO_ROOT / "ops" / "metrics" / "alloy" / "alloy-infra.alloy",
]

# Настоящая строка из Loki (пароль заменён на равный по форме).
LEAKED = (
    'time=2026-08-26T12:34:20.945Z level=ERROR source=postgres_exporter.go:681 '
    'msg="error scraping dsn" err="queryNamespaceMappings returned 1 errors" '
    'dsn="postgresql://gendesign_reader:5fc281ce5d1bcf612417dffcae51d82fe5c3db2d7c4b4eac'
    '@tradein-postgres:5432/tradein?sslmode=disable"'
)
SECRET = "5fc281ce5d1bcf612417dffcae51d82fe5c3db2d7c4b4eac"

# Строки, которые скруббер портить НЕ должен.
INNOCENT = [
    "GET https://example.com/path",
    "user@host",
    "redis://redis:6379/0",
    "https://metrics.gendsgn.ru/ingest/loki/api/v1/push",
]


def _expression(cfg: Path) -> str:
    assert cfg.is_file(), f"нет {cfg} — конфиг переехал, гейт ослеп"
    text = cfg.read_text(encoding="utf-8")
    assert "loki.process" in text and "scrub_credentials" in text, (
        f"{cfg.name}: скруббер пропал из конфига"
    )
    m = re.search(r'stage\.replace\s*\{[^}]*expression\s*=\s*"([^"]+)"', text, re.S)
    assert m, f"{cfg.name}: не нашёл expression у stage.replace"
    return m.group(1)


def _apply(expression: str, line: str) -> str:
    """Повторяет поведение Alloy: заменяется содержимое групп захвата."""
    pat = re.compile(expression)
    out, pos = [], 0
    for m in pat.finditer(line):
        for gi in range(1, (m.re.groups or 0) + 1):
            s, e = m.span(gi)
            out.append(line[pos:s])
            out.append("***")
            pos = e
    out.append(line[pos:])
    return "".join(out)


@pytest.mark.parametrize("cfg", CONFIGS, ids=lambda p: p.name)
def test_password_is_masked_in_real_leaked_line(cfg: Path) -> None:
    """Пароль из настоящей строки прода исчезает."""
    masked = _apply(_expression(cfg), LEAKED)
    assert SECRET not in masked, f"{cfg.name}: пароль пережил скруббер"
    assert "***" in masked, f"{cfg.name}: маскировка не применилась вовсе"


@pytest.mark.parametrize("cfg", CONFIGS, ids=lambda p: p.name)
def test_user_and_host_survive(cfg: Path) -> None:
    """Пользователь и адрес остаются — без них строка ошибки бесполезна.

    Ровно одна группа захвата: вторая затёрла бы имя пользователя, и диагностика
    «какая учётка не смогла подключиться» была бы потеряна.
    """
    masked = _apply(_expression(cfg), LEAKED)
    assert "gendesign_reader" in masked, f"{cfg.name}: затёрто имя пользователя"
    assert "tradein-postgres:5432" in masked, f"{cfg.name}: затёрт адрес"


@pytest.mark.parametrize("cfg", CONFIGS, ids=lambda p: p.name)
def test_innocent_lines_are_untouched(cfg: Path) -> None:
    """Обычные URL и почтовые адреса не портятся.

    Слишком жадный скруббер хуже отсутствующего: он молча исказит логи, и это
    заметят в тот момент, когда по ним будут разбирать аварию.
    """
    expr = _expression(cfg)
    for line in INNOCENT:
        assert _apply(expr, line) == line, f"{cfg.name}: испорчена строка {line!r}"


@pytest.mark.parametrize("cfg", CONFIGS, ids=lambda p: p.name)
def test_journal_goes_through_scrubber_not_straight_to_loki(cfg: Path) -> None:
    """Источник журнала направлен в скруббер, а не напрямую в loki.write.

    Обратный конец инварианта: сам по себе объявленный, но не включённый в
    конвейер скруббер не чистит ничего — и выглядит при этом как защита.
    """
    text = cfg.read_text(encoding="utf-8")
    src = re.search(r"loki\.source\.journal\s+\"[^\"]+\"\s*\{(.*?)\n\}", text, re.S)
    assert src, f"{cfg.name}: не нашёл loki.source.journal"
    body = src.group(1)
    assert "loki.process.scrub_credentials.receiver" in body, (
        f"{cfg.name}: журнал уходит мимо скруббера"
    )
    assert "loki.write.central.receiver" not in body, (
        f"{cfg.name}: журнал всё ещё пишет напрямую в Loki в обход скруббера"
    )