diff --git a/tradein-mvp/backend/app/api/v1/admin.py b/tradein-mvp/backend/app/api/v1/admin.py index 28d793ca..e7df8544 100644 --- a/tradein-mvp/backend/app/api/v1/admin.py +++ b/tradein-mvp/backend/app/api/v1/admin.py @@ -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 diff --git a/tradein-mvp/backend/app/api/v1/lead.py b/tradein-mvp/backend/app/api/v1/lead.py index 5ab93bf7..a5f1d106 100644 --- a/tradein-mvp/backend/app/api/v1/lead.py +++ b/tradein-mvp/backend/app/api/v1/lead.py @@ -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 (его ставит diff --git a/tradein-mvp/backend/app/core/request_audit.py b/tradein-mvp/backend/app/core/request_audit.py index a217cbfd..7eafefbc 100644 --- a/tradein-mvp/backend/app/core/request_audit.py +++ b/tradein-mvp/backend/app/core/request_audit.py @@ -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) diff --git a/tradein-mvp/backend/app/observability/sentry_scrub.py b/tradein-mvp/backend/app/observability/sentry_scrub.py index 197e86be..9d68d457 100644 --- a/tradein-mvp/backend/app/observability/sentry_scrub.py +++ b/tradein-mvp/backend/app/observability/sentry_scrub.py @@ -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 diff --git a/tradein-mvp/backend/tests/test_request_audit.py b/tradein-mvp/backend/tests/test_request_audit.py index 62bbf525..06bbd2d5 100644 --- a/tradein-mvp/backend/tests/test_request_audit.py +++ b/tradein-mvp/backend/tests/test_request_audit.py @@ -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} diff --git a/tradein-mvp/backend/tests/test_scraper_admin_apis.py b/tradein-mvp/backend/tests/test_scraper_admin_apis.py index 4d1a4390..ecfd4578 100644 --- a/tradein-mvp/backend/tests/test_scraper_admin_apis.py +++ b/tradein-mvp/backend/tests/test_scraper_admin_apis.py @@ -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 ─────────────────────────────────────────────── diff --git a/tradein-mvp/backend/tests/test_trade_in_lead.py b/tradein-mvp/backend/tests/test_trade_in_lead.py index 8f2f9859..8afd994f 100644 --- a/tradein-mvp/backend/tests/test_trade_in_lead.py +++ b/tradein-mvp/backend/tests/test_trade_in_lead.py @@ -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-валидатором.