fix(tradein): секрет вебхука GlitchTip не попадает в access-log — скруббер query-секретов + приём из заголовка #3353

Merged
bot-backend merged 2 commits from fix/3154-glitchtip-secret-log-leak into main 2026-09-05 18:13:00 +00:00
5 changed files with 259 additions and 7 deletions

View file

@ -13,9 +13,11 @@ _send_uptime_generic``) в итоге идут через ОДНУ И ТУ ЖЕ
1) тело запроса для issue и uptime алертов структурно ОДИНАКОВОЕ
``{"text": str, "attachments": [{"title","title_link","text","color",
"fields",...}]}`` просто у uptime пустые/отсутствующие ``fields``/``color``;
2) единственный канал для аутентификации сам URL (как и Slack-вебхуки).
Секрет ОБЯЗАН ехать query-параметром, HTTP-заголовок здесь поставить
нечем (GlitchTip-сторона его не добавляет).
2) единственный канал для аутентификации от САМОГО GlitchTip сам URL (как и
у Slack-вебхуков): заголовок GlitchTip-сторона не добавляет. Поэтому
хендлер принимает секрет и из заголовка ``X-GlitchTip-Secret``
(предпочтительно не течёт в access-log, #3154), и из query-параметра
``?secret=`` как fallback для текущего отправителя.
Переиспользуем существующий ``TRADEIN_INTERNAL_AUTH_SECRET`` (#2213
defense-in-depth, см. ``app.core.rbac``) вместо нового секрета тот же
@ -47,7 +49,7 @@ import secrets
from datetime import UTC, datetime
from typing import Annotated, Any
from fastapi import APIRouter, HTTPException, Query, Request
from fastapi import APIRouter, Header, HTTPException, Query, Request
from pydantic import BaseModel, ConfigDict, ValidationError
from app.core.config import settings
@ -175,8 +177,9 @@ def _alerts_configured() -> bool:
def _verify_secret(provided: str) -> None:
expected = settings.tradein_internal_auth_secret
# constant-time: длина/префикс секрета не утекают через время ответа.
if not secrets.compare_digest(provided or "", expected):
logger.warning("glitchtip webhook: invalid or missing secret query param")
logger.warning("glitchtip webhook: invalid or missing secret")
raise HTTPException(status_code=401, detail="invalid or missing secret")
@ -184,18 +187,27 @@ def _verify_secret(provided: str) -> None:
async def glitchtip_webhook(
request: Request,
secret: Annotated[str, Query()] = "",
header_secret: Annotated[str, Header(alias="X-GlitchTip-Secret")] = "",
) -> dict[str, str]:
"""Приёмник GlitchTip webhook-алертов (issue + uptime) → пересылка в
Telegram-тему алертов (``TELEGRAM_ALERTS_CHAT_ID``/``TELEGRAM_ALERTS_TOPIC_ID``
ОТДЕЛЬНАЯ тема от support-топика, см. docstring модуля).
Путь публичный в ``rbac_guard`` (``app.core.rbac._PUBLIC_PATHS``) этот
хендлер сам делает единственную проверку (``secret`` query-параметр).
хендлер сам делает единственную проверку секрета.
Секрет принимается ИЗ ЗАГОЛОВКА ``X-GlitchTip-Secret``, а query-параметр
``?secret=`` остаётся fallback'ом (#3154). Заголовок предпочтителен потому,
что query едет в access-log и оттуда в Loki открытым текстом; query оставлен,
т.к. САМ GlitchTip 6.1.6 заголовков не шлёт вовсе (``send_webhook()``
``session.post(url, json=...)`` без headers, см. docstring модуля), и убрать
query можно только когда заголовок начнёт подставлять кто-то перед нами
(Caddy ``header_up`` на маршруте вебхука) либо после смены отправителя.
"""
if not _alerts_configured():
raise HTTPException(status_code=503, detail="glitchtip alerts webhook not configured")
_verify_secret(secret)
_verify_secret(header_secret or secret)
raw_body = await request.body()
received_at = datetime.now(UTC)

View file

@ -0,0 +1,72 @@
"""Секреты из 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-строке ЛЮБОЙ записи логгера, к которому привязан."""
def filter(self, record: logging.LogRecord) -> bool:
message = record.getMessage()
scrubbed = scrub_query_secrets(message)
if scrubbed != message:
record.msg = scrubbed
record.args = ()
return True
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())

View file

@ -44,6 +44,7 @@ from app.core.config import settings
from app.core.db import SessionLocal
from app.core.fdw import ensure_fdw_user_mapping
from app.core.http_errors import install_validation_error_handler
from app.core.log_scrub import install_query_secret_filter
from app.core.ratelimit import RateLimitMiddleware
from app.core.rbac import rbac_guard
from app.core.request_audit import RequestAuditMiddleware
@ -64,6 +65,11 @@ logging.basicConfig(
# закрытие, что уже стоит в tgbot_main.py (см. его комментарий), нужно и здесь.
logging.getLogger("httpx").setLevel(logging.WARNING)
# #3154: uvicorn access-log печатает полный путь С QUERY, а лог уезжает в Loki —
# так секрет вебхука GlitchTip (`?secret=…`) оказался в хранилище открытым.
# Маскируем значения чувствительных query-параметров ДО записи строки.
install_query_secret_filter()
# Мониторинг ошибок — GlitchTip (Sentry-совместимый, #396).
# DSN из env GLITCHTIP_DSN; пусто (dev/текущий prod) → init не вызывается, NO-OP.
# Integrations: Starlette/FastAPI (request errors), SQLAlchemy/Httpx (breadcrumbs),

View file

@ -0,0 +1,135 @@
"""Секрет из 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

View file

@ -159,6 +159,33 @@ def test_wrong_secret_401(client: TestClient, _fake_telegram_client: Any) -> Non
assert _fake_telegram_client.calls == []
def test_secret_accepted_from_header_without_query(
client: TestClient, _fake_telegram_client: Any
) -> None:
"""#3154: секрет можно прислать заголовком — тогда он не течёт в access-log."""
r = client.post(_ENDPOINT, json=_ISSUE_PAYLOAD, headers={"X-GlitchTip-Secret": _SECRET})
assert r.status_code == 200, r.text
assert len(_fake_telegram_client.calls) == 1
def test_wrong_header_secret_401(client: TestClient, _fake_telegram_client: Any) -> None:
r = client.post(_ENDPOINT, json=_ISSUE_PAYLOAD, headers={"X-GlitchTip-Secret": "wrong-value"})
assert r.status_code == 401
assert _fake_telegram_client.calls == []
def test_query_secret_still_accepted_as_fallback(
client: TestClient, _fake_telegram_client: Any
) -> None:
"""GlitchTip 6.1.6 заголовков не шлёт вовсе — query-путь обязан работать."""
r = client.post(f"{_ENDPOINT}?secret={_SECRET}", json=_ISSUE_PAYLOAD)
assert r.status_code == 200, r.text
assert len(_fake_telegram_client.calls) == 1
def test_secret_not_configured_returns_503_not_500(
client: TestClient, monkeypatch: pytest.MonkeyPatch, _fake_telegram_client: Any
) -> None: