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

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 (его ставит

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)

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
 
 

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}

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 ───────────────────────────────────────────────

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-валидатором.