fix(tradein/security): audit findings — proxy-key leak, admin audit, login status, lead IDOR
Четыре независимых security-audit находки:
1. rotate_proxy_ip (admin.py) отдавал str(httpx exc) клиенту и в Sentry —
mobileproxy changeip-URL несёт API-ключ в query-string. Ответ теперь
нейтральный ("changeip request failed"); sentry_scrub.py получил
composable full-text редактор секрет-подобных query-параметров
(?token=/?proxy_key=/?api_key=/... — не завязан на конкретного провайдера),
встроенный в scrub_pii_event (in-place, сохраняет event identity) — main.py
не тронут, редактор подключается автоматически через существующую ссылку.
2. Меняющие состояние /api/v1/admin/* ручки (куки, авто-логin, прокси,
настройки скрапера, bulk) были полностью исключены из user_events —
установить, кто их вызвал, было нельзя. RequestAuditMiddleware теперь
пишет `admin_action` для POST/PUT/PATCH/DELETE на /admin/* с атрибуцией
(username/ip/path/method/status), БЕЗ тела запроса (там секреты). GET
дашборды по-прежнему не логируются (design как раньше — не шумят).
3. login-событие писалось безусловно, без учёта response.status_code —
отражённая RBAC-попытка (протухший внутренний секрет / неизвестная роль /
scope-блок) была неотличима от настоящего входа. Теперь event_type
расходится на login/login_failed по фактическому статусу ответа; payload
несёт status_code. Дедуп-бакет (once/user+ip+ua+day) оставлен как есть —
разбивка на success/fail потребовала бы правки user_events.py (вне
scope); задокументировано как известный trade-off.
Побочный вопрос (шум basic_auth 401 от сканеров) не требует доп. фильтра
здесь: Caddy гейтит basic_auth ДО проксирования — трафик без валидного
X-Authenticated-User в этот код вообще не попадает.
4. POST /lead принимал estimate_id без проверки владельца — можно было
привязать заявку к чужой оценке. Переиспользован owner-or-admin guard
_assert_estimate_access из trade_in.py (тот же подход, что #690).
Тесты на каждый пункт (secret не в ответе/событии, admin_action с
атрибуцией и без body, login vs login_failed, чужой estimate_id -> 404) +
regression на существующие сьюты. Полный `pytest -q --deselect
tests/test_search_api.py::test_search_cache_hit`: 2635 passed, 8 skipped.
This commit is contained in:
bot-backend
parent
c9ba1b15ba
commit
2cfd112c53
7 changed files with 458 additions and 24 deletions
|
|
@ -2394,9 +2394,14 @@ async def rotate_proxy_ip(
|
|||
data = resp.json()
|
||||
except Exception:
|
||||
data = {}
|
||||
except Exception as exc:
|
||||
except Exception:
|
||||
# НЕ отдавать str(exc) клиенту (аудит-фикс, #security-audit): httpx-исключения
|
||||
# несут полный request URL, а rotate_url — mobileproxy changeip-ссылка с API-
|
||||
# ключом провайдера в query-string (?...&proxy_key=...). str(exc) с этим URL в
|
||||
# HTTP-ответе — прямая утечка секрета вызывающему клиенту. Причина сбоя остаётся
|
||||
# в логах (exc_info=True) для диагностики; наружу — только нейтральный reason.
|
||||
logger.warning("rotate-ip: changeip failed source=%s", source, exc_info=True)
|
||||
return RotateIpResponse(ok=False, reason=f"changeip error: {exc}")
|
||||
return RotateIpResponse(ok=False, reason="changeip request failed")
|
||||
|
||||
# changeip отдаёт новый IP в одном из полей (формат провайдер-зависимый).
|
||||
new_ip = None
|
||||
|
|
|
|||
|
|
@ -5,6 +5,16 @@ POST /api/v1/trade-in/lead — контактная заявка с резуль
|
|||
trade_in_leads. Notification (Telegram/email) — вне scope: нет существующей
|
||||
SMTP/Telegram интеграции в коде (подтверждено при разборе issue), только
|
||||
persist + log; `notified_at` в таблице зарезервирован под будущую доставку.
|
||||
|
||||
IDOR-фикс (security-audit): `estimate_id` раньше только проверялся на
|
||||
СУЩЕСТВОВАНИЕ (`SELECT 1 ... WHERE id = ...`), без проверки владельца — любой
|
||||
аутентифицированный пилот мог привязать свою заявку к чужой оценке (утечка через
|
||||
последующий просмотр лида: чужой адрес/телефон/оценка в заявке, которую видит не
|
||||
её владелец). Гвард переиспользует `_assert_estimate_access` из
|
||||
`app.api.v1.trade_in` — тот же owner-or-admin подход, что и `GET /estimate/{id}`
|
||||
(#690, `tests/test_estimate_idor.py`): 401 без `X-Authenticated-User`, 403 —
|
||||
неизвестная роль, 404 — оценка не найдена ИЛИ принадлежит не этому пользователю
|
||||
(существование чужой оценки не подтверждаем).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
|
@ -14,11 +24,12 @@ import re
|
|||
from typing import Annotated, Any, Literal
|
||||
from uuid import UUID
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, Request
|
||||
from fastapi import APIRouter, Depends, Header, HTTPException, Request
|
||||
from pydantic import BaseModel, Field, field_validator
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.api.v1.trade_in import _assert_estimate_access
|
||||
from app.core.db import get_db
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
|
@ -75,15 +86,20 @@ async def create_trade_in_lead(
|
|||
payload: TradeInLeadInput,
|
||||
request: Request,
|
||||
db: Annotated[Session, Depends(get_db)],
|
||||
x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Сохраняет лид (телефон + согласие) в trade_in_leads."""
|
||||
if payload.estimate_id is not None:
|
||||
exists = db.execute(
|
||||
text("SELECT 1 FROM trade_in_estimates WHERE id = CAST(:id AS uuid)"),
|
||||
estimate_row = db.execute(
|
||||
text("SELECT created_by FROM trade_in_estimates WHERE id = CAST(:id AS uuid)"),
|
||||
{"id": str(payload.estimate_id)},
|
||||
).fetchone()
|
||||
if exists is None:
|
||||
if estimate_row is None:
|
||||
raise HTTPException(status_code=404, detail="estimate not found")
|
||||
# IDOR guard (security-audit, зеркалит #690): нельзя привязать лид к
|
||||
# чужой оценке. 404 и на "не найдено", и на "чужая" — не подтверждаем
|
||||
# существование чужого estimate_id.
|
||||
_assert_estimate_access(estimate_row.created_by, x_authenticated_user)
|
||||
|
||||
user_agent = request.headers.get("user-agent")
|
||||
# 152-ФЗ audit trail: реальный клиентский IP из X-Forwarded-For (его ставит
|
||||
|
|
|
|||
|
|
@ -1,10 +1,27 @@
|
|||
"""RequestAuditMiddleware — пишет `api_request` (+ дедуплицированный `login`)
|
||||
события в `user_events` для каждого аутентифицированного `/api/*` запроса.
|
||||
"""RequestAuditMiddleware — пишет `api_request` / `admin_action` (+ дедуплицированный
|
||||
`login`/`login_failed`) события в `user_events` для каждого аутентифицированного
|
||||
`/api/*` запроса.
|
||||
|
||||
Foundation для Feature 2 (login/IP audit) и базы Feature 3 (behavior analytics).
|
||||
Логирование выполняется ПОСЛЕ `call_next` (не задерживает и не ветвит реальный
|
||||
ответ клиенту) и через fire-and-forget `schedule_event` — сбой аудита никогда
|
||||
не влияет на HTTP-ответ.
|
||||
|
||||
Порядок middleware-стека (см. `app/main.py`: `rbac_guard` — `@app.middleware("http")`,
|
||||
объявлен ДО `app.add_middleware(RequestAuditMiddleware)`) делает `RequestAudit`
|
||||
ВНЕШНИМ по отношению к `rbac_guard` (Starlette строит стек в обратном порядке
|
||||
регистрации — последний `add_middleware` оборачивает предыдущие). Поэтому к моменту,
|
||||
когда код ниже читает `response.status_code`, в нём уже отражён исход rbac_guard
|
||||
(401/403 short-circuit) ИЛИ реального хендлера — статус несёт реальный смысл
|
||||
"успех/отказ", а не только "запрос дошёл до хендлера".
|
||||
|
||||
Заведомо неаутентифицированный трафик (сканеры, долбящиеся в /wp-login.php и т.п.
|
||||
без валидного basic_auth) сюда вообще не попадает: Caddy гейтит basic_auth ПЕРЕД
|
||||
проксированием, так что `X-Authenticated-User` в таких запросах нет — условие
|
||||
`if username and ...` ниже их уже отсекает. Поэтому шум сканеров не нужно
|
||||
дополнительно фильтровать в этом файле — тот класс проблемы («сигнал тонет в шуме
|
||||
сканера») здесь структурно невозможен: событие может появиться только для
|
||||
запроса, прошедшего Caddy basic_auth.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
|
@ -25,6 +42,13 @@ logger = logging.getLogger(__name__)
|
|||
# циклическую зависимость.
|
||||
_PUBLIC_PATHS = frozenset({"/health", "/docs", "/redoc", "/openapi.json"})
|
||||
|
||||
# Методы, меняющие состояние — для /api/v1/admin/* именно они должны попадать в
|
||||
# аудит с атрибуцией (кто именно загрузил куки / включил авто-логин / поправил
|
||||
# прокси / изменил настройки скрапера / выполнил bulk-операцию). GET/HEAD/OPTIONS
|
||||
# на /admin/* остаются вне аудита (см. комментарий ниже — это просмотр дашбордов,
|
||||
# не действие).
|
||||
_MUTATING_METHODS = frozenset({"POST", "PUT", "PATCH", "DELETE"})
|
||||
|
||||
|
||||
class RequestAuditMiddleware(BaseHTTPMiddleware):
|
||||
"""Логирует активность аутентифицированных пользователей в `user_events`."""
|
||||
|
|
@ -38,32 +62,67 @@ class RequestAuditMiddleware(BaseHTTPMiddleware):
|
|||
if username and path.startswith("/api/") and path not in _PUBLIC_PATHS:
|
||||
ip = _client_ip(request)
|
||||
ua = request.headers.get("user-agent")
|
||||
method = request.method
|
||||
success = response.status_code < 400
|
||||
is_admin_path = path.startswith("/api/v1/admin/")
|
||||
|
||||
# Общий behavior/activity-поток — каждый authenticated API-запрос.
|
||||
# /api/v1/admin/* исключаем: это ops-действия (просмотр самих
|
||||
# дашбордов аудита/аналитики), а не поведение пилота — иначе
|
||||
# запросы дашборда зашумляют top_paths и счётчики активности.
|
||||
# login ниже логируем всегда (вход админа с IP — валидный аудит).
|
||||
if not path.startswith("/api/v1/admin/"):
|
||||
# /api/v1/admin/* исключаем из `api_request`: это ops-действия
|
||||
# (просмотр дашбордов аудита/аналитики), а не поведение пилота —
|
||||
# иначе запросы дашборда зашумляют top_paths и счётчики активности.
|
||||
if not is_admin_path:
|
||||
schedule_event(
|
||||
event_type="api_request",
|
||||
username=username,
|
||||
ip=ip,
|
||||
user_agent=ua,
|
||||
path=path,
|
||||
method=request.method,
|
||||
method=method,
|
||||
payload={"status_code": response.status_code},
|
||||
)
|
||||
|
||||
# Дедуплицированный login/IP-audit сигнал — максимум раз в день
|
||||
# на (юзер, IP, устройство).
|
||||
if should_log_login(username, ip, ua):
|
||||
elif method in _MUTATING_METHODS:
|
||||
# Admin-аудит (security-audit fix): раньше ЛЮБОЙ запрос под
|
||||
# /api/v1/admin/* (включая меняющие состояние — загрузка кук,
|
||||
# авто-логин, правка прокси, настройки скраперов, bulk-операции)
|
||||
# полностью исключался из `user_events` тем же условием, что и
|
||||
# шумные GET-дашборды — установить, КТО совершил действие, было
|
||||
# невозможно. Пишем факт действия + атрибуцию (username/ip/path/
|
||||
# method/статус) — БЕЗ тела запроса (там куки/пароли/секреты
|
||||
# правки прокси), это НЕ payload-лог, а событие "что произошло".
|
||||
schedule_event(
|
||||
event_type="login",
|
||||
event_type="admin_action",
|
||||
username=username,
|
||||
ip=ip,
|
||||
user_agent=ua,
|
||||
path=path,
|
||||
method=request.method,
|
||||
method=method,
|
||||
payload={"status_code": response.status_code, "success": success},
|
||||
)
|
||||
|
||||
# Дедуплицированный login/IP-audit сигнал — максимум раз в день
|
||||
# на (юзер, IP, устройство). Security-audit fix: раньше событие
|
||||
# всегда писалось как `login` независимо от исхода запроса —
|
||||
# отражённая RBAC-попытка (валидный Caddy basic_auth, но
|
||||
# 401/403 от rbac_guard: протухший X-Internal-Auth-Secret,
|
||||
# неизвестная роль, scope-блок) была неотличима от настоящего
|
||||
# входа. Теперь тип события расходится по `response.status_code`:
|
||||
# `login` — успех, `login_failed` — otказ. Дедуп-бакет (once per
|
||||
# user+ip+ua+day) НЕ разбит отдельно на success/fail (это
|
||||
# потребовало бы менять `should_log_login` в user_events.py —
|
||||
# вне scope этого фикса): если в рамках одного дня с этого же
|
||||
# устройства сначала случился отказ, а затем реальный успешный
|
||||
# вход, второе событие в тот же день не запишется — тот же
|
||||
# компромисс дедупа, что был и раньше, разница только в том, что
|
||||
# теперь ЕДИНСТВЕННОЕ событие дня корректно отражает, чем оно было.
|
||||
if should_log_login(username, ip, ua):
|
||||
schedule_event(
|
||||
event_type="login" if success else "login_failed",
|
||||
username=username,
|
||||
ip=ip,
|
||||
user_agent=ua,
|
||||
path=path,
|
||||
method=method,
|
||||
payload={"status_code": response.status_code},
|
||||
)
|
||||
except Exception:
|
||||
logger.warning("RequestAuditMiddleware: failed to record event", exc_info=True)
|
||||
|
|
|
|||
|
|
@ -47,6 +47,31 @@ _TG_BOT_TOKEN_REPLACEMENT = "/bot[REDACTED]"
|
|||
# `id:value` в логах, напр. `chat_id:12345`).
|
||||
_TG_BOT_TOKEN_BARE_RE = re.compile(r"\b\d{6,12}:[A-Za-z0-9_-]{30,}\b")
|
||||
|
||||
# Query-string секреты в исходящих URL сторонних API (аудит-фикс, #security-audit):
|
||||
# mobileproxy changeip-ссылка (`AVITO_PROXY_ROTATE_URL` и др., admin.py
|
||||
# rotate_proxy_ip) несёт провайдерский API-ключ в query (`?...&proxy_key=...`).
|
||||
# Два независимых пути утечки в GlitchTip, зеркалящих TG-токен выше:
|
||||
# 1. `HttpxIntegration.send()` парсит URL через `parse_url(str(request.url),
|
||||
# sanitize=False)` (ЯВНЫЙ opt-out из sentry_sdk `sanitize_url`, который иначе
|
||||
# сам вырезал бы query-параметры) и кладёт полный URL в span `data["url"]` —
|
||||
# сейчас неактивно (`traces_sample_rate=0.0` в app/main.py/scheduler_main.py →
|
||||
# span не сэмплится/не уходит), но молча перестанет спасать, если трейсинг
|
||||
# когда-нибудь включат.
|
||||
# 2. `include_local_variables=True` (sentry_sdk default в app/main.py — в отличие
|
||||
# от tgbot_main.py, где явно False) кладёт stack-frame locals (`rotate_url`,
|
||||
# `exc` в rotate_proxy_ip) в traceback открытым текстом.
|
||||
# Как и TG-токен — full-text regex по КАЖДОЙ строке event (не ключ-based): секрет
|
||||
# может всплыть где угодно (frame locals, breadcrumb, exception message). НЕ
|
||||
# завязано на конкретного провайдера — покрывает любой query-параметр из
|
||||
# общеупотребимого набора секретных имён (api_key/proxy_key/token/secret/password/
|
||||
# access_token/auth), т.к. cian/yandex у нас имеют СВОИ rotate-URL (потенциально
|
||||
# другой провайдер, другое имя параметра).
|
||||
_URL_SECRET_QUERY_RE = re.compile(
|
||||
r"(?i)([?&](?:api[_-]?key|proxy[_-]?key|token|secret|password|pwd|"
|
||||
r"access[_-]?token|auth)=)[^&\s\"'<>]+"
|
||||
)
|
||||
_URL_SECRET_QUERY_REPLACEMENT = r"\g<1>" + _REDACTED
|
||||
|
||||
|
||||
def _scrub(obj: Any) -> None:
|
||||
"""Рекурсивно заменить значения PII-ключей в dict на [REDACTED] (in-place)."""
|
||||
|
|
@ -61,8 +86,50 @@ def _scrub(obj: Any) -> None:
|
|||
_scrub(item)
|
||||
|
||||
|
||||
def _redact_url_secrets_inplace(obj: Any) -> None:
|
||||
"""Рекурсивно (IN-PLACE, как `_scrub`) заменяет значения секрет-подобных
|
||||
query-параметров (`?token=...`, `?proxy_key=...` и т.п.) на [REDACTED] в
|
||||
КАЖДОЙ строке event — не ключ-based: секрет утекает через httpx span
|
||||
`url`/`query` data и через текст исключений (`str(exc)` httpx содержит полный
|
||||
request URL), а не только через известные PII-поля формы. Мутирует dict/list
|
||||
на месте (НЕ пересоздаёт структуру, в отличие от `_redact_strings`) —
|
||||
сохраняет identity верхнеуровневого `event`, на что опирается контракт
|
||||
`scrub_pii_event`/`before_send` и существующие тесты (`out is event`).
|
||||
"""
|
||||
if isinstance(obj, dict):
|
||||
for key, value in obj.items():
|
||||
if isinstance(value, str):
|
||||
redacted = _URL_SECRET_QUERY_RE.sub(_URL_SECRET_QUERY_REPLACEMENT, value)
|
||||
if redacted != value:
|
||||
obj[key] = redacted
|
||||
else:
|
||||
_redact_url_secrets_inplace(value)
|
||||
elif isinstance(obj, list):
|
||||
for i, value in enumerate(obj):
|
||||
if isinstance(value, str):
|
||||
redacted = _URL_SECRET_QUERY_RE.sub(_URL_SECRET_QUERY_REPLACEMENT, value)
|
||||
if redacted != value:
|
||||
obj[i] = redacted
|
||||
else:
|
||||
_redact_url_secrets_inplace(value)
|
||||
# tuple намеренно не обрабатываем: sentry_sdk event — это JSON-совместимая
|
||||
# структура (dict/list/str/int/...), tuple там не встречается, а даже если бы
|
||||
# встретился — он immutable, in-place правка невозможна (см. `_scrub`, тот же
|
||||
# выбор для dict/list).
|
||||
|
||||
|
||||
def scrub_pii_event(event: Event, _hint: dict[str, Any]) -> Event | None:
|
||||
"""Redact consumer-PII из error event перед отправкой. Возвращает event (не None)."""
|
||||
"""Redact consumer-PII + URL query-string секретов из error event перед отправкой.
|
||||
|
||||
Композиция (обе — in-place, сохраняют identity `event`): (1) ключ-based
|
||||
dict-scrub consumer-PII полей формы (как раньше), (2) full-text regex-проход
|
||||
по ВСЕМУ event, вырезающий значения секрет-подобных query-параметров в любой
|
||||
строке (proxy/API-ключи в исходящих URL сторонних сервисов, напр. mobileproxy
|
||||
changeip — #security-audit). Второй шаг не завязан на конкретные ключи полей —
|
||||
ловит секрет в frame locals, breadcrumb, exception message и т.д., где он может
|
||||
оказаться независимо от include_local_variables/traces_sample_rate. Возвращает
|
||||
event (не None).
|
||||
"""
|
||||
if not isinstance(event, dict):
|
||||
return event
|
||||
request = event.get("request")
|
||||
|
|
@ -70,6 +137,7 @@ def scrub_pii_event(event: Event, _hint: dict[str, Any]) -> Event | None:
|
|||
_scrub(request.get("data"))
|
||||
_scrub(event.get("extra"))
|
||||
_scrub(event.get("contexts"))
|
||||
_redact_url_secrets_inplace(event)
|
||||
return event
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -15,7 +15,7 @@ os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:
|
|||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from fastapi import FastAPI
|
||||
from fastapi import FastAPI, Response
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.core.request_audit import RequestAuditMiddleware
|
||||
|
|
@ -138,3 +138,152 @@ def test_admin_path_excluded_from_api_request_but_login_kept() -> None:
|
|||
assert resp.status_code == 200
|
||||
event_types = [c.kwargs["event_type"] for c in mock_schedule.call_args_list]
|
||||
assert event_types == ["login"]
|
||||
|
||||
|
||||
# ── Admin audit (security-audit fix): mutating /admin/* -> admin_action ────────
|
||||
|
||||
|
||||
def test_admin_mutating_post_schedules_admin_action_with_attribution() -> None:
|
||||
"""POST на /api/v1/admin/* (напр. правка прокси / настройки скрапера) должен
|
||||
писать `admin_action` с атрибуцией (кто), а НЕ игнорироваться целиком, как
|
||||
раньше (security-audit: не было возможности установить, кто это сделал)."""
|
||||
app = FastAPI()
|
||||
app.add_middleware(RequestAuditMiddleware)
|
||||
|
||||
@app.post("/api/v1/admin/scraper/avito/rotate-ip")
|
||||
def rotate_ip() -> dict[str, bool]:
|
||||
return {"ok": True}
|
||||
|
||||
with (
|
||||
patch("app.core.request_audit.schedule_event") as mock_schedule,
|
||||
patch("app.core.request_audit.should_log_login", return_value=False),
|
||||
):
|
||||
resp = TestClient(app).post(
|
||||
"/api/v1/admin/scraper/avito/rotate-ip",
|
||||
headers={"X-Authenticated-User": "admin"},
|
||||
)
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert mock_schedule.call_count == 1
|
||||
kwargs = mock_schedule.call_args.kwargs
|
||||
assert kwargs["event_type"] == "admin_action"
|
||||
assert kwargs["username"] == "admin"
|
||||
assert kwargs["path"] == "/api/v1/admin/scraper/avito/rotate-ip"
|
||||
assert kwargs["method"] == "POST"
|
||||
assert kwargs["payload"] == {"status_code": 200, "success": True}
|
||||
|
||||
|
||||
def test_admin_get_does_not_schedule_admin_action() -> None:
|
||||
"""GET на /admin/* (просмотр дашборда) НЕ должен писать admin_action — только
|
||||
мутирующие методы считаются "действием"."""
|
||||
app = FastAPI()
|
||||
app.add_middleware(RequestAuditMiddleware)
|
||||
|
||||
@app.get("/api/v1/admin/scraper/health")
|
||||
def health() -> dict[str, bool]:
|
||||
return {"ok": True}
|
||||
|
||||
with (
|
||||
patch("app.core.request_audit.schedule_event") as mock_schedule,
|
||||
patch("app.core.request_audit.should_log_login", return_value=False),
|
||||
):
|
||||
TestClient(app).get(
|
||||
"/api/v1/admin/scraper/health", headers={"X-Authenticated-User": "admin"}
|
||||
)
|
||||
|
||||
mock_schedule.assert_not_called()
|
||||
|
||||
|
||||
def test_admin_action_payload_excludes_request_body() -> None:
|
||||
"""security-audit: тело запроса (куки/пароли/секреты правки прокси) НЕ должно
|
||||
попадать в audit-payload — только факт действия + атрибуция."""
|
||||
app = FastAPI()
|
||||
app.add_middleware(RequestAuditMiddleware)
|
||||
|
||||
@app.post("/api/v1/admin/scraper/cookies")
|
||||
def upload_cookies() -> dict[str, bool]:
|
||||
# Хендлер намеренно НЕ объявляет body-параметр — middleware проверяет
|
||||
# только headers/path/method/status, JSON-тело запроса ниже (куки) в
|
||||
# audit-payload попасть не может структурно, не только "по договорённости".
|
||||
return {"ok": True}
|
||||
|
||||
with (
|
||||
patch("app.core.request_audit.schedule_event") as mock_schedule,
|
||||
patch("app.core.request_audit.should_log_login", return_value=False),
|
||||
):
|
||||
TestClient(app).post(
|
||||
"/api/v1/admin/scraper/cookies",
|
||||
headers={"X-Authenticated-User": "admin"},
|
||||
json={"cookies": "super-secret-session-cookie"},
|
||||
)
|
||||
|
||||
kwargs = mock_schedule.call_args.kwargs
|
||||
assert kwargs["event_type"] == "admin_action"
|
||||
assert "super-secret-session-cookie" not in repr(kwargs)
|
||||
|
||||
|
||||
def test_admin_mutating_failure_status_recorded_in_payload() -> None:
|
||||
"""admin_action на неуспешный ответ (напр. 500 от нижестоящего сервиса) должен
|
||||
нести success=False + реальный status_code — не маскироваться под успех."""
|
||||
app = FastAPI()
|
||||
app.add_middleware(RequestAuditMiddleware)
|
||||
|
||||
@app.post("/api/v1/admin/scraper/pacing")
|
||||
def pacing() -> Response:
|
||||
return Response(status_code=502)
|
||||
|
||||
with (
|
||||
patch("app.core.request_audit.schedule_event") as mock_schedule,
|
||||
patch("app.core.request_audit.should_log_login", return_value=False),
|
||||
):
|
||||
TestClient(app).post(
|
||||
"/api/v1/admin/scraper/pacing", headers={"X-Authenticated-User": "admin"}
|
||||
)
|
||||
|
||||
kwargs = mock_schedule.call_args.kwargs
|
||||
assert kwargs["event_type"] == "admin_action"
|
||||
assert kwargs["payload"] == {"status_code": 502, "success": False}
|
||||
|
||||
|
||||
# ── login vs login_failed (security-audit fix) ──────────────────────────────────
|
||||
|
||||
|
||||
def test_login_event_type_when_request_succeeds(client: TestClient) -> None:
|
||||
"""Ответ < 400 -> event_type='login' (успешный вход/активность), payload несёт
|
||||
status_code."""
|
||||
with (
|
||||
patch("app.core.request_audit.schedule_event") as mock_schedule,
|
||||
patch("app.core.request_audit.should_log_login", return_value=True),
|
||||
):
|
||||
client.get("/api/v1/ping", headers={"X-Authenticated-User": "alice"})
|
||||
|
||||
login_calls = [c for c in mock_schedule.call_args_list if c.kwargs["event_type"] == "login"]
|
||||
assert len(login_calls) == 1
|
||||
assert login_calls[0].kwargs["payload"] == {"status_code": 200}
|
||||
|
||||
|
||||
def test_login_failed_event_type_when_rbac_rejects_request() -> None:
|
||||
"""Ответ >= 400 (напр. RBAC-отказ downstream: неизвестная роль / протухший
|
||||
внутренний секрет) -> event_type='login_failed', а НЕ 'login' — раньше эти
|
||||
два случая были неразличимы в журнале (security-audit)."""
|
||||
app = FastAPI()
|
||||
app.add_middleware(RequestAuditMiddleware)
|
||||
|
||||
@app.get("/api/v1/ping")
|
||||
def ping() -> Response:
|
||||
return Response(status_code=403, content="forbidden")
|
||||
|
||||
with (
|
||||
patch("app.core.request_audit.schedule_event") as mock_schedule,
|
||||
patch("app.core.request_audit.should_log_login", return_value=True),
|
||||
):
|
||||
TestClient(app).get("/api/v1/ping", headers={"X-Authenticated-User": "alice"})
|
||||
|
||||
login_calls = [
|
||||
c
|
||||
for c in mock_schedule.call_args_list
|
||||
if c.kwargs["event_type"] in ("login", "login_failed")
|
||||
]
|
||||
assert len(login_calls) == 1
|
||||
assert login_calls[0].kwargs["event_type"] == "login_failed"
|
||||
assert login_calls[0].kwargs["payload"] == {"status_code": 403}
|
||||
|
|
|
|||
|
|
@ -305,7 +305,44 @@ def test_rotate_ip_changeip_error(client: TestClient) -> None:
|
|||
assert r.status_code == 200
|
||||
body = r.json()
|
||||
assert body["ok"] is False
|
||||
assert "changeip error" in body["reason"]
|
||||
# security-audit: нейтральный reason, БЕЗ текста исходного исключения
|
||||
# (str(exc) httpx мог нести rotate_url с proxy-ключом в query — см. тест ниже).
|
||||
assert body["reason"] == "changeip request failed"
|
||||
|
||||
|
||||
def test_rotate_ip_changeip_error_does_not_leak_proxy_key(client: TestClient) -> None:
|
||||
"""security-audit: секретный API-ключ провайдера в rotate_url НЕ должен попасть
|
||||
в HTTP-ответ клиенту через текст httpx-исключения (раньше
|
||||
`reason=f"changeip error: {exc}"` отдавал str(exc) с полным URL, включая
|
||||
query-параметр ключа, наружу)."""
|
||||
from app.api.v1 import admin as admin_module
|
||||
|
||||
secret_url = "http://ch/changeip?proxy_key=TOP-SECRET-KEY-1234"
|
||||
|
||||
class _BoomClient:
|
||||
def __init__(self, *a: Any, **k: Any) -> None:
|
||||
pass
|
||||
|
||||
async def __aenter__(self) -> _BoomClient:
|
||||
return self
|
||||
|
||||
async def __aexit__(self, *a: Any) -> None:
|
||||
return None
|
||||
|
||||
async def get(self, *a: Any, **k: Any) -> Any:
|
||||
raise RuntimeError(f"All connection attempts failed for {secret_url}&format=json")
|
||||
|
||||
with (
|
||||
patch.object(admin_module.httpx, "AsyncClient", _BoomClient),
|
||||
patch.object(admin_module.settings, "avito_proxy_rotate_url", secret_url),
|
||||
):
|
||||
r = client.post("/api/v1/admin/scraper/avito/rotate-ip")
|
||||
|
||||
assert r.status_code == 200
|
||||
assert "TOP-SECRET-KEY-1234" not in r.text
|
||||
body = r.json()
|
||||
assert body["ok"] is False
|
||||
assert "TOP-SECRET-KEY-1234" not in (body["reason"] or "")
|
||||
|
||||
|
||||
# ── API 4: GET /scraper/pacing ───────────────────────────────────────────────
|
||||
|
|
|
|||
|
|
@ -7,6 +7,9 @@
|
|||
- телефон без цифр / слишком мало цифр -> 422 (digit-guard, #2376 hardening)
|
||||
- source="landing" (мёртвая воронка) -> 422 (литерал убран из схемы)
|
||||
- estimate_id, которого нет в trade_in_estimates -> 404
|
||||
- IDOR guard (security-audit, зеркалит #690/test_estimate_idor.py): estimate_id
|
||||
чужого пользователя -> 404; admin может привязать любой; нет
|
||||
X-Authenticated-User -> 401; неизвестная роль -> 403
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
|
@ -16,6 +19,7 @@ import os
|
|||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||||
|
||||
from datetime import UTC, datetime
|
||||
from types import SimpleNamespace
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock
|
||||
from uuid import uuid4
|
||||
|
|
@ -45,6 +49,16 @@ def client(db: MagicMock) -> TestClient:
|
|||
return TestClient(app)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _restore_get_role():
|
||||
"""Restore app.core.auth.get_role after each test (mirror test_estimate_idor.py)."""
|
||||
from app.core import auth as auth_mod
|
||||
|
||||
original = auth_mod.get_role
|
||||
yield
|
||||
auth_mod.get_role = original
|
||||
|
||||
|
||||
def _insert_result(lead_id: str) -> MagicMock:
|
||||
result = MagicMock()
|
||||
result.mappings.return_value.one.return_value = {
|
||||
|
|
@ -135,9 +149,14 @@ def test_lead_with_unknown_estimate_id_404(client: TestClient, db: MagicMock) ->
|
|||
|
||||
|
||||
def test_lead_with_known_estimate_id_200(client: TestClient, db: MagicMock) -> None:
|
||||
"""Owner привязывает лид к своей же оценке -> 200."""
|
||||
from app.core import auth as auth_mod
|
||||
|
||||
auth_mod.get_role = lambda _u: "pilot" # type: ignore[assignment]
|
||||
|
||||
lead_id = str(uuid4())
|
||||
exists_result = MagicMock()
|
||||
exists_result.fetchone.return_value = (1,)
|
||||
exists_result.fetchone.return_value = SimpleNamespace(created_by="kopylov")
|
||||
db.execute.side_effect = [exists_result, _insert_result(lead_id)]
|
||||
|
||||
estimate_id = str(uuid4())
|
||||
|
|
@ -148,6 +167,7 @@ def test_lead_with_known_estimate_id_200(client: TestClient, db: MagicMock) -> N
|
|||
"consent": True,
|
||||
"estimate_id": estimate_id,
|
||||
},
|
||||
headers={"X-Authenticated-User": "kopylov"},
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
params = db.execute.call_args.args[1]
|
||||
|
|
@ -155,6 +175,86 @@ def test_lead_with_known_estimate_id_200(client: TestClient, db: MagicMock) -> N
|
|||
assert params["source"] == "result"
|
||||
|
||||
|
||||
# ── IDOR guard (security-audit): estimate_id ownership ─────────────────────────
|
||||
|
||||
|
||||
def test_lead_estimate_id_owned_by_other_user_gets_404(client: TestClient, db: MagicMock) -> None:
|
||||
"""Чужой estimate_id -> 404 (существование не подтверждаем), лид НЕ создаётся."""
|
||||
from app.core import auth as auth_mod
|
||||
|
||||
auth_mod.get_role = lambda _u: "pilot" # type: ignore[assignment]
|
||||
|
||||
exists_result = MagicMock()
|
||||
exists_result.fetchone.return_value = SimpleNamespace(created_by="victim")
|
||||
db.execute.return_value = exists_result
|
||||
|
||||
r = client.post(
|
||||
"/api/v1/trade-in/lead",
|
||||
json={"phone": "+79123456789", "consent": True, "estimate_id": str(uuid4())},
|
||||
headers={"X-Authenticated-User": "attacker"},
|
||||
)
|
||||
assert r.status_code == 404, r.text
|
||||
assert not db.commit.called
|
||||
|
||||
|
||||
def test_lead_estimate_id_admin_can_attach_any_200(client: TestClient, db: MagicMock) -> None:
|
||||
"""Admin может привязать лид к чужой оценке (owner-or-admin, зеркалит #690)."""
|
||||
from app.core import auth as auth_mod
|
||||
|
||||
auth_mod.get_role = lambda _u: "admin" # type: ignore[assignment]
|
||||
|
||||
lead_id = str(uuid4())
|
||||
exists_result = MagicMock()
|
||||
exists_result.fetchone.return_value = SimpleNamespace(created_by="someone_else")
|
||||
db.execute.side_effect = [exists_result, _insert_result(lead_id)]
|
||||
|
||||
r = client.post(
|
||||
"/api/v1/trade-in/lead",
|
||||
json={"phone": "+79123456789", "consent": True, "estimate_id": str(uuid4())},
|
||||
headers={"X-Authenticated-User": "admin"},
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
|
||||
|
||||
def test_lead_estimate_id_requires_authenticated_user_401(
|
||||
client: TestClient, db: MagicMock
|
||||
) -> None:
|
||||
"""estimate_id задан, но нет X-Authenticated-User -> 401 (defense-in-depth: в
|
||||
проде rbac_guard уже требует заголовок раньше, см. app/main.py)."""
|
||||
exists_result = MagicMock()
|
||||
exists_result.fetchone.return_value = SimpleNamespace(created_by="kopylov")
|
||||
db.execute.return_value = exists_result
|
||||
|
||||
r = client.post(
|
||||
"/api/v1/trade-in/lead",
|
||||
json={"phone": "+79123456789", "consent": True, "estimate_id": str(uuid4())},
|
||||
)
|
||||
assert r.status_code == 401, r.text
|
||||
assert not db.commit.called
|
||||
|
||||
|
||||
def test_lead_estimate_id_unknown_role_403(client: TestClient, db: MagicMock) -> None:
|
||||
"""Аутентифицирован через Caddy, но роль отсутствует в roles.yaml -> 403."""
|
||||
from app.core import auth as auth_mod
|
||||
|
||||
def _raise_keyerror(_u: str):
|
||||
raise KeyError(_u)
|
||||
|
||||
auth_mod.get_role = _raise_keyerror # type: ignore[assignment]
|
||||
|
||||
exists_result = MagicMock()
|
||||
exists_result.fetchone.return_value = SimpleNamespace(created_by="kopylov")
|
||||
db.execute.return_value = exists_result
|
||||
|
||||
r = client.post(
|
||||
"/api/v1/trade-in/lead",
|
||||
json={"phone": "+79123456789", "consent": True, "estimate_id": str(uuid4())},
|
||||
headers={"X-Authenticated-User": "ghost"},
|
||||
)
|
||||
assert r.status_code == 403, r.text
|
||||
assert not db.commit.called
|
||||
|
||||
|
||||
def test_lead_digit_free_phone_422(client: TestClient, db: MagicMock) -> None:
|
||||
# "(()) -- .." проходит regex-маску (только +/скобки/дефисы/точки/пробелы),
|
||||
# но содержит 0 цифр -> должно отклоняться digit-валидатором.
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue