feat(mera/b2c): анти-абуз для анонимного трафика — этап 2 из 8

Блокер номер один перед открытием эндпоинта оценки наружу: анонимный запрос
означал БЕЗЛИМИТ. В сервисе квот отсутствие имени пользователя трактовалось
как unlimited во всех функциях, с комментарием «dev без Caddy, fail-open».
Единственной защитой был общий лимит 300 запросов в минуту на IP — это
анти-флуд для дешёвых запросов, а не бизнес-лимит для пайплайна на десятки
секунд.

1. Fail-open больше не по умолчанию. Вызывающая сторона передаёт пустую
   личность только при явно включённом флаге разработки (по умолчанию выкл).

2. Анонимная личность — подписанная кука с HMAC-SHA256, отдельным каналом
   от X-Authenticated-User. Тот заголовок ставит Caddy и валидирует внутренним
   секретом; смешивать схемы нельзя, это сломало бы модель безопасности.
   Ключ подписи из окружения; если не задан — эфемерный на процесс, с
   предупреждением в лог.

3. Анонимная квота на паре «сессия + IP», переиспользует существующую таблицу
   и тот же атомарный инкремент под WHERE used < lim (защита от гонки #747).
   Честно закомментировано: смена IP или чистка куки обходит лимит — задача
   поднять стоимость злоупотребления, а не сделать его невозможным.

4. Отдельный жёсткий лимит частоты на оценку, проверяется ДО квоты.
   Переиспользован готовый SlidingWindowLimiter. Redis намеренно не задействован:
   прод работает одним воркером, состояние теряется только при рестарте, а
   основная защита — месячная квота в Postgres. Компромисс задокументирован.

5. Потолок времени ответа. Вызов Avito IMV шёл БЕЗ бюджета, в отличие от всех
   соседних — единственный источник неограниченного времени. Обёрнут.
   Суммарный худший случай: было ~186 с (36 с ограниченных плюс IMV без
   границы ~150 с), стало 56 с.

Тесты: 2754 passed. Два существующих теста обновлены под изменившееся
поведение fail-open — это ожидаемое изменение, не регрессия.

tradein-mvp/backend/app/api/v1/trade_in.py

@@ -15,9 +15,10 @@ from fastapi import APIRouter, Depends, File, Header, HTTPException, Request, Re
 from sqlalchemy import text
 from sqlalchemy.orm import Session
 
+from app.core.anon_session import get_or_create_anon_session_id
 from app.core.config import settings
 from app.core.db import get_db
-from app.core.ratelimit import _client_ip
+from app.core.ratelimit import SlidingWindowLimiter, _client_ip
 from app.schemas.trade_in import (
     AggregatedEstimate,
     AnalogLot,
@@ -52,6 +53,48 @@ logger = logging.getLogger(__name__)
 
 router = APIRouter()
 
+# ── B2C anti-abuse этап 2 (#b2c-antiabuse-2) ────────────────────────────────
+# Отдельный, куда более строгий лимит частоты specifically на POST /estimate —
+# см. settings.estimate_rate_limit/_window_s (app/core/config.py) для обоснования
+# значений. Тот же паттерн, что _send_limiter в app/api/v1/support.py: singleton
+# SlidingWindowLimiter поверх общего RateLimitMiddleware (app/main.py), который
+# уже применяется КО ВСЕМ /api/* путям, но с щедрым порогом, рассчитанным на
+# дешёвые запросы — один /estimate запускает цепочку внешних вызовов, суммарно
+# занимающую десятки секунд (см. estimator._with_budget budgets).
+_estimate_limiter = SlidingWindowLimiter(
+    limit=settings.estimate_rate_limit, window_s=settings.estimate_rate_limit_window_s
+)
+
+
+def _resolve_quota_identity(
+    request: Request,
+    response: Response,
+    x_authenticated_user: str | None,
+) -> tuple[str | None, int]:
+    """Резолвит (quota_key, default_limit) для account_quota.* — #b2c-antiabuse-2.
+
+    - X-Authenticated-User присутствует → (username, MONTHLY_LIMIT) — существующий
+      pilot/admin-флоу БЕЗ изменений (персональные override в
+      account_quota_overrides применяются как раньше через account_quota.user_limit).
+    - Заголовка нет И settings.quota_dev_fail_open=True (явный dev-флаг локальной
+      разработки без Caddy) → (None, MONTHLY_LIMIT) — account_quota трактует None
+      как unlimited. Флаг по умолчанию ВЫКЛЮЧЕН — это НЕ дефолтный прод-путь.
+    - Заголовка нет И флаг не задан (default, прод-путь для анонимов — продукт
+      открывается наружу) → анонимный ключ на основе подписанной session-cookie
+      (app.core.anon_session) + client IP, default_limit =
+      settings.anon_estimate_quota_limit (гораздо строже пилот-лимита). Честно:
+      смена IP или чистка cookie обходит этот лимит — цель поднять стоимость
+      злоупотребления, а не сделать его невозможным (тот же принцип, что и во
+      всей account_quota-схеме, #747).
+    """
+    if x_authenticated_user:
+        return x_authenticated_user, account_quota.MONTHLY_LIMIT
+    if settings.quota_dev_fail_open:
+        return None, account_quota.MONTHLY_LIMIT
+    session_id = get_or_create_anon_session_id(request, response)
+    anon_key = f"anon:{session_id}:{_client_ip(request)}"
+    return anon_key, settings.anon_estimate_quota_limit
+
 
 def _assert_estimate_access(created_by: str | None, x_authenticated_user: str | None) -> None:
     """IDOR guard (#690): только владелец оценки или admin могут её читать.
@@ -150,6 +193,7 @@ def _resolve_target_house_id(
 async def estimate(
     payload: TradeInEstimateInput,
     request: Request,
+    response: Response,
     db: Annotated[Session, Depends(get_db)],
     x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
 ) -> AggregatedEstimate:
@@ -160,9 +204,30 @@ async def estimate(
     3. Tukey IQR outlier filter
     4. Median + Q1 + Q3 + confidence с explanation
 
-    Применяется лимит 15 успешных оценок в месяц на аккаунт (кроме admin/kopylov).
+    Применяется лимит 15 успешных оценок в месяц на аккаунт (кроме admin/kopylov);
+    анонимные запросы (без X-Authenticated-User) — свой, гораздо более строгий
+    лимит на anon-сессию+IP (#b2c-antiabuse-2).
     """
-    account_quota.check_and_raise(db, x_authenticated_user)
+    # #b2c-antiabuse-2 п.4: отдельный жёсткий лимит частоты на дорогой публичный
+    # путь — самая дешёвая проверка первой, до квоты и до дорогой цепочки внешних
+    # вызовов. Ключ user:/ip: — тот же принцип, что общий RateLimitMiddleware
+    # (app/main.py); НЕ anon-сессия квоты (rate limit — про network-identity и
+    # burst-защиту capacity сервера, а не про месячный business-лимит).
+    _rl_key = (
+        f"user:{x_authenticated_user}" if x_authenticated_user else f"ip:{_client_ip(request)}"
+    )
+    _retry_after = _estimate_limiter.check(_rl_key)
+    if _retry_after is not None:
+        raise HTTPException(
+            status_code=429,
+            detail="Слишком много запросов на оценку. Попробуйте через несколько минут.",
+            headers={"Retry-After": str(int(_retry_after) + 1)},
+        )
+
+    quota_key, quota_default_limit = _resolve_quota_identity(
+        request, response, x_authenticated_user
+    )
+    account_quota.check_and_raise(db, quota_key, default_limit=quota_default_limit)
     from app.services.estimator import estimate_quality
 
     # #654: ранее любое исключение estimate_quality всплывало необработанным и
@@ -170,7 +235,9 @@ async def estimate(
     # через logger.exception (→ GlitchTip/Sentry получает stack trace) и отдаём
     # явный 503 — так любая БУДУЩАЯ реальная ошибка становится видимой, а не
     # «глотается» шлюзом. HTTPException пробрасываем как есть (это не сбой).
-    # created_by (#656) прокидываем в estimate_quality для скоупа /history.
+    # created_by (#656) прокидываем в estimate_quality для скоупа /history — ТОЛЬКО
+    # реальный account username (НЕ anon-ключ квоты): у анонимов нет "аккаунта",
+    # по которому имеет смысл скоупить /history.
     try:
         result = await estimate_quality(payload, db, created_by=x_authenticated_user)
     except HTTPException:
@@ -186,8 +253,19 @@ async def estimate(
     # тут: при гонке двух /estimate на used=lim-1 второй получит False.
     # Не списываем квоту за пустой результат (нерезолвящийся адрес и т.п.) — иначе
     # платный слот сгорает за insufficient_data=True (median=0, n_analogs=0) с HTTP 200.
-    if not result.insufficient_data and not account_quota.increment(db, x_authenticated_user):
-        raise HTTPException(status_code=429, detail=account_quota.LIMIT_EXHAUSTED_MESSAGE)
+    if not result.insufficient_data and not account_quota.increment(
+        db, quota_key, default_limit=quota_default_limit
+    ):
+        # Аутентифицированный путь — байт-в-байт прежнее сообщение (может не
+        # отражать персональный override, это pre-existing поведение, вне
+        # scope этого фикса). Анонимный путь — динамический текст с ПРАВИЛЬНЫМ
+        # anon-лимитом (#b2c-antiabuse-2), а не захардкоженным MONTHLY_LIMIT.
+        detail = (
+            account_quota.LIMIT_EXHAUSTED_MESSAGE
+            if x_authenticated_user
+            else account_quota.limit_exhausted_message(quota_default_limit)
+        )
+        raise HTTPException(status_code=429, detail=detail)
 
     # Feature 2/3 foundation: "что искали" — обогащённая estimate_request-запись
     # в user_events (адрес/площадь/комнаты + estimate_id для join с trade_in_estimates).
@@ -211,15 +289,22 @@ async def estimate(
 
 @router.get("/quota", response_model=QuotaStatus)
 def get_quota(
+    request: Request,
+    response: Response,
     db: Annotated[Session, Depends(get_db)],
     x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
 ) -> QuotaStatus:
-    """Статус квоты оценок для текущего аккаунта.
+    """Статус квоты оценок для текущего аккаунта (или анонимной сессии).
 
     Возвращает limit / used / remaining / unlimited для X-Authenticated-User.
-    Без заголовка (dev-режим без Caddy) — unlimited True, used 0.
+    Без заголовка (прод, публичный путь) — статус СВОЕЙ анонимной квоты
+    (anon-session cookie + IP), а НЕ безлимитный, если явно не включён
+    settings.quota_dev_fail_open (#b2c-antiabuse-2, dev без Caddy).
     """
-    status = account_quota.get_status(db, x_authenticated_user)
+    quota_key, quota_default_limit = _resolve_quota_identity(
+        request, response, x_authenticated_user
+    )
+    status = account_quota.get_status(db, quota_key, default_limit=quota_default_limit)
     return QuotaStatus(**status)
 
 

tradein-mvp/backend/app/core/anon_session.py

@@ -0,0 +1,111 @@
+"""Подписанный анонимный session-cookie — B2C anti-abuse этап 2 (#b2c-antiabuse-2).
+
+Продукт открывается для анонимных пользователей (без Caddy basic_auth /
+X-Authenticated-User). Чтобы применить анонимную квоту оценок (см.
+app.services.account_quota), нужен стабильный, но НЕ подделываемый идентификатор
+анонимной сессии — отдельный от X-Authenticated-User (этот заголовок ставит
+Caddy и валидируется внутренним секретом; смешивать схемы идентичности нельзя,
+это сломало бы модель безопасности #2213).
+
+Дизайн: session_id — случайный токен (secrets.token_urlsafe), подписанный
+HMAC-SHA256 вместе с expiry в cookie-значении `mera_anon_sid`. Сервер верифицирует
+подпись на каждом запросе; невалидная/просроченная/отсутствующая cookie → минтится
+новая сессия (клиент просто теряет накопленную анонимную квоту — это ОЖИДАЕМО и
+безопасно: анонимная квота и так обходится чисткой cookie/сменой IP, задача
+поднять стоимость злоупотребления, а не сделать его невозможным).
+
+Подпись предотвращает две вещи: (1) клиент не может подставить ЧУЖОЙ session_id
+(например, скопированный у другого пользователя) без знания секрета — иначе он
+мог бы попытаться "унаследовать" чужую квоту или испортить чужой учёт; (2) любая
+порча/усечение cookie-значения детектируется явно (HMAC mismatch), а не тихо
+парсится как валиден мусорный session_id.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+import logging
+import secrets
+import time
+
+from fastapi import Request, Response
+
+from app.core.config import settings
+
+logger = logging.getLogger(__name__)
+
+ANON_COOKIE_NAME = "mera_anon_sid"
+# 180 дней — грубо совпадает с "долгоживущий браузер, не чистящий cookies";
+# не критично для безопасности (лимит всё равно per-период_month), только для
+# TTL самой cookie в браузере/HMAC-подписи.
+ANON_SESSION_TTL_S = 180 * 24 * 3600
+
+# Пусто (default) → эфемерный per-process секрет: криптографически стойкий
+# (клиент не знает его и не может подделать сессию), но сессии не переживают
+# рестарт процесса и не общие между воркерами (см. anon_session_secret в
+# core/config.py). Вычисляется один раз при импорте модуля.
+if settings.anon_session_secret:
+    _SESSION_SECRET: bytes = settings.anon_session_secret.encode("utf-8")
+else:
+    _SESSION_SECRET = secrets.token_bytes(32)
+    logger.warning(
+        "ANON_SESSION_SECRET не задан — используется process-local эфемерный "
+        "секрет анонимной сессии (сессии сбрасываются при рестарте / не общие "
+        "между воркерами). Задай ANON_SESSION_SECRET в .env.runtime для "
+        "стабильности между рестартами."
+    )
+
+
+def _sign(session_id: str, expires_at: int) -> str:
+    msg = f"{session_id}.{expires_at}".encode()
+    return hmac.new(_SESSION_SECRET, msg, hashlib.sha256).hexdigest()
+
+
+def _encode(session_id: str, expires_at: int) -> str:
+    return f"{session_id}.{expires_at}.{_sign(session_id, expires_at)}"
+
+
+def _decode_and_verify(cookie_value: str) -> str | None:
+    """Возвращает session_id если cookie валидна (подпись + не просрочена), иначе None."""
+    parts = cookie_value.split(".", 2)
+    if len(parts) != 3:
+        return None
+    session_id, expires_at_raw, sig = parts
+    try:
+        expires_at = int(expires_at_raw)
+    except ValueError:
+        return None
+    if time.time() > expires_at:
+        return None
+    expected = _sign(session_id, expires_at)
+    if not hmac.compare_digest(sig, expected):
+        return None
+    return session_id
+
+
+def get_or_create_anon_session_id(request: Request, response: Response) -> str:
+    """Возвращает стабильный анонимный session_id, выставляя cookie при необходимости.
+
+    Читает `mera_anon_sid` из запроса; если отсутствует, повреждена, просрочена
+    или не проходит проверку подписи — минтит НОВУЮ сессию и выставляет свежую
+    cookie на *response* (httponly, samesite=lax; secure вне dev-окружения).
+    """
+    raw = request.cookies.get(ANON_COOKIE_NAME)
+    if raw:
+        verified = _decode_and_verify(raw)
+        if verified is not None:
+            return verified
+        logger.info("anon_session: invalid/tampered/expired cookie — minting new session")
+
+    session_id = secrets.token_urlsafe(16)
+    expires_at = int(time.time()) + ANON_SESSION_TTL_S
+    response.set_cookie(
+        ANON_COOKIE_NAME,
+        _encode(session_id, expires_at),
+        max_age=ANON_SESSION_TTL_S,
+        httponly=True,
+        samesite="lax",
+        secure=settings.environment != "dev",
+    )
+    return session_id

tradein-mvp/backend/app/core/config.py

@@ -86,6 +86,22 @@ class Settings(BaseSettings):
         default=5, validation_alias="RATE_LIMIT_AUTHENTICATED_MULTIPLIER"
     )
 
+    # ── B2C anti-abuse этап 2: отдельный жёсткий лимит частоты на POST /estimate ──
+    # Общий rate_limit (300/60с) рассчитан на дешёвые запросы; один вызов /estimate
+    # запускает цепочку внешних вызовов (geocode → Overpass → IMV → Yandex → Cian),
+    # каждый забюджетирован, но суммарно может занимать десятки секунд. Отдельный,
+    # куда более строгий бюджет burst'а поверх общего — не даёт одному ключу
+    # (user:/ip:, тот же принцип что и general-лимит) запустить много параллельных
+    # дорогих цепочек за короткое окно. НЕ заменяет account_quota (месячная квота,
+    # персистентная в Postgres) — это защита от burst, а не от abuse за месяц.
+    # Применяется К ЛЮБОМУ ключу (auth и анон одинаково) — цель защитить capacity
+    # сервера/upstream-скрейперов, а не различать роли. ENV: ESTIMATE_RATE_LIMIT,
+    # ESTIMATE_RATE_LIMIT_WINDOW_S.
+    estimate_rate_limit: int = Field(default=5, validation_alias="ESTIMATE_RATE_LIMIT")
+    estimate_rate_limit_window_s: float = Field(
+        default=300.0, validation_alias="ESTIMATE_RATE_LIMIT_WINDOW_S"
+    )
+
     # Password for tradein_fdw_reader role — used by backend startup to create/refresh
     # USER MAPPING for postgres_fdw → gendesign DB (gendesign_remote server).
     # Пусто = USER MAPPING не создаётся, gendesign_cad_buildings не работает (dev).
@@ -437,11 +453,56 @@ class Settings(BaseSettings):
     estimate_cian_valuation_timeout_s: float = 8.0
     estimate_geocode_budget_s: float = 12.0
     estimate_house_meta_timeout_s: float = 8.0
+    # #b2c-antiabuse-2: Avito IMV (evaluate_via_imv) была ЕДИНСТВЕННЫМ внешним
+    # вызовом в /estimate БЕЗ _with_budget — до 3 последовательных HTTP-запросов
+    # (warm-up + geocode + evaluate), каждый со своим таймаутом 25s
+    # (_HTTP_TIMEOUT_SEC в scraper_kit.providers.avito.imv), плюс возможен ОДИН
+    # internal retry с "очищенным" адресом на IMVAddressNotFoundError — необёрнутый
+    # worst-case доходил до ~150s. 20s щедрее соседних бюджетов (8s Yandex/Cian/
+    # house_meta) намеренно — IMV делает МНОГО последовательных round-trip'ов, а
+    # не один запрос, поэтому реалистичный "медленный, но живой" ответ длиннее.
+    # ENV: ESTIMATE_AVITO_IMV_TIMEOUT_S.
+    estimate_avito_imv_timeout_s: float = Field(
+        default=20.0, validation_alias="ESTIMATE_AVITO_IMV_TIMEOUT_S"
+    )
 
     # Лимит успешных оценок trade-in за календарный месяц на аккаунт (#658).
     # Конфигурируется через env ESTIMATE_QUOTA_LIMIT. Default 15.
     estimate_quota_limit: int = 15
 
+    # ── B2C anti-abuse этап 2 (#b2c-antiabuse-2) ──────────────────────────────
+    # Продукт открывается для анонимных пользователей — анонимный запрос БЕЗ
+    # X-Authenticated-User (Caddy basic_auth) раньше трактовался как unlimited
+    # безусловно (dev без Caddy). Недопустимо для публичного пути: любой
+    # анонимный клиент получал бы безлимитные дорогие оценки.
+    #
+    # quota_dev_fail_open: явный флаг ТОЛЬКО для локальной разработки без Caddy.
+    # По умолчанию ВЫКЛЮЧЕН — анонимный запрос без заголовка получает анонимную
+    # квоту (anon-session cookie + IP), а не безлимит. True включает старое
+    # fail-open поведение (username=None → unlimited) — задавай только в dev.
+    # ENV: QUOTA_DEV_FAIL_OPEN.
+    quota_dev_fail_open: bool = Field(default=False, validation_alias="QUOTA_DEV_FAIL_OPEN")
+
+    # anon_estimate_quota_limit: месячный лимит успешных оценок на связку
+    # (anon-session-cookie + client IP) для запросов БЕЗ X-Authenticated-User.
+    # Существенно строже пилот-лимита (estimate_quota_limit=15) — анонимный
+    # трафик не аутентифицирован и открыт всему интернету. Честно: смена IP или
+    # чистка cookie обходит этот лимит — цель поднять стоимость злоупотребления,
+    # а не сделать его невозможным (тот же принцип, что и account_estimate_usage
+    # для пилотов). ENV: ANON_ESTIMATE_QUOTA_LIMIT.
+    anon_estimate_quota_limit: int = Field(default=3, validation_alias="ANON_ESTIMATE_QUOTA_LIMIT")
+
+    # anon_session_secret: ключ HMAC-подписи анонимного session-cookie (см.
+    # app/core/anon_session.py). Пусто (дефолт) → используется process-local
+    # эфемерный секрет, сгенерированный при старте (secrets.token_bytes) —
+    # криптографически стойкий (клиент его не знает и не может подделать
+    # session_id), но НЕ переживает рестарт/не общий между несколькими
+    # воркерами (каждый рестарт — новая генерация → старые cookie невалидны,
+    # клиенты просто получают новую anon-сессию, деградация допустима). Для
+    # стабильности между рестартами (текущий деплой — single-worker uvicorn)
+    # задай явный секрет в .env.runtime. ENV: ANON_SESSION_SECRET.
+    anon_session_secret: str = Field(default="", validation_alias="ANON_SESSION_SECRET")
+
     # Фильтр junk-/премиум-порога для asking→sold derivation (#767).
     # Нижняя граница 30 000 ₽/м² отсекает нежилые/технические сделки; менять не стоит.
     # Верхняя граница — поднята с 600 000 до 1 200 000 ₽/м², чтобы покрыть ЕКБ-premium

tradein-mvp/backend/app/services/account_quota.py

@@ -1,12 +1,19 @@
-"""Сервис квоты оценок trade-in — N успешных оценок в месяц на аккаунт.
+"""Сервис квоты оценок trade-in — N успешных оценок в месяц на ключ (аккаунт ИЛИ
+анонимная сессия+IP, см. #b2c-antiabuse-2).
 
 Правила:
 - Лимит по умолчанию = settings.estimate_quota_limit успешных оценок за календарный
   месяц (UTC, период 'YYYY-MM'); конфигурируется через env ESTIMATE_QUOTA_LIMIT,
-  default 15.
+  default 15. Это дефолт для АУТЕНТИФИЦИРОВАННЫХ (X-Authenticated-User) ключей.
+  Анонимные ключи (см. app.api.v1.trade_in._resolve_quota_identity) используют
+  СВОЙ, гораздо более строгий default через параметр `default_limit=` —
+  все функции ниже принимают username-подобный `key: str | None` без разбора,
+  реальный аккаунт это или составной anon-ключ ("anon:<session>:<ip>").
 - Персональный override: таблица account_quota_overrides (username → monthly_limit),
   см. миграцию 185_account_quota_overrides.sql. Заменяет прежний хак бонусных попыток
-  через negative `used` (ломал /quota — «Осталось 50 из 15»).
+  через negative `used` (ломал /quota — «Осталось 50 из 15»). Работает одинаково
+  для anon-ключей (в норме нет override-строки → falls back на переданный
+  `default_limit`), так и для обычных username.
 - `used` в account_estimate_usage защищён CHECK (used >= 0) на уровне схемы, см.
   миграцию 189_account_estimate_usage_nonnegative.sql — 185 сбросила негативный
   used только для user2, 189 закрывает остальные аккаунты + запрещает регресс.
@@ -17,9 +24,16 @@
   До миграции 191 unlimited для non-admin аккаунтов был захардкожен как
   `username == 'kopylov'` прямо в коде — данные (kopylov + praktika) заменяют этот
   хардкод целиком, единый источник правды для всех безлимитных non-admin грантов.
+  Анонимные ключи никогда не unlimited (get_role() кидает KeyError на составной
+  anon-ключ → is_unlimited() шорткатится в False БЕЗ похода в БД).
 - Учитываются ТОЛЬКО успешные оценки (инкремент ПОСЛЕ estimate_quality).
-- Если заголовок X-Authenticated-User отсутствует (dev без Caddy) → unlimited,
-  лимит не применяется (fail-open).
+- key is None → unlimited, лимит не применяется (fail-open). #b2c-antiabuse-2:
+  ЭТОТ модуль как был, так и остаётся fail-open на None — но с этапа anti-abuse
+  вызывающая сторона (app.api.v1.trade_in._resolve_quota_identity) передаёт None
+  ТОЛЬКО за явным флагом settings.quota_dev_fail_open (по умолчанию ВЫКЛЮЧЕН).
+  Анонимный запрос без этого флага получает anon-ключ (см. app.core.anon_session),
+  а не None — то есть на практике анонимные пользователи в проде квоту получают,
+  а не безлимит.
 - При исчерпании лимита поднимается HTTPException(429).
 """
 
@@ -40,12 +54,23 @@ logger = logging.getLogger(__name__)
 # Лимит успешных оценок за календарный месяц — конфигурируется через
 # env ESTIMATE_QUOTA_LIMIT (core.config.Settings), default 15 (#658).
 MONTHLY_LIMIT = settings.estimate_quota_limit
-LIMIT_EXHAUSTED_MESSAGE = (
-    f"Лимит из {MONTHLY_LIMIT} оценок в этом месяце исчерпан. "
+
+
+def limit_exhausted_message(limit: int) -> str:
+    """Текст 429 при исчерпании лимита — параметризован реальным лимитом (может
+    отличаться от глобального MONTHLY_LIMIT для персонального override ИЛИ
+    anon default_limit, см. #b2c-antiabuse-2)."""
+    return (
+        f"Лимит из {limit} оценок в этом месяце исчерпан. "
         "За полной версией обращайтесь к Копылову."
     )
 
 
+# Backward-compat константа для MONTHLY_LIMIT-based сценариев (тесты, existing
+# imports) — байт-в-байт совпадает с limit_exhausted_message(MONTHLY_LIMIT).
+LIMIT_EXHAUSTED_MESSAGE = limit_exhausted_message(MONTHLY_LIMIT)
+
+
 def current_period() -> str:
     """Возвращает текущий период в формате 'YYYY-MM' (UTC)."""
     return datetime.now(UTC).strftime("%Y-%m")
@@ -84,13 +109,18 @@ def is_unlimited(db: Session, username: str) -> bool:
     return bool(row is not None and row.unlimited)
 
 
-def user_limit(db: Session, username: str) -> int:
-    """Персональный месячный лимит для username, иначе глобальный MONTHLY_LIMIT.
+def user_limit(db: Session, username: str, *, default: int = MONTHLY_LIMIT) -> int:
+    """Персональный месячный лимит для username, иначе *default*.
 
     Источник override — таблица account_quota_overrides (см. миграцию
     185_account_quota_overrides.sql). Заменяет прежний хак бонусных попыток через
     negative `used`, который ломал /quota (limit=15, used=-35 → remaining=50 —
     «Осталось 50 из 15»).
+
+    *default* параметризован (не всегда MONTHLY_LIMIT) ради anon-ключей
+    (#b2c-antiabuse-2): анонимный ("anon:<session>:<ip>") ключ в норме не имеет
+    override-строки → падает на *default*, который вызывающая сторона задаёт
+    равным settings.anon_estimate_quota_limit (гораздо строже пилот-лимита).
     """
     row = db.execute(
         text(
@@ -103,27 +133,29 @@ def user_limit(db: Session, username: str) -> int:
     ).fetchone()
     if row is not None and row.monthly_limit is not None:
         return int(row.monthly_limit)
-    return MONTHLY_LIMIT
+    return default
 
 
-def get_status(db: Session, username: str | None) -> dict:
-    """Возвращает статус квоты для пользователя.
+def get_status(db: Session, username: str | None, *, default_limit: int = MONTHLY_LIMIT) -> dict:
+    """Возвращает статус квоты для пользователя (или anon-ключа, #b2c-antiabuse-2).
 
-    Если username is None → unlimited True, used 0, remaining = MONTHLY_LIMIT.
+    Если username is None → unlimited True, used 0, remaining = default_limit
+    (fail-open — вызывающая сторона передаёт None ТОЛЬКО за явным dev-флагом,
+    см. app.api.v1.trade_in._resolve_quota_identity).
     Если unlimited → used = фактический или 0, remaining = limit (per-user override
-    или глобальный MONTHLY_LIMIT).
+    или *default_limit*).
     """
     if username is None:
         return {
-            "limit": MONTHLY_LIMIT,
+            "limit": default_limit,
             "used": 0,
-            "remaining": MONTHLY_LIMIT,
+            "remaining": default_limit,
             "unlimited": True,
         }
 
     unlimited = is_unlimited(db, username)
     period = current_period()
-    limit = user_limit(db, username)
+    limit = user_limit(db, username, default=default_limit)
 
     row = db.execute(
         text(
@@ -157,10 +189,14 @@ def get_status(db: Session, username: str | None) -> dict:
     }
 
 
-def check_and_raise(db: Session, username: str | None) -> None:
+def check_and_raise(
+    db: Session, username: str | None, *, default_limit: int = MONTHLY_LIMIT
+) -> None:
     """Проверяет лимит квоты и поднимает 429 если исчерпан.
 
-    Если username is None или пользователь unlimited → no-op.
+    Если username is None или пользователь unlimited → no-op. *default_limit*
+    задаёт лимит для ключей без персонального override (пилот → MONTHLY_LIMIT,
+    anon-ключ → settings.anon_estimate_quota_limit, см. #b2c-antiabuse-2).
     """
     if username is None:
         return
@@ -169,7 +205,7 @@ def check_and_raise(db: Session, username: str | None) -> None:
         return
 
     period = current_period()
-    limit = user_limit(db, username)
+    limit = user_limit(db, username, default=default_limit)
     row = db.execute(
         text(
             """
@@ -189,14 +225,14 @@ def check_and_raise(db: Session, username: str | None) -> None:
             used,
             limit,
         )
-        raise HTTPException(status_code=429, detail=LIMIT_EXHAUSTED_MESSAGE)
+        raise HTTPException(status_code=429, detail=limit_exhausted_message(limit))
 
 
-def increment(db: Session, username: str | None) -> bool:
+def increment(db: Session, username: str | None, *, default_limit: int = MONTHLY_LIMIT) -> bool:
     """Атомарно-условный инкремент счётчика успешных оценок (#747).
 
     Возвращает True если инкремент успешен; False если лимит исчерпан.
-    None / unlimited → True (no-op success).
+    None / unlimited → True (no-op success). *default_limit* — см. check_and_raise.
 
     Защита от TOCTOU: предикат `WHERE used < :lim` применяется к ветке DO UPDATE —
     два параллельных запроса при used=lim-1 не могут оба инкрементировать (второй
@@ -208,7 +244,7 @@ def increment(db: Session, username: str | None) -> bool:
         return True
 
     period = current_period()
-    lim = user_limit(db, username)
+    lim = user_limit(db, username, default=default_limit)
     row = db.execute(
         text(
             """

tradein-mvp/backend/app/services/estimator.py

@@ -3418,7 +3418,16 @@ async def estimate_quality(
         and imv_house_type is not None
         and imv_renovation is not None
     ):
-        imv_eval = await _get_or_fetch_imv_cached(
+        # #654/#b2c-antiabuse-2: единственный внешний вызов, ранее БЕЗ time-budget
+        # guard — evaluate_via_imv делает до 3 последовательных HTTP-запросов
+        # (warm-up + geocode + evaluate), каждый со своим таймаутом 25s
+        # (_HTTP_TIMEOUT_SEC в scraper_kit.providers.avito.imv), плюс возможен ОДИН
+        # internal retry с "очищенным" адресом на IMVAddressNotFoundError (см.
+        # _get_or_fetch_imv_cached) — необёрнутый worst-case доходил до ~150s.
+        # Оборачиваем так же, как соседние ungated-вызовы (geocode/house_meta/
+        # yandex_valuation/cian_valuation).
+        imv_eval = await _with_budget(
+            _get_or_fetch_imv_cached(
                 db,
                 address=geo.full_address,
                 rooms=payload.rooms,
@@ -3429,6 +3438,9 @@ async def estimate_quality(
                 renovation_type=imv_renovation,
                 has_balcony=bool(payload.has_balcony),
                 has_loggia=False,
+            ),
+            settings.estimate_avito_imv_timeout_s,
+            label="avito_imv",
         )
 
     # ── Stage 8: Yandex Valuation as on-demand source (anonymous, cached 24h) ──

tradein-mvp/backend/tests/conftest.py

@@ -1,13 +1,16 @@
 """Repo-wide test config for tradein-mvp/backend.
 
-Currently only registers custom pytest markers so they don't emit
-PytestUnknownMarkWarning when used (`--strict-markers` is not enabled in
-pyproject.toml, so an unregistered marker would only warn, not fail — this
-just keeps output clean and documents intent in one place).
+Registers custom pytest markers so they don't emit PytestUnknownMarkWarning
+when used (`--strict-markers` is not enabled in pyproject.toml, so an
+unregistered marker would only warn, not fail — this just keeps output clean
+and documents intent in one place). Also resets cross-test-file shared
+rate-limiter state (see fixture docstring below).
 """
 
 from __future__ import annotations
 
+import pytest
+
 
 def pytest_configure(config) -> None:
     config.addinivalue_line(
@@ -16,3 +19,26 @@ def pytest_configure(config) -> None:
         "Pango/cairo/GObject libs, self-skips where unavailable (see "
         "tests/test_pdf_real_render.py docstring for how to run it for real).",
     )
+
+
+@pytest.fixture(autouse=True)
+def _reset_estimate_rate_limiter() -> None:
+    """`app.api.v1.trade_in._estimate_limiter` (#b2c-antiabuse-2) is a module-level
+    `SlidingWindowLimiter` singleton that accumulates hits across ALL tests
+    hitting POST /estimate within one pytest process (a dozen+ test files build
+    their own FastAPI app around `trade_in_module.router` — see grep for
+    `trade_in_module.router` under tests/). Without a reset, unrelated test
+    files could trip the 429 rate-limit purely from cross-test state leakage
+    (same class of issue `test_support.py::_fresh_rate_limiter` solves locally
+    for `_send_limiter` — this one needs to be global since so many files touch
+    the trade_in router). Lazy import: keeps conftest.py import-light and avoids
+    forcing DATABASE_URL to be set before any test module has had a chance to
+    default it.
+    """
+    from app.api.v1 import trade_in as trade_in_module
+    from app.core.config import settings
+    from app.core.ratelimit import SlidingWindowLimiter
+
+    trade_in_module._estimate_limiter = SlidingWindowLimiter(
+        limit=settings.estimate_rate_limit, window_s=settings.estimate_rate_limit_window_s
+    )

tradein-mvp/backend/tests/test_account_quota.py

@@ -5,7 +5,9 @@ Coverage:
   (b) обычный pilot-юзер блокируется на 16-м запросе (429 + нужный detail)
   (c) increment растит used счётчик
   (d) get_status корректен для different сценариев
-  (e) отсутствие заголовка X-Authenticated-User = unlimited (fail-open)
+  (e) отсутствие заголовка X-Authenticated-User → анонимная квота, НЕ unlimited
+      (#b2c-antiabuse-2 — fail-open убран из дефолтного пути; opt-in dev-флаг
+      settings.quota_dev_fail_open восстанавливает старое unlimited-поведение)
   (f) #747 — атомарно-условный increment (TOCTOU fix)
   (g) account_quota_overrides.monthly_limit — персональный лимит вместо negative-used
       хака
@@ -13,6 +15,8 @@ Coverage:
   (i) account_quota_overrides.unlimited — data-driven безлимит (migration 191):
       kopylov (перенесён из хардкода) и praktika (восстановленный пилот) безлимитны
       через таблицу, не через код
+  (j) #b2c-antiabuse-2: account_quota.*(default_limit=...) — anon-ключ falls back
+      на переданный default (НЕ MONTHLY_LIMIT) при отсутствии override-строки
 
 DB мокируется через _FakeDB (роутинг по SQL-тексту, см. ниже) — реальная БД не
 требуется.
@@ -403,7 +407,42 @@ def quota_app() -> FastAPI:
 
 
 def test_quota_endpoint_no_header(quota_app: FastAPI) -> None:
-    """GET /quota без заголовка → unlimited=True, remaining=15."""
+    """GET /quota без заголовка (анонимный, #b2c-antiabuse-2) → НЕ безлимитный:
+    анонимная квота (anon-session cookie + IP), unlimited=False, limit =
+    settings.anon_estimate_quota_limit. Fail-open больше НЕ дефолт — см.
+    test_quota_dev_fail_open_flag_restores_unlimited для явного opt-in."""
+    from app.core.config import settings
+
+    client = TestClient(quota_app)
+    resp = client.get("/api/v1/trade-in/quota")
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["unlimited"] is False
+    assert data["limit"] == settings.anon_estimate_quota_limit
+    assert data["remaining"] == settings.anon_estimate_quota_limit
+    assert data["used"] == 0
+    # Подписанная анонимная session-cookie выставлена — стабильная идентичность
+    # анонимного пользователя между запросами (app.core.anon_session).
+    assert "mera_anon_sid" in resp.cookies
+
+
+def test_quota_dev_fail_open_flag_default_false() -> None:
+    """settings.quota_dev_fail_open по умолчанию False — fail-open НЕ включён без
+    явного флага (#b2c-antiabuse-2)."""
+    from app.core.config import settings
+
+    assert settings.quota_dev_fail_open is False
+
+
+def test_quota_dev_fail_open_flag_restores_unlimited(
+    quota_app: FastAPI, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """GET /quota без заголовка С явным settings.quota_dev_fail_open=True →
+    восстанавливает старое unlimited-поведение (dev-режим без Caddy, opt-in)."""
+    from app.core import config as config_module
+
+    monkeypatch.setattr(config_module.settings, "quota_dev_fail_open", True)
+
     client = TestClient(quota_app)
     resp = client.get("/api/v1/trade-in/quota")
     assert resp.status_code == 200
@@ -525,8 +564,12 @@ def test_estimate_admin_not_blocked(estimate_app_exhausted: FastAPI) -> None:
     assert resp.status_code != 429
 
 
-def test_estimate_no_header_not_blocked(estimate_app_exhausted: FastAPI) -> None:
-    """POST /estimate без заголовка → не 429 (fail-open, dev-режим)."""
+def test_estimate_anon_blocked_when_db_reports_exhausted(
+    estimate_app_exhausted: FastAPI,
+) -> None:
+    """POST /estimate без заголовка (анонимный) при used=MONTHLY_LIMIT (>> анонимного
+    лимита) → 429. #b2c-antiabuse-2: fail-open по умолчанию убран — аноним получает
+    квоту, а НЕ безлимит."""
     client = TestClient(estimate_app_exhausted, raise_server_exceptions=False)
     resp = client.post(
         "/api/v1/trade-in/estimate",
@@ -536,9 +579,66 @@ def test_estimate_no_header_not_blocked(estimate_app_exhausted: FastAPI) -> None
             "rooms": 2,
         },
     )
+    assert resp.status_code == 429
+
+
+def test_estimate_anon_not_blocked_under_quota(quota_app: FastAPI) -> None:
+    """POST /estimate без заголовка, used=0 (свежий anon-ключ, под лимитом) → НЕ
+    блокируется квотой (может упасть на другой ошибке — нет реального estimator/
+    geocoder; здесь важно только что это не 429 от квоты)."""
+    client = TestClient(quota_app, raise_server_exceptions=False)
+    resp = client.post(
+        "/api/v1/trade-in/estimate",
+        json={
+            "address": "г. Екатеринбург, ул. Малышева, 1",
+            "area_m2": 50.0,
+            "rooms": 2,
+        },
+    )
     assert resp.status_code != 429
 
 
+@pytest.fixture()
+def estimate_app_anon_exhausted() -> FastAPI:
+    """FastAPI app где БД возвращает used == anon_estimate_quota_limit для ЛЮБОГО
+    ключа — аноним исчерпал СВОЙ, гораздо более строгий лимит."""
+    from app.api.v1 import trade_in as trade_in_module
+    from app.core.config import settings
+    from app.core.db import get_db
+
+    application = FastAPI()
+    application.include_router(trade_in_module.router, prefix="/api/v1/trade-in")
+
+    def _override_db():
+        yield _FakeDB(used=settings.anon_estimate_quota_limit)
+
+    application.dependency_overrides[get_db] = _override_db
+    return application
+
+
+def test_estimate_anon_blocked_at_anon_limit_with_correct_message(
+    estimate_app_anon_exhausted: FastAPI,
+) -> None:
+    """POST /estimate анонимный, used == anon_estimate_quota_limit → 429 с
+    anon-специфичным detail (правильное, меньшее число — НЕ захардкоженный
+    MONTHLY_LIMIT=15 из LIMIT_EXHAUSTED_MESSAGE)."""
+    from app.core.config import settings
+
+    client = TestClient(estimate_app_anon_exhausted, raise_server_exceptions=False)
+    resp = client.post(
+        "/api/v1/trade-in/estimate",
+        json={
+            "address": "г. Екатеринбург, ул. Малышева, 1",
+            "area_m2": 50.0,
+            "rooms": 2,
+        },
+    )
+    assert resp.status_code == 429
+    detail = resp.json()["detail"]
+    assert str(settings.anon_estimate_quota_limit) in detail
+    assert str(MONTHLY_LIMIT) not in detail
+
+
 @pytest.fixture()
 def estimate_app_praktika_unlimited() -> FastAPI:
     """FastAPI app где БД отдаёт unlimited=true для praktika (used заведомо
@@ -884,3 +984,80 @@ def test_estimate_real_result_still_increments_quota(estimate_app_ok: FastAPI) -
     assert resp.status_code == 200
     assert resp.json()["insufficient_data"] is False
     mock_increment.assert_called_once()
+
+
+# ---------------------------------------------------------------------------
+# (j) #b2c-antiabuse-2: account_quota.*(default_limit=...) — anon-ключ falls
+# back на переданный default (НЕ MONTHLY_LIMIT) при отсутствии override-строки
+# ---------------------------------------------------------------------------
+
+
+_ANON_KEY = "anon:fake-session-id:203.0.113.7"
+_ANON_DEFAULT = 3  # существенно строже MONTHLY_LIMIT (15) — см. settings.anon_estimate_quota_limit
+
+
+def test_user_limit_anon_key_falls_back_to_custom_default() -> None:
+    """anon-ключ без override-строки → user_limit(default=X) возвращает X, НЕ
+    MONTHLY_LIMIT."""
+    db = MagicMock()
+    db.execute.return_value = _override_result(None)
+    assert user_limit(db, _ANON_KEY, default=_ANON_DEFAULT) == _ANON_DEFAULT
+    assert user_limit(db, _ANON_KEY, default=_ANON_DEFAULT) != MONTHLY_LIMIT
+
+
+def test_is_unlimited_anon_key_short_circuits_without_db() -> None:
+    """anon-ключ никогда не в roles.yaml → is_unlimited() шорткатится в False
+    БЕЗ похода в БД (get_role() KeyError раньше любого SELECT)."""
+    db = MagicMock()
+    assert is_unlimited(db, _ANON_KEY) is False
+    db.execute.assert_not_called()
+
+
+def test_check_and_raise_anon_default_limit_blocks_at_anon_threshold() -> None:
+    """used == default_limit (anon) → 429, хотя это существенно МЕНЬШЕ
+    MONTHLY_LIMIT — доказывает, что anon использует СВОЙ лимит, а не глобальный."""
+    from fastapi import HTTPException
+
+    db = _FakeDB(used=_ANON_DEFAULT)
+    with pytest.raises(HTTPException) as exc_info:
+        check_and_raise(db, _ANON_KEY, default_limit=_ANON_DEFAULT)
+    assert exc_info.value.status_code == 429
+    assert str(_ANON_DEFAULT) in exc_info.value.detail
+
+
+def test_check_and_raise_anon_default_limit_allows_below_anon_threshold() -> None:
+    """used < default_limit (anon) → не блокируется."""
+    db = _FakeDB(used=_ANON_DEFAULT - 1)
+    check_and_raise(db, _ANON_KEY, default_limit=_ANON_DEFAULT)  # не должно поднять
+
+
+def test_increment_anon_default_limit_respected() -> None:
+    """increment с anon default_limit: used=default-1 → True; used=default → False —
+    тот же #747 atomic-guard, применённый к anon-специфичному, а не глобальному лимиту."""
+    db = _AtomicQuotaFakeDB(used=_ANON_DEFAULT - 1)
+    assert increment(db, _ANON_KEY, default_limit=_ANON_DEFAULT) is True
+    assert db.used == _ANON_DEFAULT
+    assert increment(db, _ANON_KEY, default_limit=_ANON_DEFAULT) is False
+    assert db.used == _ANON_DEFAULT
+
+
+def test_get_status_anon_default_limit() -> None:
+    """get_status с anon default_limit: limit/remaining отражают anon-лимит, не
+    MONTHLY_LIMIT."""
+    db = _FakeDB(used=1)
+    status = get_status(db, _ANON_KEY, default_limit=_ANON_DEFAULT)
+    assert status["unlimited"] is False
+    assert status["limit"] == _ANON_DEFAULT
+    assert status["used"] == 1
+    assert status["remaining"] == _ANON_DEFAULT - 1
+
+
+def test_limit_exhausted_message_reflects_actual_limit() -> None:
+    """limit_exhausted_message(N) параметризован — не всегда MONTHLY_LIMIT."""
+    from app.services.account_quota import limit_exhausted_message
+
+    msg_anon = limit_exhausted_message(_ANON_DEFAULT)
+    assert str(_ANON_DEFAULT) in msg_anon
+    assert str(MONTHLY_LIMIT) not in msg_anon
+    # Backward-compat: LIMIT_EXHAUSTED_MESSAGE == limit_exhausted_message(MONTHLY_LIMIT).
+    assert limit_exhausted_message(MONTHLY_LIMIT) == LIMIT_EXHAUSTED_MESSAGE

tradein-mvp/backend/tests/test_estimate_rate_limit.py

@@ -0,0 +1,154 @@
+"""Tests for the dedicated POST /estimate rate limiter (#b2c-antiabuse-2 п.4).
+
+Отдельный, куда более строгий лимит частоты specifically на дорогой публичный
+путь POST /estimate (`app.api.v1.trade_in._estimate_limiter`,
+settings.estimate_rate_limit/_window_s) — поверх общего RateLimitMiddleware
+(300 req/60с, app/main.py), который рассчитан на дешёвые запросы. Один вызов
+/estimate запускает цепочку внешних вызовов, суммарно занимающую десятки секунд
+(см. estimator._with_budget), поэтому burst нескольких параллельных вызовов от
+одного ключа нужно резать раньше, гораздо строже.
+
+Изолировано от account_quota (mock'ается no-op) и от estimate_quality (canned
+result, без реальной сети/DB) — цель проверить ИМЕННО срабатывание узкого
+rate-limit гейта, который стоит ПЕРВЫМ в хендлере (до квоты и до дорогой цепочки).
+
+`tests/conftest.py::_reset_estimate_rate_limiter` сбрасывает `_estimate_limiter`
+перед каждым тестом (иначе состояние утекало бы между файлами) — здесь мы поверх
+этого сброса ещё и сужаем лимит через autouse-фикстуру, чтобы не тестировать
+прод-значения (5/300с) напрямую (медленно/шумно).
+"""
+
+from __future__ import annotations
+
+import os
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from datetime import UTC, datetime, timedelta
+from unittest.mock import AsyncMock, patch
+from uuid import uuid4
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from app.api.v1 import trade_in as trade_in_module
+from app.core.db import get_db
+from app.core.ratelimit import SlidingWindowLimiter
+from app.schemas.trade_in import AggregatedEstimate
+
+
+def _canned_estimate() -> AggregatedEstimate:
+    return AggregatedEstimate(
+        estimate_id=uuid4(),
+        median_price_rub=5_000_000,
+        range_low_rub=4_500_000,
+        range_high_rub=5_500_000,
+        median_price_per_m2=100_000,
+        confidence="medium",
+        n_analogs=8,
+        period_months=24,
+        analogs=[],
+        actual_deals=[],
+        expires_at=datetime.now(tz=UTC) + timedelta(hours=24),
+    )
+
+
+@pytest.fixture()
+def app() -> FastAPI:
+    """Минимальное приложение вокруг trade_in-роутера; DB не используется реально —
+    account_quota мокается no-op в каждом тесте отдельно."""
+    application = FastAPI()
+    application.include_router(trade_in_module.router, prefix="/api/v1/trade-in")
+
+    def _override_db():
+        yield None
+
+    application.dependency_overrides[get_db] = _override_db
+    return application
+
+
+@pytest.fixture(autouse=True)
+def _narrow_estimate_limiter(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Узкий лимитер (2 запроса / 60с) — тестируем СРАБАТЫВАНИЕ механизма, не
+    прод-пороги (settings.estimate_rate_limit=5 / estimate_rate_limit_window_s=300
+    было бы медленно/шумно гонять напрямую в юнит-тесте)."""
+    monkeypatch.setattr(
+        trade_in_module, "_estimate_limiter", SlidingWindowLimiter(limit=2, window_s=60.0)
+    )
+
+
+def _post_estimate(client: TestClient, headers: dict[str, str] | None = None):
+    return client.post(
+        "/api/v1/trade-in/estimate",
+        json={"address": "г. Екатеринбург, ул. Малышева, 1", "area_m2": 50.0, "rooms": 2},
+        headers=headers or {},
+    )
+
+
+def test_estimate_rate_limit_fires_after_narrow_threshold(app: FastAPI) -> None:
+    """3-й запрос той же анонимной корзины (лимит=2) → 429 с Retry-After и текстом
+    ПРО ОЦЕНКУ (отличимо от общего RateLimitMiddleware "Слишком много запросов")."""
+    client = TestClient(app, raise_server_exceptions=False)
+    with (
+        patch("app.services.account_quota.check_and_raise"),
+        patch("app.services.account_quota.increment", return_value=True),
+        patch(
+            "app.services.estimator.estimate_quality",
+            new=AsyncMock(return_value=_canned_estimate()),
+        ),
+    ):
+        for _ in range(2):
+            resp = _post_estimate(client)
+            assert resp.status_code == 200
+
+        blocked = _post_estimate(client)
+
+    assert blocked.status_code == 429
+    assert "оценку" in blocked.json()["detail"]
+    assert "Retry-After" in blocked.headers
+
+
+def test_estimate_rate_limit_per_key_isolation(app: FastAPI) -> None:
+    """alice упирается в узкий лимит; bob (свой ключ) — не задет."""
+    client = TestClient(app, raise_server_exceptions=False)
+    with (
+        patch("app.services.account_quota.check_and_raise"),
+        patch("app.services.account_quota.increment", return_value=True),
+        patch(
+            "app.services.estimator.estimate_quality",
+            new=AsyncMock(return_value=_canned_estimate()),
+        ),
+    ):
+        alice = {"X-Authenticated-User": "alice"}
+        for _ in range(2):
+            assert _post_estimate(client, alice).status_code == 200
+        assert _post_estimate(client, alice).status_code == 429
+
+        bob = {"X-Authenticated-User": "bob"}
+        assert _post_estimate(client, bob).status_code == 200
+
+
+def test_estimate_rate_limit_applies_regardless_of_quota_role(app: FastAPI) -> None:
+    """Rate limit — самая дешёвая проверка, стоит ПЕРВОЙ в хендлере (до
+    account_quota). Даже quota-unlimited роль (admin/kopylov) упирается в него —
+    burst-защита capacity сервера не зависит от business-роли."""
+    client = TestClient(app, raise_server_exceptions=False)
+    admin_headers = {"X-Authenticated-User": "admin"}
+    with (
+        patch("app.services.account_quota.check_and_raise") as mock_check,
+        patch("app.services.account_quota.increment", return_value=True),
+        patch(
+            "app.services.estimator.estimate_quality",
+            new=AsyncMock(return_value=_canned_estimate()),
+        ),
+    ):
+        for _ in range(2):
+            assert _post_estimate(client, admin_headers).status_code == 200
+
+        blocked = _post_estimate(client, admin_headers)
+
+    assert blocked.status_code == 429
+    # account_quota.check_and_raise НЕ вызывается для 3-го запроса — rate limit
+    # короткозамкнул обработку раньше, чем дело дошло до квоты.
+    assert mock_check.call_count == 2

tradein-mvp/backend/tests/test_estimator_imv_budget.py

@@ -0,0 +1,132 @@
+"""Avito IMV time-budget — slow/hanging IMV must degrade gracefully, not block
+/estimate (#b2c-antiabuse-2).
+
+Prod risk: `_get_or_fetch_imv_cached` was the ONLY external call in the estimate
+pipeline WITHOUT a `_with_budget` guard. `evaluate_via_imv` chains up to 3
+sequential HTTP requests (warm-up + geocode + evaluate), each with its own 25s
+timeout (`_HTTP_TIMEOUT_SEC` in scraper_kit.providers.avito.imv), plus a possible
+ONE internal retry with a "cleaned" address on `IMVAddressNotFoundError` — the
+unbounded worst-case reached ~150s. Every other slow enrichment (geocode/
+house_metadata/yandex_valuation/cian_valuation) was already wrapped in
+`_with_budget`; this mirrors that guard for IMV.
+
+Contracts locked here:
+  1. `estimate_avito_imv_timeout_s` exists with the documented default (20.0s).
+  2. An IMV timeout (TimeoutError surfaced by the _with_budget wrapper) degrades
+     to an AggregatedEstimate WITHOUT 'avito_imv' in sources_used and no 5xx —
+     the same graceful None path as a network error.
+  3. `_with_budget` actually enforces the timeout (elapsed time bound), proving
+     the guard is live and not merely present in code.
+
+Style mirrors test_estimator_cian_budget.py.
+"""
+
+import os
+
+# Settings requires DATABASE_URL at init time. Set dummy DSN before any app import.
+os.environ.setdefault("DATABASE_URL", "postgresql://test:test@localhost/test_db")
+
+import asyncio
+import time
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import anyio
+
+
+def _make_fake_geo():
+    from app.services.geocoder import GeocodeResult
+
+    return GeocodeResult(
+        lat=56.838,
+        lon=60.595,
+        full_address="Свердловская обл., Екатеринбург, ул. Учителей, 18",
+        provider="nominatim",
+    )
+
+
+def _make_payload_full():
+    """house_type/repair_state set so the IMV-gated guard is actually satisfied
+    (imv_house_type/imv_renovation both non-None — see _IMV_HOUSE_TYPE_MAP /
+    _IMV_REPAIR_MAP in estimator.py)."""
+    from app.schemas.trade_in import TradeInEstimateInput
+
+    return TradeInEstimateInput(
+        address="ЕКБ, ул. Учителей, 18",
+        area_m2=38.8,
+        rooms=1,
+        floor=4,
+        total_floors=16,
+        house_type="panel",
+        repair_state="standard",
+    )
+
+
+def test_avito_imv_timeout_setting_default() -> None:
+    """The new budget setting exists and defaults to 20.0s."""
+    from app.core.config import settings
+
+    assert hasattr(settings, "estimate_avito_imv_timeout_s")
+    assert settings.estimate_avito_imv_timeout_s == 20.0
+
+
+def test_estimate_avito_imv_timeout_degrades_no_5xx() -> None:
+    """IMV raising TimeoutError → estimate without it, no exception.
+
+    The _with_budget() guard wraps the IMV call; a TimeoutError must map to the
+    same graceful None path as a network error. Estimator returns an
+    AggregatedEstimate; 'avito_imv' absent from sources_used.
+    """
+    from app.services.estimator import estimate_quality
+
+    db = MagicMock()
+    payload = _make_payload_full()
+
+    imv_mock = AsyncMock(side_effect=TimeoutError("imv slow"))
+
+    async def _run() -> None:
+        with (
+            patch("app.services.estimator.geocode", new=AsyncMock(return_value=_make_fake_geo())),
+            patch("app.services.estimator.get_house_metadata", new=AsyncMock(return_value=None)),
+            patch("app.services.estimator._fetch_analogs", return_value=([], False, "W")),
+            patch("app.services.estimator._fetch_deals", return_value=[]),
+            patch("app.services.estimator._get_or_fetch_imv_cached", new=imv_mock),
+            patch(
+                "app.services.estimator._get_or_fetch_yandex_valuation_cached",
+                new=AsyncMock(return_value=None),
+            ),
+            patch(
+                "app.services.estimator.estimate_via_cian_valuation",
+                new=AsyncMock(return_value=None),
+            ),
+            patch("app.services.estimator._get_asking_sold_ratio", return_value=(None, None)),
+        ):
+            result = await estimate_quality(payload, db)
+
+        assert result.estimate_id is not None  # no 5xx — degraded gracefully
+        assert "avito_imv" not in result.sources_used
+
+    anyio.run(_run)
+
+
+def test_estimate_avito_imv_timeout_within_budget() -> None:
+    """`_with_budget` actually enforces the configured timeout — a hanging IMV
+    coroutine is cancelled at the budget, not left to run the unbounded
+    (~150s worst-case) evaluate_via_imv chain. Uses a short synthetic budget
+    (not the real 20.0s default) so this test stays fast."""
+    from app.services.estimator import _with_budget
+
+    short_budget_s = 0.2
+
+    async def _hangs_forever() -> None:
+        await asyncio.sleep(999)
+
+    async def _run() -> None:
+        start = time.monotonic()
+        result = await _with_budget(_hangs_forever(), short_budget_s, label="avito_imv")
+        elapsed = time.monotonic() - start
+        assert result is None
+        # Generous slack for CI scheduling jitter — proves the guard actually
+        # cancels near the budget, not merely that it exists in code.
+        assert elapsed < short_budget_s + 2.0
+
+    anyio.run(_run)