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

@@ -1,156 +0,0 @@
-"""POST /api/v1/auth/login + /logout — DB-backed session auth (#2552, эпик #2549).
-
-Переходный механизм, параллельный legacy Caddy trusted-header auth (roles.yaml).
-См. `app.core.rbac.rbac_guard` (dual-mode resolver) и `app.services.auth_session`
-(session CRUD). Mounted at `/api/v1/auth`; через Caddy `uri strip_prefix /trade-in`
-это `/trade-in/api/v1/auth/*` снаружи.
-
-Security:
-  - Неверные creds (неизвестный username / неактивен / password_hash NULL /
-    неверный пароль) → ОДИНАКОВЫЙ 401 с generic сообщением — не раскрываем,
-    существует ли username (user-enumeration защита).
-  - #2552 post-review Medium 2: `verify_password` ВСЕГДА вызывается ровно
-    один раз — для несуществующего username / NULL password_hash сверяем
-    против статичного dummy-хеша (`_DUMMY_PASSWORD_HASH`, сгенерирован один
-    раз на импорте модуля), результат игнорируется. Без этого короткое
-    замыкание (`user is None → сразу 401`) давало наблюдаемую разницу во
-    времени ответа (~1мс без bcrypt vs ~100-300мс с ним) — классический
-    timing-oracle для user-enumeration, даже при одинаковом detail-сообщении.
-  - Rate-limit по (username, IP) — ЖЁСТЧЕ общего `RateLimitMiddleware`
-    (`/api/*`), т.к. login — типичная brute-force поверхность. Использует
-    `SlidingWindowLimiter` (тот же примитив, что и общий rate-limit). Ключ
-    length-prefixed (`len(username):username:ip`) — без этого произвольный
-    username с `:` внутри мог бы схлопнуть бюджет с другой (username, ip)
-    парой (IPv6-адреса тоже содержат `:`, так что просто эскейпить разделитель
-    в username недостаточно — паразитная граница возможна с обеих сторон).
-  - Raw-пароль НИКОГДА не логируется и не попадает в user_events payload —
-    только username/ip/user_agent/path/method (см. schedule_event ниже).
-"""
-
-from __future__ import annotations
-
-import logging
-import secrets
-from typing import Annotated
-
-from fastapi import APIRouter, Depends, HTTPException, Request, Response
-from pydantic import BaseModel
-from sqlalchemy.orm import Session
-
-from app.core.config import settings
-from app.core.db import get_db
-from app.core.password import hash_password, verify_password
-from app.core.ratelimit import SlidingWindowLimiter, _client_ip
-from app.services.auth_session import create_session, get_user_by_username, revoke_session
-from app.services.user_events import schedule_event
-
-logger = logging.getLogger(__name__)
-
-router = APIRouter()
-
-# Отдельный, более узкий бюджет чем общий per-user/per-IP `/api/*` лимит
-# (см. app.core.ratelimit.SlidingWindowLimiter docstring — designed именно для
-# такого случая). Ключ = username+IP: не даёт распределённому brute-force по
-# ОДНОМУ аккаунту с разных IP уйти от лимита целиком (per-IP было бы недостаточно),
-# и не блокирует ВЕСЬ IP из-за перебора чужих логинов одним же клиентом.
-_LOGIN_LIMITER = SlidingWindowLimiter(
-    limit=settings.login_rate_limit,
-    window_s=settings.login_rate_limit_window_s,
-)
-
-# Timing-oracle защита (см. module docstring): bcrypt-хеш случайного пароля,
-# сгенерированный ОДИН РАЗ на импорте модуля — используется вместо
-# password_hash, когда юзер не найден/деактивирован/без пароля, чтобы
-# `verify_password` (доминирующая по времени операция, ~100-300мс) всегда
-# отрабатывала полный bcrypt-компар, независимо от того, существует ли аккаунт.
-_DUMMY_PASSWORD_HASH = hash_password(secrets.token_urlsafe(16))
-
-_INVALID_CREDENTIALS_DETAIL = "неверный логин или пароль"
-
-
-class LoginRequest(BaseModel):
-    username: str
-    password: str
-
-
-class LoginResponse(BaseModel):
-    ok: bool = True
-
-
-@router.post("/login", response_model=LoginResponse)
-async def login(
-    body: LoginRequest,
-    request: Request,
-    response: Response,
-    db: Annotated[Session, Depends(get_db)],
-) -> LoginResponse:
-    ip = _client_ip(request)
-    user_agent = request.headers.get("user-agent")
-    rate_key = f"{len(body.username)}:{body.username}:{ip}"
-
-    retry_after = _LOGIN_LIMITER.check(rate_key)
-    if retry_after is not None:
-        raise HTTPException(
-            status_code=429,
-            detail="слишком много попыток входа, попробуйте позже",
-            headers={"Retry-After": str(int(retry_after) + 1)},
-        )
-
-    user = get_user_by_username(db, body.username)
-    hash_to_check = (
-        user["password_hash"]
-        if user is not None and user["password_hash"] is not None
-        else _DUMMY_PASSWORD_HASH
-    )
-    # ВСЕГДА вызывается — dummy-хеш при отсутствующем юзере/NULL password_hash
-    # держит время ответа одинаковым независимо от существования аккаунта.
-    password_ok = verify_password(body.password, hash_to_check)
-    credentials_ok = user is not None and user["is_active"] and password_ok
-
-    if not credentials_ok:
-        schedule_event(
-            event_type="login_failed",
-            username=body.username,
-            ip=ip,
-            user_agent=user_agent,
-            path="/api/v1/auth/login",
-            method="POST",
-        )
-        raise HTTPException(status_code=401, detail=_INVALID_CREDENTIALS_DETAIL)
-
-    assert user is not None  # narrowed by credentials_ok above
-    token = create_session(db, user_id=user["user_id"], ip=ip, user_agent=user_agent)
-
-    response.set_cookie(
-        key=settings.session_cookie_name,
-        value=token,
-        max_age=settings.session_ttl_hours * 3600,
-        httponly=True,
-        secure=True,
-        samesite="lax",
-        path="/",
-    )
-
-    schedule_event(
-        event_type="login_success",
-        username=user["username"],
-        ip=ip,
-        user_agent=user_agent,
-        path="/api/v1/auth/login",
-        method="POST",
-    )
-
-    return LoginResponse(ok=True)
-
-
-@router.post("/logout")
-async def logout(
-    request: Request,
-    response: Response,
-    db: Annotated[Session, Depends(get_db)],
-) -> dict[str, bool]:
-    token = request.cookies.get(settings.session_cookie_name)
-    if token:
-        revoke_session(db, token)
-    response.delete_cookie(key=settings.session_cookie_name, path="/")
-    return {"ok": True}

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

@@ -7,26 +7,16 @@ Mounted at /api/v1/me; через Caddy `uri strip_prefix /trade-in` это ст
 Caddy basic_auth пропускает `X-Authenticated-User: <username>` через
 `header_up` в каждом reverse_proxy. Frontend дёргает /me чтобы понять
 кому что показывать.
-
-#2552: session-first. Валидная DB-session cookie (см. app.services.auth_session)
-отдаёт scope из tradein_users (role/display_name/org/email) БЕЗ похода в
-roles.yaml. Без cookie (или невалидная/истёкшая) — legacy X-Authenticated-User
-путь, БЕЗ ИЗМЕНЕНИЙ (regression недопустим — существующие тесты держат его
-бит-в-бит).
 """
 
 from __future__ import annotations
 
 import logging
-from typing import Annotated, Any
+from typing import Annotated
 
-from fastapi import APIRouter, Depends, Header, HTTPException, Request
+from fastapi import APIRouter, Header, HTTPException
-from sqlalchemy.orm import Session
 
 from app.core.auth import UserScope, get_user_scope
-from app.core.config import settings
-from app.core.db import get_db
-from app.services.auth_session import get_db_role_scope, get_session_user
 
 logger = logging.getLogger(__name__)
 
@@ -35,40 +25,9 @@ router = APIRouter()
 
 @router.get("/me")
 async def me(
-    request: Request,
-    db: Annotated[Session, Depends(get_db)],
     x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
-) -> UserScope | dict[str, Any]:
+) -> UserScope:
-    """Return the current user's RBAC scope (role + allowed/deny paths).
+    """Return the current user's RBAC scope (role + allowed/deny paths)."""
-
-    Return type is a union (не только `UserScope`) — `UserScope.role` — это
-    `Literal["admin","pilot","analyst","expired"]` (legacy roles.yaml names),
-    а DB-роли (tradein_users.role) — `"admin"/"manager"/"employee"`. FastAPI
-    строит response-схему из return-аннотации; жёсткий `UserScope` завернул бы
-    "employee"/"manager" в ResponseValidationError. Итоговая JSON-форма
-    ОДИНАКОВАЯ (те же 8 ключей) для обеих веток.
-    """
-    token = request.cookies.get(settings.session_cookie_name)
-    if token:
-        try:
-            session_user = get_session_user(db, token)
-        except Exception:
-            logger.exception("me: session lookup failed")
-            session_user = None
-        if session_user is not None:
-            role = session_user["role"]
-            allowed_paths, deny_paths = get_db_role_scope(role)
-            return {
-                "username": session_user["username"],
-                "role": role,
-                "allowed_paths": allowed_paths,
-                "deny_paths": deny_paths,
-                "brand": None,
-                "display_name": session_user["display_name"],
-                "org": session_user["org_name"],
-                "email": session_user["email"],
-            }
-
     if not x_authenticated_user:
         raise HTTPException(
             status_code=401,

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

@@ -7,16 +7,12 @@ manually". The copy drifted: it was missing the #2213
 ``X-Internal-Auth-Secret`` defense-in-depth check that the real guard has,
 so a regression in that check would NOT have failed CI.
 
-This module holds the real guard. Historically it had "no DB/lifespan/scheduler
+This module holds the real guard with no DB/lifespan/scheduler side effects
-side effects" beyond ``app.core.auth``/``app.core.config`` (both side-effect-free
+(only ``app.core.auth`` + ``app.core.config``, both side-effect-free at
-at import time). #2552 (dual-mode DB-session auth) adds a conditional per-request
+import time beyond requiring ``DATABASE_URL`` in the environment for
-DB round trip via ``app.core.db.SessionLocal`` — но ТОЛЬКО когда запрос реально
+``Settings()``). ``app/main.py`` and the test apps both import THIS module,
-несёт session-cookie (``request.cookies.get(settings.session_cookie_name)``);
+so tests exercise the exact production code path instead of a copy that can
-без cookie (весь существующий тестовый трафик, legacy Caddy trusted-header
+silently fall out of sync.
-запросы) ветка не выполняется — ноль новых DB-побочных эффектов для старых
-путей. ``app/main.py`` and the test apps both import THIS module, so tests
-exercise the exact production code path instead of a copy that can silently
-fall out of sync.
 """
 
 from __future__ import annotations
@@ -25,15 +21,12 @@ import logging
 import re
 import secrets
 from collections.abc import Awaitable, Callable
-from typing import Any
 
 from fastapi import Request
 from fastapi.responses import JSONResponse, Response
 
 from app.core.auth import get_role, is_path_allowed
 from app.core.config import settings
-from app.core.db import SessionLocal
-from app.services.auth_session import get_db_role_scope, get_session_user
 
 logger = logging.getLogger(__name__)
 
@@ -47,22 +40,7 @@ logger = logging.getLogger(__name__)
 # Public paths без auth (/health, /docs, /openapi.json) пропускаем —
 # X-Authenticated-User там не приходит из Caddy.
 _ADMIN_API_RE = re.compile(r"^/api/v1/admin/")
-# #2552: /api/v1/auth/login + /logout — по определению вызываются ДО того, как
+_PUBLIC_PATHS = frozenset({"/health", "/docs", "/redoc", "/openapi.json"})
-# клиент аутентифицирован (login) или могут вызываться с уже протухшей/отсутствующей
-# сессией (logout — должен уметь чистить stale cookie без валидной auth). Свой
-# rate-limit у /login отдельный (app.api.v1.auth._LOGIN_LIMITER), RateLimitMiddleware
-# на /api/* всё равно применяется — это ослабляет ТОЛЬКО rbac_guard'овский
-# auth-required gate, не остальные защиты.
-_PUBLIC_PATHS = frozenset(
-    {
-        "/health",
-        "/docs",
-        "/redoc",
-        "/openapi.json",
-        "/api/v1/auth/login",
-        "/api/v1/auth/logout",
-    }
-)
 # #R2-H3: Caddy срезает внешний префикс /trade-in (uri strip_prefix) перед
 # tradein-backend, а globs в roles.yaml — ВНЕШНИЕ (/trade-in/api/v1/**). Для
 # scope-проверки восстанавливаем внешний путь.
@@ -73,77 +51,6 @@ _EXTERNAL_PREFIX = "/trade-in"
 _RBAC_BOOTSTRAP_EXEMPT = ("/api/v1/me", "/api/v1/brand")
 
 
-def _db_glob_match(pattern: str, path: str) -> bool:
-    """Мини-матчер для фиксированного набора DB-role паттернов
-    (``app.services.auth_session.DB_ROLE_PATHS`` — только формы ``/**`` и
-    ``<prefix>/**``, не нужна полная semantics ``app.core.auth._glob_to_regex``
-    — тот модуль private и MIRROR'ится вручную с основным бэкендом, лишний
-    импорт private-символа оттуда увеличивал бы drift-риск)."""
-    if pattern == "/**":
-        return True
-    if pattern.endswith("/**"):
-        prefix = pattern[: -len("/**")]
-        return path == prefix or path.startswith(prefix + "/")
-    return path == pattern
-
-
-def _db_role_path_allowed(role: str, path: str) -> bool:
-    paths, deny = get_db_role_scope(role)
-    if any(_db_glob_match(p, path) for p in deny):
-        return False
-    return any(_db_glob_match(p, path) for p in paths)
-
-
-def _propagate_authenticated_user(request: Request, username: str) -> None:
-    """Инжектит ``X-Authenticated-User`` в ASGI scope — ПЕРЕЗАПИСЫВАЯ, а не
-    только добавляя при отсутствии, — чтобы ``RateLimitMiddleware``/
-    ``RequestAuditMiddleware`` (оба читают сырой заголовок напрямую,
-    #2213/#2550) и downstream route-хендлеры (читающие его через FastAPI
-    ``Header()``) видели РЕЗОЛВЛЕННОГО ИЗ СЕССИИ юзера — без правок в каждом
-    из этих мест по отдельности (минимально инвазивный способ).
-
-    #2552 post-review fix (CRITICAL): раньше это была skip-if-present
-    мутация (``if request.headers.get(...): return``) — сессия резолвилась
-    ПЕРВОЙ (см. rbac_guard), но клиент-контролируемый ``X-Authenticated-User``
-    (который Caddy шлёт на КАЖДЫЙ прод-запрос) выигрывал у неё для ВСЕГО
-    downstream-трафика: атакующий с валидной cookie юзера ``alice`` мог
-    подделать заголовок ``X-Authenticated-User: victim`` и получить доступ к
-    данным victim в ~15 роутах, читающих заголовок напрямую
-    (``_assert_estimate_access*``, ``account_quota``, ``/trade-in/history``,
-    ``support.py``) — работало в ОБОИХ auth_mode (dual и db_only), т.к. эти
-    хендлеры не знают про rbac_guard'овский ``from_session`` флаг, только про
-    сырой заголовок. Session-identity ДОЛЖНА быть источником истины, если
-    сессия резолвлена — полная перезапись, не skip.
-
-    Механизм: ``request.scope`` — ОДИН и тот же dict-объект, прокинутый по
-    ссылке через весь ASGI call chain (Starlette не копирует scope между
-    слоями middleware). Мутация ``scope["headers"]`` ЗДЕСЬ видна:
-      - downstream call_next() цепочке (ExceptionMiddleware → Router →
-        endpoint) — т.к. rbac_guard мутирует scope ДО вызова call_next();
-      - ``RequestAuditMiddleware`` — он внешний относительно rbac_guard
-        (см. app/main.py: последний ``add_middleware`` оборачивает
-        предыдущие) и читает ``request.headers`` уже ПОСЛЕ ``call_next()``
-        отработал весь внутренний стек, включая эту мутацию.
-
-    ASGI header-имена — всегда lowercase bytes (см. ASGI spec), поэтому
-    фильтр по ``b"x-authenticated-user"`` ловит заголовок независимо от
-    регистра, в котором его прислал клиент (Starlette уже нормализует).
-
-    Известное ограничение: ``RateLimitMiddleware`` тоже внешний относительно
-    rbac_guard, но читает заголовок ДО вызова call_next() (до того, как этот
-    guard успевает отработать) — для ЭТОГО конкретного запроса сессионный
-    юзер лимитируется по IP, а не по username (per-user множитель не
-    применяется). Не регрессия (IP-лимит применялся бы и раньше — до
-    добавления session-auth такие запросы вообще были 401), просто более
-    строгий бюджет специфично для session-cookie-запросов; при необходимости
-    точного per-user квотинга для DB-юзеров — переносить резолв сессии выше
-    RateLimit в app/main.py отдельным issue.
-    """
-    request.scope["headers"] = [
-        (k, v) for k, v in request.scope.get("headers", []) if k != b"x-authenticated-user"
-    ] + [(b"x-authenticated-user", username.encode("latin-1", "replace"))]
-
-
 async def rbac_guard(
     request: Request,
     call_next: Callable[[Request], Awaitable[Response]],
@@ -152,74 +59,40 @@ async def rbac_guard(
     if path in _PUBLIC_PATHS:
         return await call_next(request)
 
-    username: str | None = None
+    username = request.headers.get("X-Authenticated-User")
-    role: str | None = None
+    if not username:
-    from_session = False
+        return JSONResponse(
+            status_code=401,
+            content={"detail": "no authenticated user (Caddy basic_auth required)"},
+        )
 
-    # #2552: session-cookie резолвится ПЕРВЫМ. Если cookie нет вообще —
+    # #2213 defense-in-depth: если общий секрет задан — запрос с X-Authenticated-User
-    # request.cookies.get() возвращает None без единого похода в БД (ноль
+    # ОБЯЗАН нести валидный X-Internal-Auth-Secret (его добавляет Caddy из env).
-    # side-effects для всего существующего трафика без cookie).
+    # Иначе это подделка заголовка мимо Caddy (напр. изнутри gendesign_shared) → 401.
-    token = request.cookies.get(settings.session_cookie_name)
+    # Constant-time compare против timing-атак. Пусто = защита не активна (fail-open).
-    if token:
+    secret = settings.tradein_internal_auth_secret
-        session_user: dict[str, Any] | None = None
+    if secret:
-        try:
+        provided = request.headers.get("X-Internal-Auth-Secret", "")
-            with SessionLocal() as db:
+        if not secrets.compare_digest(provided, secret):
-                session_user = get_session_user(db, token)
+            logger.warning(
-        except Exception:
+                "RBAC: X-Authenticated-User=%r без валидного X-Internal-Auth-Secret "
-            logger.exception("RBAC: session lookup failed for %s", path)
+                "на %s — возможная подделка заголовка мимо Caddy",
-        if session_user is not None:
+                username,
-            username = session_user["username"]
+                path,
-            role = session_user["role"]
+            )
-            from_session = True
-            _propagate_authenticated_user(request, username)
-
-    if not from_session:
-        # auth_mode == "db_only" — легаси trusted-header путь ПОЛНОСТЬЮ
-        # отключён, даже если валидный X-Authenticated-User присутствует.
-        if settings.auth_mode != "dual":
             return JSONResponse(
                 status_code=401,
-                content={"detail": "valid session required"},
+                content={"detail": "invalid or missing internal auth secret"},
             )
 
-        # ---- legacy trusted-header path — BIT-FOR-BIT как было до #2552 ----
+    try:
-        username = request.headers.get("X-Authenticated-User")
+        role = get_role(username)
-        if not username:
+    except KeyError:
-            return JSONResponse(
+        logger.warning("RBAC: unknown user %r tried %s", username, path)
-                status_code=401,
+        return JSONResponse(
-                content={"detail": "no authenticated user (Caddy basic_auth required)"},
+            status_code=403,
-            )
+            content={"detail": "user not in roles config"},
-
+        )
-        # #2213 defense-in-depth: если общий секрет задан — запрос с X-Authenticated-User
-        # ОБЯЗАН нести валидный X-Internal-Auth-Secret (его добавляет Caddy из env).
-        # Иначе это подделка заголовка мимо Caddy (напр. изнутри gendesign_shared) → 401.
-        # Constant-time compare против timing-атак. Пусто = защита не активна (fail-open).
-        secret = settings.tradein_internal_auth_secret
-        if secret:
-            provided = request.headers.get("X-Internal-Auth-Secret", "")
-            if not secrets.compare_digest(provided, secret):
-                logger.warning(
-                    "RBAC: X-Authenticated-User=%r без валидного X-Internal-Auth-Secret "
-                    "на %s — возможная подделка заголовка мимо Caddy",
-                    username,
-                    path,
-                )
-                return JSONResponse(
-                    status_code=401,
-                    content={"detail": "invalid or missing internal auth secret"},
-                )
-
-        try:
-            role = get_role(username)
-        except KeyError:
-            logger.warning("RBAC: unknown user %r tried %s", username, path)
-            return JSONResponse(
-                status_code=403,
-                content={"detail": "user not in roles config"},
-            )
-
-    assert username is not None
-    assert role is not None
 
     if _ADMIN_API_RE.match(path) and role != "admin":
         logger.info("RBAC: blocked %s (role=%s) from %s", username, role, path)
@@ -228,24 +101,25 @@ async def rbac_guard(
             content={"detail": "admin only"},
         )
 
-    # #R2-H3: энфорсим scope (paths/deny) для ВСЕХ non-admin путей, а не
+    # #R2-H3: энфорсим roles.yaml scope (paths/deny) для ВСЕХ non-admin путей, а не
-    # только /admin/*. Bootstrap-пути (/me, /brand) исключены — иначе revoked/
+    # только /admin/*. Иначе revoked (role=expired, paths:[] deny:/**) или узко-
-    # scope-narrowed юзер не смог бы получить свою роль вовсе.
+    # скоупленный аккаунт достаёт non-admin API (напр. POST /api/v1/search —
+    # экспорт листингов), который roles.yaml ему запрещает. Bootstrap-пути (/me,
+    # /brand) исключены выше по списку. roles.yaml globs внешние → восстанавливаем
+    # внешний путь (Caddy срезал /trade-in). На сбой парса — fail-open + громкий
+    # лог: не лочим платящего pilot из-за конфиг-бага (admin-гейт выше остаётся).
     if not path.startswith(_RBAC_BOOTSTRAP_EXEMPT):
         external_path = _EXTERNAL_PREFIX + path
-        if from_session:
+        try:
-            allowed = _db_role_path_allowed(role, external_path)
+            allowed = is_path_allowed(role, external_path)
-        else:
+        except Exception:
-            try:
+            logger.exception(
-                allowed = is_path_allowed(role, external_path)
+                "RBAC scope-check raised for %s %s (ext=%s) — fail-open",
-            except Exception:
+                username,
-                logger.exception(
+                path,
-                    "RBAC scope-check raised for %s %s (ext=%s) — fail-open",
+                external_path,
-                    username,
+            )
-                    path,
+            allowed = True
-                    external_path,
-                )
-                allowed = True
         if not allowed:
             logger.info(
                 "RBAC: scope-blocked %s (role=%s) from %s (ext=%s)",

tradein-mvp/backend/app/main.py

@@ -23,7 +23,6 @@ from sentry_sdk.integrations.starlette import StarletteIntegration
 from app.api.v1 import (
     admin,
     audit,
-    auth,
     brand,
     buildings,
     geocode,
@@ -107,19 +106,6 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
             ".env.runtime ОБОИХ стеков (Caddy главного стека + tradein-backend)"
         )
 
-    # #2552: session_secret зарезервирован на будущее (напр. подписанные токены) —
-    # opaque session-токены (secrets.token_urlsafe, см. app.services.auth_session)
-    # НЕ требуют подписи, их валидность проверяется исключительно наличием строки
-    # в tradein_sessions + expires_at/is_active. Пустой session_secret НЕ должен
-    # ронять старт контейнера (не startup-fail) — только громкий WARNING, чтобы
-    # прод не остался без него незамеченно до момента, когда он реально понадобится.
-    if not settings.session_secret:
-        logger.warning(
-            "SESSION_SECRET пуст — не блокирует старт (opaque session-токены не "
-            "требуют подписи), но задай его в .env.runtime до появления фич, "
-            "которым подпись реально нужна"
-        )
-
     # FDW bootstrap: create/refresh USER MAPPING for gendesign_remote postgres_fdw server.
     # Best-effort: failure does not abort startup, just logs.
     try:
@@ -172,7 +158,6 @@ def health() -> dict[str, str]:
     return {"status": "ok", "environment": settings.environment}
 
 
-app.include_router(auth.router, prefix="/api/v1/auth", tags=["auth"])
 app.include_router(geocode.router, prefix="/api/v1/geocode", tags=["geocode"])
 app.include_router(admin.router, prefix="/api/v1/admin", tags=["admin"])
 app.include_router(audit.router, prefix="/api/v1/admin", tags=["admin-audit"])

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

@@ -1,221 +0,0 @@
-"""Session-сервис для DB-backed auth (#2552, эпик #2549 — auth-core).
-
-Схема: `tradein_users` + `tradein_sessions` (migration `192_tradein_users_auth.sql`).
-Опаковые (`secrets.token_urlsafe`) токены-сессии — не JWT, не подписаны: валидность
-проверяется исключительно наличием + `expires_at`/`is_active` строкой в БД, поэтому
-`SESSION_SECRET` НЕ обязателен для работы этого модуля (зарезервирован на будущее,
-см. `app.core.config.Settings.session_secret` docstring).
-
-Все функции здесь принимают уже открытую `db: Session` — сами НЕ открывают
-`SessionLocal()` (вызывающая сторона решает время жизни транзакции: `rbac_guard`
-и `app.core.db.get_db()`-роуты открывают её по-разному). Это делает модуль
-тривиально unit-тестируемым без патчинга `SessionLocal` — тесты просто передают
-fake/real `Session`.
-
-Ни одна функция не должна ронять вызывающий HTTP-запрос: DB-ошибки логируются
-через `logger` вызывающей стороной (см. `app.core.rbac.rbac_guard`,
-`app.api.v1.me`), сам сервис поднимает исключения как есть (это НЕ fire-and-forget
-аудит-лог вроде `app.services.user_events`, а часть auth-decision — сбой обязан
-быть виден вызывающему, чтобы тот мог fail-closed).
-"""
-
-from __future__ import annotations
-
-import logging
-import secrets
-from datetime import UTC, datetime, timedelta
-from typing import Any
-
-from sqlalchemy import text
-from sqlalchemy.orm import Session
-
-from app.core.config import settings
-
-logger = logging.getLogger(__name__)
-
-# Sliding-window refresh: last_seen_at/expires_at продлеваются НЕ чаще раза в
-# 5 минут — иначе каждый API-запрос авторизованного юзера бил бы в БД лишним
-# UPDATE (RBAC гоняет get_session_user на КАЖДЫЙ non-public запрос).
-_SLIDING_REFRESH_INTERVAL = timedelta(minutes=5)
-
-_TOKEN_BYTES = 32  # secrets.token_urlsafe(32) — 256 бит энтропии, ~43 символа
-
-
-def create_session(
-    db: Session,
-    user_id: int,
-    ip: str | None = None,
-    user_agent: str | None = None,
-) -> str:
-    """Создаёт новую сессию для *user_id* и возвращает opaque-токен.
-
-    `expires_at = now() + settings.session_ttl_hours`. Коммитит сам (self-contained,
-    как `app.services.user_events.record_event`).
-    """
-    token = secrets.token_urlsafe(_TOKEN_BYTES)
-    db.execute(
-        text(
-            """
-            INSERT INTO tradein_sessions (token, user_id, expires_at, ip_address, user_agent)
-            VALUES (
-                :token, :user_id,
-                now() + make_interval(hours => CAST(:ttl_hours AS integer)),
-                CAST(:ip AS inet), :user_agent
-            )
-            """
-        ),
-        {
-            "token": token,
-            "user_id": user_id,
-            "ttl_hours": settings.session_ttl_hours,
-            "ip": ip,
-            "user_agent": user_agent,
-        },
-    )
-    db.commit()
-    return token
-
-
-def get_session_user(db: Session, token: str) -> dict[str, Any] | None:
-    """Резолвит сессионный токен в данные юзера, или None если сессия
-    невалидна (не найдена / истекла / юзер деактивирован).
-
-    Sliding refresh: если с последнего `last_seen_at` прошло >=5 минут —
-    продлевает `expires_at`/`last_seen_at` ОДНИМ UPDATE. Сбой refresh
-    (напр. read-replica) логируется и НЕ мешает вернуть валидного юзера —
-    это best-effort продление, а не часть решения "валидна ли сессия".
-    """
-    if not token:
-        return None
-
-    row = db.execute(
-        text(
-            """
-            SELECT s.user_id, s.expires_at, s.last_seen_at,
-                   u.username, u.role, u.display_name, u.org_name, u.email, u.is_active
-            FROM tradein_sessions s
-            JOIN tradein_users u ON u.id = s.user_id
-            WHERE s.token = :token
-            """
-        ),
-        {"token": token},
-    ).fetchone()
-
-    if row is None:
-        return None
-
-    now = datetime.now(UTC)
-    if row.expires_at is None or row.expires_at <= now:
-        return None
-    if not row.is_active:
-        return None
-
-    if row.last_seen_at is None or (now - row.last_seen_at) >= _SLIDING_REFRESH_INTERVAL:
-        try:
-            db.execute(
-                text(
-                    """
-                    UPDATE tradein_sessions
-                    SET last_seen_at = now(),
-                        expires_at = now() + make_interval(hours => CAST(:ttl_hours AS integer))
-                    WHERE token = :token
-                    """
-                ),
-                {"ttl_hours": settings.session_ttl_hours, "token": token},
-            )
-            db.commit()
-        except Exception:
-            logger.warning(
-                "auth_session: sliding refresh failed for user_id=%r", row.user_id, exc_info=True
-            )
-            db.rollback()
-
-    return {
-        "user_id": row.user_id,
-        "username": row.username,
-        "role": row.role,
-        "display_name": row.display_name,
-        "org_name": row.org_name,
-        "email": row.email,
-        "is_active": row.is_active,
-    }
-
-
-def get_user_by_username(db: Session, username: str) -> dict[str, Any] | None:
-    """Возвращает строку `tradein_users` по username, или None если не найден.
-
-    Используется login-флоу (`app.api.v1.auth.login`) для password-проверки.
-    Отдаёт `password_hash` как есть (может быть NULL — переходный период,
-    см. migration 192 docstring) — вызывающая сторона решает, что с ним делать.
-    """
-    row = db.execute(
-        text(
-            """
-            SELECT id, username, password_hash, role, is_active,
-                   display_name, org_name, email
-            FROM tradein_users
-            WHERE username = :username
-            """
-        ),
-        {"username": username},
-    ).fetchone()
-
-    if row is None:
-        return None
-
-    return {
-        "user_id": row.id,
-        "username": row.username,
-        "password_hash": row.password_hash,
-        "role": row.role,
-        "is_active": row.is_active,
-        "display_name": row.display_name,
-        "org_name": row.org_name,
-        "email": row.email,
-    }
-
-
-def revoke_session(db: Session, token: str) -> None:
-    """Удаляет одну сессию по токену (logout). No-op если токен не найден."""
-    db.execute(text("DELETE FROM tradein_sessions WHERE token = :token"), {"token": token})
-    db.commit()
-
-
-def revoke_user_sessions(db: Session, user_id: int) -> None:
-    """Удаляет ВСЕ сессии юзера (напр. смена пароля / принудительный logout всех
-    устройств — не используется этим PR напрямую, задел для будущих admin-действий)."""
-    db.execute(text("DELETE FROM tradein_sessions WHERE user_id = :user_id"), {"user_id": user_id})
-    db.commit()
-
-
-# ---------------------------------------------------------------------------
-# DB-role → RBAC scope (paths/deny) — #2552 dual-mode.
-# ---------------------------------------------------------------------------
-#
-# tradein_users.role ('admin'|'manager'|'employee', CHECK-констрейнт migration 192)
-# НЕ являются ключами auth/roles.yaml (тот файл — legacy Caddy trusted-header путь,
-# который этот эпик намеренно не трогает). Маппинг ниже даёт DB-ролям тот же
-# paths/deny-смысл, что и legacy-ролям, БЕЗ правки roles.yaml:
-#   employee -> те же права, что legacy pilot (/trade-in/** только).
-#   manager  -> employee + задел /api/v1/team/** (роутер появится в #2554).
-#   admin    -> полный доступ, как legacy admin.
-DB_ROLE_PATHS: dict[str, tuple[list[str], list[str]]] = {
-    "employee": (
-        ["/trade-in/**", "/trade-in/api/v1/**"],
-        ["/admin/**", "/api/v1/admin/**", "/trade-in/api/v1/admin/**"],
-    ),
-    "manager": (
-        ["/trade-in/**", "/trade-in/api/v1/**", "/api/v1/team/**"],
-        ["/admin/**", "/api/v1/admin/**", "/trade-in/api/v1/admin/**"],
-    ),
-    "admin": (["/**"], []),
-}
-
-
-def get_db_role_scope(role: str) -> tuple[list[str], list[str]]:
-    """Возвращает (allowed_paths, deny_paths) для DB-роли.
-
-    Неизвестная роль (не должно случиться — CHECK-констрейнт на колонке
-    ограничивает role тремя значениями) -> fail-closed (пустой allow, deny всё).
-    """
-    return DB_ROLE_PATHS.get(role, ([], ["/**"]))

tradein-mvp/backend/data/sql/192_tradein_users_auth.sql

@@ -1,88 +0,0 @@
--- Migration 192: tradein_users + tradein_sessions — DB-backed auth (issue #2551, эпик #2549)
---
--- WHY:
---   Trade-in auth сейчас держится на legacy Caddy basic-auth fallback (см. auth/roles.yaml,
---   упомянут в 191_account_quota_unlimited_flag.sql как "хардкод username в коде"). Эпик #2549
---   переводит auth на DB-backed модель: пользователи + сессии как данные, роли admin/manager/
---   employee с иерархией manager -> employee. Эта миграция — только схема (Foundation),
---   без seed-данных (seed — отдельная задача #2557) и без Python-кода (backend wiring — отдельно).
---
--- WHAT:
---   1. tradein_users — identity + role + org-иерархия.
---      - password_hash NULL допустим: переходный период, когда логин ещё идёт через
---        legacy Caddy fallback, а не через password verify в приложении.
---      - role CHECK ('admin','manager','employee') — три уровня доступа.
---      - manager_id — self-FK, ON DELETE SET NULL (увольнение/удаление manager'а не должно
---        каскадно сносить его employees, они просто остаются без привязки).
---      - CHECK role_manager_hierarchy: admin/manager обязаны иметь manager_id IS NULL
---        (это top-level роли, у них нет "начальника" в этой модели); employee — manager_id
---        любой, включая NULL (свободный слот employee без организации допустим).
---   2. tradein_sessions — токен-based сессии, привязаны к user_id, ON DELETE CASCADE
---      (удалили пользователя — его сессии теряют смысл, каскадная очистка корректна).
---      last_seen_at отдельно от created_at — для idle-timeout / активности сессии.
---   3. Индексы: expires_at (уборка протухших сессий), user_id (список сессий юзера),
---      partial на manager_id (иерархия) — WHERE manager_id IS NOT NULL, т.к. большинство
---      admin/manager строк это NULL и не участвуют в lookup "employees этого manager'а".
---
--- IDEMPOTENCY:
---   CREATE TABLE IF NOT EXISTS + CREATE INDEX IF NOT EXISTS. Повторный прогон — no-op.
---   CHECK-констрейнты добавлены inline в CREATE TABLE (не через ALTER) — при повторном
---   запуске CREATE TABLE IF NOT EXISTS не выполнится вообще, констрейнт не задублируется.
---
--- Dependencies: нет (новые таблицы, ничего существующего не меняем).
--- Deploy order: эта миграция — Foundation эпика #2549. Seed (#2557) и backend auth-код —
---   отдельные PR'ы ПОСЛЕ этой (SQL-схема первой, см. .claude/rules/sql.md "Migration order").
-
-BEGIN;
-
-CREATE TABLE IF NOT EXISTS tradein_users (
-    id             bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
-    username       text NOT NULL UNIQUE,
-    password_hash  text NULL,
-    role           text NOT NULL CHECK (role IN ('admin', 'manager', 'employee')),
-    manager_id     bigint NULL REFERENCES tradein_users(id) ON DELETE SET NULL,
-    display_name   text NULL,
-    org_name       text NULL,
-    email          text NULL,
-    is_active      boolean NOT NULL DEFAULT true,
-    created_at     timestamptz NOT NULL DEFAULT now(),
-    updated_at     timestamptz NOT NULL DEFAULT now(),
-    CONSTRAINT tradein_users_role_manager_hierarchy_ck CHECK (
-        role NOT IN ('admin', 'manager') OR manager_id IS NULL
-    )
-);
-
-COMMENT ON TABLE tradein_users IS
-    'Trade-in DB-backed auth — пользователи (issue #2551, эпик #2549). password_hash NULL '
-    'допустим в переходный период (логин через legacy Caddy fallback). Seed — отдельно (#2557).';
-COMMENT ON COLUMN tradein_users.password_hash IS
-    'NULL = логин только через legacy Caddy basic-auth fallback, не через password verify.';
-COMMENT ON COLUMN tradein_users.manager_id IS
-    'Self-FK на tradein_users(id). NULL для admin/manager (top-level, CHECK ниже) или для '
-    'employee без назначенной организации.';
-
-CREATE TABLE IF NOT EXISTS tradein_sessions (
-    token         text PRIMARY KEY,
-    user_id       bigint NOT NULL REFERENCES tradein_users(id) ON DELETE CASCADE,
-    created_at    timestamptz NOT NULL DEFAULT now(),
-    expires_at    timestamptz NOT NULL,
-    last_seen_at  timestamptz NOT NULL DEFAULT now(),
-    ip_address    inet NULL,
-    user_agent    text NULL
-);
-
-COMMENT ON TABLE tradein_sessions IS
-    'Trade-in DB-backed auth — активные сессии (issue #2551, эпик #2549). '
-    'ON DELETE CASCADE от tradein_users: удалённый пользователь теряет все сессии.';
-
-CREATE INDEX IF NOT EXISTS tradein_sessions_expires_at_idx
-    ON tradein_sessions (expires_at);
-
-CREATE INDEX IF NOT EXISTS tradein_sessions_user_id_idx
-    ON tradein_sessions (user_id);
-
-CREATE INDEX IF NOT EXISTS tradein_users_manager_id_idx
-    ON tradein_users (manager_id)
-    WHERE manager_id IS NOT NULL;
-
-COMMIT;

tradein-mvp/backend/tests/test_auth_api.py

@@ -1,536 +0,0 @@
-"""Integration tests for #2552 auth-core: POST /login, /logout, dual-mode GET /me,
-and rbac_guard session-cookie resolution.
-
-Uses the REAL `rbac_guard` (app.core.rbac) + REAL `auth.router` / `me.router` wired
-into an isolated FastAPI test app (same pattern as tests/test_rbac.py), with an
-in-memory fake DB standing in for `tradein_users`/`tradein_sessions`:
-  - `app.core.rbac.SessionLocal` is monkeypatched (rbac_guard opens its own session,
-    it's middleware — no FastAPI DI available there).
-  - `app.core.db.get_db` is overridden via `app.dependency_overrides` (auth.py /
-    me.py use `Depends(get_db)`, the idiomatic FastAPI-testable path).
-
-Both point at the SAME `_Store` instance per test, so a session created by POST
-/login is immediately visible to rbac_guard's own DB round trip on the next request.
-"""
-
-from __future__ import annotations
-
-import os
-from datetime import UTC, datetime, timedelta
-from types import SimpleNamespace
-from typing import Annotated, Any
-
-os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
-
-import pytest
-from fastapi import FastAPI, Header
-from fastapi.testclient import TestClient
-
-from app.api.v1 import auth as auth_router
-from app.api.v1 import me as me_router
-from app.core import auth as auth_mod
-from app.core import config
-from app.core.db import get_db
-from app.core.password import hash_password
-from app.core.rbac import rbac_guard
-
-# ---------------------------------------------------------------------------
-# Fake DB backing tradein_users / tradein_sessions
-# ---------------------------------------------------------------------------
-
-
-class _Store:
-    def __init__(self) -> None:
-        self.users: dict[str, dict[str, Any]] = {}
-        self.sessions: dict[str, dict[str, Any]] = {}
-        self._next_id = 1
-
-    def add_user(
-        self,
-        username: str,
-        password_hash: str | None,
-        *,
-        role: str = "employee",
-        is_active: bool = True,
-        display_name: str | None = "Alice A.",
-        org_name: str | None = "Org LLC",
-        email: str | None = "alice@example.com",
-    ) -> int:
-        uid = self._next_id
-        self._next_id += 1
-        self.users[username] = {
-            "id": uid,
-            "username": username,
-            "password_hash": password_hash,
-            "role": role,
-            "is_active": is_active,
-            "display_name": display_name,
-            "org_name": org_name,
-            "email": email,
-        }
-        return uid
-
-    def user_by_id(self, uid: int) -> dict[str, Any] | None:
-        for u in self.users.values():
-            if u["id"] == uid:
-                return u
-        return None
-
-    def add_expired_session(self, token: str, user_id: int) -> None:
-        now = datetime.now(UTC)
-        self.sessions[token] = {
-            "user_id": user_id,
-            "expires_at": now - timedelta(minutes=1),
-            "last_seen_at": now - timedelta(minutes=1),
-        }
-
-
-class _FakeDB:
-    """Minimal Session stand-in dispatching on SQL text — see module docstring."""
-
-    def __init__(self, store: _Store) -> None:
-        self.store = store
-
-    def __enter__(self) -> _FakeDB:
-        return self
-
-    def __exit__(self, *exc: object) -> bool:
-        return False
-
-    def close(self) -> None:
-        pass
-
-    def commit(self) -> None:
-        pass
-
-    def rollback(self) -> None:
-        pass
-
-    def execute(self, stmt: object, params: dict[str, Any] | None = None) -> SimpleNamespace:
-        sql = str(stmt)
-        p = params or {}
-
-        if "INSERT INTO tradein_sessions" in sql:
-            now = datetime.now(UTC)
-            self.store.sessions[p["token"]] = {
-                "user_id": p["user_id"],
-                "expires_at": now + timedelta(hours=p["ttl_hours"]),
-                "last_seen_at": now,
-            }
-            return SimpleNamespace(fetchone=lambda: None)
-
-        if "UPDATE tradein_sessions" in sql and "SET last_seen_at" in sql:
-            sess = self.store.sessions.get(p["token"])
-            if sess is not None:
-                now = datetime.now(UTC)
-                sess["last_seen_at"] = now
-                sess["expires_at"] = now + timedelta(hours=p["ttl_hours"])
-            return SimpleNamespace(fetchone=lambda: None)
-
-        if "DELETE FROM tradein_sessions WHERE token" in sql:
-            self.store.sessions.pop(p["token"], None)
-            return SimpleNamespace(fetchone=lambda: None)
-
-        if "DELETE FROM tradein_sessions WHERE user_id" in sql:
-            uid = p["user_id"]
-            for tok in [t for t, s in self.store.sessions.items() if s["user_id"] == uid]:
-                del self.store.sessions[tok]
-            return SimpleNamespace(fetchone=lambda: None)
-
-        if "FROM tradein_sessions s" in sql and "JOIN tradein_users u" in sql:
-            sess = self.store.sessions.get(p["token"])
-            if sess is None:
-                return SimpleNamespace(fetchone=lambda: None)
-            user = self.store.user_by_id(sess["user_id"])
-            if user is None:
-                return SimpleNamespace(fetchone=lambda: None)
-            row = SimpleNamespace(
-                user_id=sess["user_id"],
-                expires_at=sess["expires_at"],
-                last_seen_at=sess["last_seen_at"],
-                username=user["username"],
-                role=user["role"],
-                display_name=user["display_name"],
-                org_name=user["org_name"],
-                email=user["email"],
-                is_active=user["is_active"],
-            )
-            return SimpleNamespace(fetchone=lambda: row)
-
-        if "FROM tradein_users" in sql:
-            user = self.store.users.get(p["username"])
-            if user is None:
-                return SimpleNamespace(fetchone=lambda: None)
-            row = SimpleNamespace(**user)
-            return SimpleNamespace(fetchone=lambda: row)
-
-        raise AssertionError(f"unhandled fake SQL in test_auth_api: {sql!r}")
-
-
-# ---------------------------------------------------------------------------
-# Test app
-# ---------------------------------------------------------------------------
-
-
-def _build_test_app(store: _Store) -> FastAPI:
-    app = FastAPI()
-    app.middleware("http")(rbac_guard)
-    app.include_router(auth_router.router, prefix="/api/v1/auth", tags=["auth"])
-    app.include_router(me_router.router, prefix="/api/v1", tags=["me"])
-
-    @app.get("/api/v1/trade-in/dummy")
-    async def tradein_dummy() -> dict:
-        return {"ok": True}
-
-    @app.get("/api/v1/trade-in/whoami")
-    async def tradein_whoami(
-        x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
-    ) -> dict:
-        """Echoes the X-Authenticated-User header exactly as a downstream handler
-        (`_assert_estimate_access*`, `account_quota`, etc.) would see it — used to
-        assert session-identity wins over a client-forged header (#2552 spoof fix)."""
-        return {"user": x_authenticated_user}
-
-    def _override_get_db():  # generator dependency — matches app.core.db.get_db shape
-        yield _FakeDB(store)
-
-    app.dependency_overrides[get_db] = _override_get_db
-    return app
-
-
-@pytest.fixture(autouse=True)
-def _reset_state(monkeypatch: pytest.MonkeyPatch) -> None:
-    auth_mod.reset_cache_for_tests()
-    auth_router._LOGIN_LIMITER._hits.clear()
-    monkeypatch.setattr(config.settings, "auth_mode", "dual")
-
-
-@pytest.fixture
-def store() -> _Store:
-    return _Store()
-
-
-@pytest.fixture
-def client(store: _Store, monkeypatch: pytest.MonkeyPatch) -> TestClient:
-    monkeypatch.setattr("app.core.rbac.SessionLocal", lambda: _FakeDB(store))
-    # base_url=https:// — login sets the session cookie with Secure=True (real prod
-    # behaviour, not weakened for tests); httpx's cookie jar silently drops Secure
-    # cookies on a plain-http connection, so a plain http://testserver client would
-    # never resend the cookie on subsequent requests within the same test.
-    return TestClient(_build_test_app(store), base_url="https://testserver")
-
-
-# ---------------------------------------------------------------------------
-# POST /login
-# ---------------------------------------------------------------------------
-
-
-def test_login_happy_path_sets_cookie(client: TestClient, store: _Store) -> None:
-    store.add_user("alice", hash_password("Secret123!"), role="employee")
-
-    resp = client.post("/api/v1/auth/login", json={"username": "alice", "password": "Secret123!"})
-
-    assert resp.status_code == 200, resp.text
-    assert resp.json() == {"ok": True}
-    cookie_name = config.settings.session_cookie_name
-    assert cookie_name in resp.cookies
-    assert resp.cookies[cookie_name]
-    # Сессия реально создана в сторе под этим токеном.
-    assert resp.cookies[cookie_name] in store.sessions
-
-
-def test_login_wrong_password_401_and_logs_failed_event(
-    client: TestClient, store: _Store, monkeypatch: pytest.MonkeyPatch
-) -> None:
-    store.add_user("alice", hash_password("Secret123!"), role="employee")
-    events: list[dict[str, Any]] = []
-    monkeypatch.setattr(auth_router, "schedule_event", lambda **kw: events.append(kw))
-
-    resp = client.post("/api/v1/auth/login", json={"username": "alice", "password": "wrong"})
-
-    assert resp.status_code == 401
-    assert "detail" in resp.json()
-    assert config.settings.session_cookie_name not in resp.cookies
-    assert any(e["event_type"] == "login_failed" for e in events)
-    failed = next(e for e in events if e["event_type"] == "login_failed")
-    assert failed["username"] == "alice"
-    # Raw-пароль никогда не попадает в событие.
-    assert "wrong" not in str(failed)
-
-
-def test_login_unknown_username_401_generic_message(client: TestClient) -> None:
-    resp = client.post("/api/v1/auth/login", json={"username": "ghost", "password": "whatever"})
-    assert resp.status_code == 401
-    # НЕ раскрываем, что юзера не существует — то же сообщение, что и wrong-password.
-    body_ghost = resp.json()["detail"]
-
-    resp2 = client.post("/api/v1/auth/login", json={"username": "ghost2", "password": "x"})
-    assert resp2.json()["detail"] == body_ghost
-
-
-def test_login_inactive_user_401(client: TestClient, store: _Store) -> None:
-    store.add_user("bob", hash_password("Secret123!"), role="employee", is_active=False)
-    resp = client.post("/api/v1/auth/login", json={"username": "bob", "password": "Secret123!"})
-    assert resp.status_code == 401
-
-
-def test_login_null_password_hash_401(client: TestClient, store: _Store) -> None:
-    store.add_user("carol", None, role="employee")
-    resp = client.post("/api/v1/auth/login", json={"username": "carol", "password": "anything"})
-    assert resp.status_code == 401
-
-
-def test_login_always_calls_verify_password_timing_oracle_guard(
-    client: TestClient, store: _Store, monkeypatch: pytest.MonkeyPatch
-) -> None:
-    """#2552 post-review Medium 2: `verify_password` должен выполняться ровно
-    один раз на КАЖДУЮ попытку логина — включая неизвестный username и NULL
-    password_hash — иначе короткое замыкание даёт наблюдаемый timing-oracle
-    для user-enumeration. Тест не измеряет тайминг (флейки в CI), а проверяет
-    сам факт + аргумент вызова через monkeypatch-счётчик."""
-    store.add_user("alice", hash_password("Secret123!"), role="employee")
-    store.add_user("nullhash", None, role="employee")
-
-    calls: list[str] = []
-    real_verify = auth_router.verify_password
-
-    def _counting_verify(plain: str, hashed: str) -> bool:
-        calls.append(hashed)
-        return real_verify(plain, hashed)
-
-    monkeypatch.setattr(auth_router, "verify_password", _counting_verify)
-
-    resp_unknown = client.post("/api/v1/auth/login", json={"username": "ghost", "password": "x"})
-    assert resp_unknown.status_code == 401
-
-    resp_null_hash = client.post(
-        "/api/v1/auth/login", json={"username": "nullhash", "password": "x"}
-    )
-    assert resp_null_hash.status_code == 401
-
-    resp_wrong_pw = client.post(
-        "/api/v1/auth/login", json={"username": "alice", "password": "wrong"}
-    )
-    assert resp_wrong_pw.status_code == 401
-
-    assert len(calls) == 3
-    # Unknown user / NULL hash — сверяется против dummy-хеша, не против NULL.
-    assert calls[0] == auth_router._DUMMY_PASSWORD_HASH
-    assert calls[1] == auth_router._DUMMY_PASSWORD_HASH
-    # Реальный юзер с реальным hash — НЕ dummy.
-    assert calls[2] != auth_router._DUMMY_PASSWORD_HASH
-
-
-def test_login_rate_limit_429(client: TestClient, store: _Store) -> None:
-    store.add_user("dave", hash_password("Secret123!"), role="employee")
-    limit = config.settings.login_rate_limit
-
-    for _ in range(limit):
-        resp = client.post("/api/v1/auth/login", json={"username": "dave", "password": "wrong"})
-        assert resp.status_code == 401
-
-    resp = client.post("/api/v1/auth/login", json={"username": "dave", "password": "wrong"})
-    assert resp.status_code == 429
-    assert "Retry-After" in resp.headers
-
-
-# ---------------------------------------------------------------------------
-# POST /logout
-# ---------------------------------------------------------------------------
-
-
-def test_logout_revokes_session_and_clears_cookie(client: TestClient, store: _Store) -> None:
-    store.add_user("alice", hash_password("Secret123!"), role="employee")
-    login_resp = client.post(
-        "/api/v1/auth/login", json={"username": "alice", "password": "Secret123!"}
-    )
-    token = login_resp.cookies[config.settings.session_cookie_name]
-    assert token in store.sessions
-
-    logout_resp = client.post("/api/v1/auth/logout")
-    assert logout_resp.status_code == 200
-    assert logout_resp.json() == {"ok": True}
-    assert token not in store.sessions
-
-
-# ---------------------------------------------------------------------------
-# GET /me — session-first + dual-mode legacy fallback
-# ---------------------------------------------------------------------------
-
-
-def test_me_with_session_cookie_returns_db_role(client: TestClient, store: _Store) -> None:
-    store.add_user(
-        "alice",
-        hash_password("Secret123!"),
-        role="employee",
-        display_name="Алиса",
-        org_name="ООО Ромашка",
-        email="alice@romashka.ru",
-    )
-    client.post("/api/v1/auth/login", json={"username": "alice", "password": "Secret123!"})
-
-    resp = client.get("/api/v1/me")
-    assert resp.status_code == 200, resp.text
-    body = resp.json()
-    assert body["username"] == "alice"
-    assert body["role"] == "employee"
-    assert "/trade-in/**" in body["allowed_paths"]
-    assert "/admin/**" in body["deny_paths"]
-    assert body["display_name"] == "Алиса"
-    assert body["org"] == "ООО Ромашка"
-    assert body["email"] == "alice@romashka.ru"
-
-
-def test_me_manager_role_gets_team_path(client: TestClient, store: _Store) -> None:
-    store.add_user("mgr", hash_password("Secret123!"), role="manager")
-    client.post("/api/v1/auth/login", json={"username": "mgr", "password": "Secret123!"})
-
-    resp = client.get("/api/v1/me")
-    assert resp.status_code == 200
-    assert "/api/v1/team/**" in resp.json()["allowed_paths"]
-
-
-def test_me_without_cookie_dual_mode_legacy_still_works(client: TestClient) -> None:
-    """Regression guard: без сессии, auth_mode=dual — legacy X-Authenticated-User
-    путь через roles.yaml работает БЕЗ ИЗМЕНЕНИЙ."""
-    resp = client.get("/api/v1/me", headers={"X-Authenticated-User": "admin"})
-    assert resp.status_code == 200, resp.text
-    body = resp.json()
-    assert body["username"] == "admin"
-    assert body["role"] == "admin"
-    assert body["allowed_paths"] == ["/**"]
-
-
-def test_me_no_cookie_no_header_401(client: TestClient) -> None:
-    resp = client.get("/api/v1/me")
-    assert resp.status_code == 401
-
-
-def test_expired_session_falls_back_to_no_auth(client: TestClient, store: _Store) -> None:
-    """Истёкшая сессия трактуется как отсутствие cookie: без legacy-заголовка — 401."""
-    uid = store.add_user("alice", hash_password("Secret123!"), role="employee")
-    store.add_expired_session("expired-tok", uid)
-
-    client.cookies.set(config.settings.session_cookie_name, "expired-tok")
-    resp = client.get("/api/v1/trade-in/dummy")
-    assert resp.status_code == 401
-
-
-def test_expired_session_falls_back_to_legacy_header_in_dual_mode(
-    client: TestClient, store: _Store
-) -> None:
-    """Истёкшая сессия + валидный legacy header в dual-mode — header отрабатывает."""
-    uid = store.add_user("alice", hash_password("Secret123!"), role="employee")
-    store.add_expired_session("expired-tok", uid)
-
-    client.cookies.set(config.settings.session_cookie_name, "expired-tok")
-    resp = client.get(
-        "/api/v1/trade-in/dummy",
-        headers={"X-Authenticated-User": "kopylov"},
-    )
-    assert resp.status_code == 200
-    assert resp.json() == {"ok": True}
-
-
-# ---------------------------------------------------------------------------
-# rbac_guard dual vs db_only
-# ---------------------------------------------------------------------------
-
-
-def test_db_only_mode_rejects_legacy_header_without_session(
-    client: TestClient, monkeypatch: pytest.MonkeyPatch
-) -> None:
-    monkeypatch.setattr(config.settings, "auth_mode", "db_only")
-
-    resp = client.get(
-        "/api/v1/trade-in/dummy",
-        headers={"X-Authenticated-User": "admin"},
-    )
-    assert resp.status_code == 401
-    assert "session" in resp.json()["detail"].lower()
-
-
-def test_db_only_mode_accepts_valid_session(
-    client: TestClient, store: _Store, monkeypatch: pytest.MonkeyPatch
-) -> None:
-    store.add_user("alice", hash_password("Secret123!"), role="employee")
-    client.post("/api/v1/auth/login", json={"username": "alice", "password": "Secret123!"})
-
-    monkeypatch.setattr(config.settings, "auth_mode", "db_only")
-
-    resp = client.get("/api/v1/trade-in/dummy")
-    assert resp.status_code == 200
-    assert resp.json() == {"ok": True}
-
-
-def test_session_user_can_reach_tradein_but_not_admin(client: TestClient, store: _Store) -> None:
-    store.add_user("alice", hash_password("Secret123!"), role="employee")
-    client.post("/api/v1/auth/login", json={"username": "alice", "password": "Secret123!"})
-
-    ok = client.get("/api/v1/trade-in/dummy")
-    assert ok.status_code == 200
-
-    denied = client.get("/api/v1/admin/dummy")
-    # rbac_guard's admin-gate matches the path regex BEFORE routing even happens
-    # (route isn't registered on this test app) — role=employee != admin -> 403,
-    # never a 404 (a bare "any non-2xx" assertion would mask a rbac_guard typo).
-    assert denied.status_code == 403
-
-
-# ---------------------------------------------------------------------------
-# #2552 post-review CRITICAL fix: session identity must win over a spoofed
-# client-sent X-Authenticated-User header (was a skip-if-present bug — the
-# forged header used to override the session for every downstream reader of
-# the raw header: _assert_estimate_access*, account_quota, /trade-in/history,
-# support.py — in BOTH auth_mode=dual and db_only).
-# ---------------------------------------------------------------------------
-
-
-def test_session_identity_wins_over_spoofed_header_dual_mode(
-    client: TestClient, store: _Store
-) -> None:
-    store.add_user("alice", hash_password("Secret123!"), role="employee")
-    store.add_user("victim", hash_password("Secret123!"), role="employee")
-    client.post("/api/v1/auth/login", json={"username": "alice", "password": "Secret123!"})
-
-    resp = client.get(
-        "/api/v1/trade-in/whoami",
-        headers={"X-Authenticated-User": "victim"},
-    )
-    assert resp.status_code == 200
-    assert resp.json()["user"] == "alice"
-
-
-def test_session_identity_wins_over_spoofed_header_db_only_mode(
-    client: TestClient, store: _Store, monkeypatch: pytest.MonkeyPatch
-) -> None:
-    store.add_user("alice", hash_password("Secret123!"), role="employee")
-    store.add_user("victim", hash_password("Secret123!"), role="employee")
-    client.post("/api/v1/auth/login", json={"username": "alice", "password": "Secret123!"})
-
-    monkeypatch.setattr(config.settings, "auth_mode", "db_only")
-
-    resp = client.get(
-        "/api/v1/trade-in/whoami",
-        headers={"X-Authenticated-User": "victim"},
-    )
-    assert resp.status_code == 200
-    assert resp.json()["user"] == "alice"
-
-
-def test_cyrillic_username_session_propagation_does_not_500(
-    client: TestClient, store: _Store
-) -> None:
-    """#2552 post-review Medium 1: `.encode("latin-1")` без errors="replace" на
-    кириллическом username крашил бы КАЖДЫЙ запрос такого юзера с 500."""
-    store.add_user("алиса", hash_password("Secret123!"), role="employee")
-    login_resp = client.post(
-        "/api/v1/auth/login", json={"username": "алиса", "password": "Secret123!"}
-    )
-    assert login_resp.status_code == 200, login_resp.text
-
-    resp = client.get("/api/v1/trade-in/whoami")
-    assert resp.status_code == 200, resp.text
-    # latin-1 "replace" гарантированно не крашит — точное значение (что именно
-    # получится из non-latin1 байт) не является контрактом, важно отсутствие 500.
-    assert resp.json()["user"] is not None

tradein-mvp/backend/tests/test_auth_session.py

@@ -1,306 +0,0 @@
-"""Tests for app.services.auth_session — session CRUD + DB-role scope mapping (#2552).
-
-Coverage:
-  - create_session: INSERT with CAST(...) (never `:x::type`), commit, unique tokens.
-  - get_session_user: valid/expired/inactive/missing-row + sliding refresh (only when
-    last_seen_at is stale, best-effort — a refresh failure still returns the user).
-  - get_user_by_username: found/not-found.
-  - revoke_session / revoke_user_sessions: DELETE + commit.
-  - get_db_role_scope: employee/manager/admin/unknown mapping.
-
-All functions here take `db: Session` as a plain argument (no SessionLocal() opened
-internally) — unit tests just pass a hand-rolled fake, mirroring the `_FakeSession`
-pattern from tests/test_user_events.py but adapted for `.fetchone()`-based reads.
-"""
-
-from __future__ import annotations
-
-import os
-import re
-from datetime import UTC, datetime, timedelta
-from types import SimpleNamespace
-from typing import Any
-
-os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
-
-from app.services import auth_session as svc
-
-# ---------------------------------------------------------------------------
-# Fake DB session
-# ---------------------------------------------------------------------------
-
-
-class _FakeDB:
-    """Minimal `Session` stand-in: queued `.fetchone()` results per `execute()` call,
-    in call order. `execute()` beyond the queue returns a result with `fetchone()
-    -> None`. Set `.raise_on_call = N` to make the Nth `execute()` (1-indexed) raise.
-    """
-
-    def __init__(self, rows: list[Any] | None = None) -> None:
-        self._rows: list[Any] = list(rows or [])
-        self.executed: list[tuple[str, dict[str, Any] | None]] = []
-        self.committed = 0
-        self.rolled_back = 0
-        self.raise_on_call: int | None = None
-
-    def execute(self, stmt: object, params: dict[str, Any] | None = None) -> SimpleNamespace:
-        call_no = len(self.executed) + 1
-        self.executed.append((str(stmt), params))
-        if self.raise_on_call == call_no:
-            raise RuntimeError("simulated DB failure")
-        row = self._rows.pop(0) if self._rows else None
-        return SimpleNamespace(fetchone=lambda: row)
-
-    def commit(self) -> None:
-        self.committed += 1
-
-    def rollback(self) -> None:
-        self.rolled_back += 1
-
-
-def _session_row(
-    *,
-    user_id: int = 1,
-    expires_at: datetime | None = None,
-    last_seen_at: datetime | None = None,
-    username: str = "alice",
-    role: str = "employee",
-    is_active: bool = True,
-) -> SimpleNamespace:
-    now = datetime.now(UTC)
-    return SimpleNamespace(
-        user_id=user_id,
-        expires_at=expires_at if expires_at is not None else now + timedelta(hours=1),
-        last_seen_at=last_seen_at if last_seen_at is not None else now,
-        username=username,
-        role=role,
-        display_name="Alice A.",
-        org_name="Org LLC",
-        email="alice@example.com",
-        is_active=is_active,
-    )
-
-
-def _user_row(
-    *,
-    user_id: int = 1,
-    username: str = "alice",
-    password_hash: str | None = "hash",
-    role: str = "employee",
-    is_active: bool = True,
-) -> SimpleNamespace:
-    return SimpleNamespace(
-        id=user_id,
-        username=username,
-        password_hash=password_hash,
-        role=role,
-        is_active=is_active,
-        display_name="Alice A.",
-        org_name="Org LLC",
-        email="alice@example.com",
-    )
-
-
-# ---------------------------------------------------------------------------
-# create_session
-# ---------------------------------------------------------------------------
-
-
-def test_create_session_inserts_and_commits() -> None:
-    db = _FakeDB()
-    token = svc.create_session(db, user_id=42, ip="1.2.3.4", user_agent="pytest")
-
-    assert db.committed == 1
-    assert len(db.executed) == 1
-    sql, params = db.executed[0]
-    assert "INSERT INTO tradein_sessions" in sql
-    assert params is not None
-    assert params["user_id"] == 42
-    assert params["ip"] == "1.2.3.4"
-    assert params["user_agent"] == "pytest"
-    assert params["token"] == token
-    assert isinstance(token, str)
-    assert len(token) >= 32
-
-
-def test_create_session_cast_not_doublecolon() -> None:
-    db = _FakeDB()
-    svc.create_session(db, user_id=1)
-    sql, _ = db.executed[0]
-    assert not re.search(r":\w+::\w", sql)
-    assert "CAST(:ttl_hours AS integer)" in sql
-    assert "CAST(:ip AS inet)" in sql
-
-
-def test_create_session_tokens_are_unique() -> None:
-    db = _FakeDB()
-    t1 = svc.create_session(db, user_id=1)
-    t2 = svc.create_session(db, user_id=1)
-    assert t1 != t2
-
-
-# ---------------------------------------------------------------------------
-# get_session_user
-# ---------------------------------------------------------------------------
-
-
-def test_get_session_user_no_token_returns_none() -> None:
-    db = _FakeDB()
-    assert svc.get_session_user(db, "") is None
-    assert db.executed == []
-
-
-def test_get_session_user_missing_row_returns_none() -> None:
-    db = _FakeDB(rows=[None])
-    assert svc.get_session_user(db, "tok") is None
-    sql, params = db.executed[0]
-    assert "FROM tradein_sessions s" in sql
-    assert "JOIN tradein_users u" in sql
-    assert params == {"token": "tok"}
-
-
-def test_get_session_user_expired_returns_none() -> None:
-    now = datetime.now(UTC)
-    db = _FakeDB(rows=[_session_row(expires_at=now - timedelta(minutes=1))])
-    assert svc.get_session_user(db, "tok") is None
-    # Никакого sliding-refresh UPDATE для невалидной сессии.
-    assert len(db.executed) == 1
-
-
-def test_get_session_user_inactive_returns_none() -> None:
-    db = _FakeDB(rows=[_session_row(is_active=False)])
-    assert svc.get_session_user(db, "tok") is None
-    assert len(db.executed) == 1
-
-
-def test_get_session_user_valid_recent_no_refresh() -> None:
-    """last_seen_at свежий (<5 мин) — sliding refresh НЕ триггерится."""
-    now = datetime.now(UTC)
-    db = _FakeDB(rows=[_session_row(last_seen_at=now - timedelta(minutes=1))])
-    result = svc.get_session_user(db, "tok")
-
-    assert result is not None
-    assert result["username"] == "alice"
-    assert result["role"] == "employee"
-    assert result["display_name"] == "Alice A."
-    assert result["org_name"] == "Org LLC"
-    assert result["email"] == "alice@example.com"
-    assert result["user_id"] == 1
-    # Только 1 execute (SELECT) — никакого UPDATE.
-    assert len(db.executed) == 1
-    assert db.committed == 0
-
-
-def test_get_session_user_stale_last_seen_triggers_refresh() -> None:
-    """last_seen_at старше 5 минут — один UPDATE (sliding refresh) + commit."""
-    now = datetime.now(UTC)
-    db = _FakeDB(rows=[_session_row(last_seen_at=now - timedelta(minutes=10))])
-    result = svc.get_session_user(db, "tok")
-
-    assert result is not None
-    assert len(db.executed) == 2
-    update_sql, update_params = db.executed[1]
-    assert "UPDATE tradein_sessions" in update_sql
-    assert "SET last_seen_at" in update_sql
-    assert not re.search(r":\w+::\w", update_sql)
-    assert "CAST(:ttl_hours AS integer)" in update_sql
-    assert update_params == {"ttl_hours": 720, "token": "tok"}
-    assert db.committed == 1
-
-
-def test_get_session_user_refresh_failure_is_swallowed() -> None:
-    """Sliding-refresh UPDATE падает — всё равно возвращаем валидного юзера
-    (best-effort refresh, не часть решения "валидна ли сессия")."""
-    now = datetime.now(UTC)
-    db = _FakeDB(rows=[_session_row(last_seen_at=now - timedelta(minutes=10))])
-    db.raise_on_call = 2
-
-    result = svc.get_session_user(db, "tok")
-
-    assert result is not None
-    assert result["username"] == "alice"
-    assert db.rolled_back == 1
-    assert db.committed == 0
-
-
-# ---------------------------------------------------------------------------
-# get_user_by_username
-# ---------------------------------------------------------------------------
-
-
-def test_get_user_by_username_found() -> None:
-    db = _FakeDB(rows=[_user_row()])
-    user = svc.get_user_by_username(db, "alice")
-
-    assert user is not None
-    assert user["username"] == "alice"
-    assert user["password_hash"] == "hash"
-    assert user["role"] == "employee"
-    assert user["is_active"] is True
-    sql, params = db.executed[0]
-    assert "FROM tradein_users" in sql
-    assert params == {"username": "alice"}
-
-
-def test_get_user_by_username_not_found() -> None:
-    db = _FakeDB(rows=[None])
-    assert svc.get_user_by_username(db, "ghost") is None
-
-
-# ---------------------------------------------------------------------------
-# revoke_session / revoke_user_sessions
-# ---------------------------------------------------------------------------
-
-
-def test_revoke_session_deletes_and_commits() -> None:
-    db = _FakeDB()
-    svc.revoke_session(db, "tok")
-
-    assert db.committed == 1
-    sql, params = db.executed[0]
-    assert "DELETE FROM tradein_sessions" in sql
-    assert "token" in sql
-    assert params == {"token": "tok"}
-
-
-def test_revoke_user_sessions_deletes_and_commits() -> None:
-    db = _FakeDB()
-    svc.revoke_user_sessions(db, 7)
-
-    assert db.committed == 1
-    sql, params = db.executed[0]
-    assert "DELETE FROM tradein_sessions" in sql
-    assert "user_id" in sql
-    assert params == {"user_id": 7}
-
-
-# ---------------------------------------------------------------------------
-# get_db_role_scope
-# ---------------------------------------------------------------------------
-
-
-def test_get_db_role_scope_employee_matches_legacy_pilot() -> None:
-    paths, deny = svc.get_db_role_scope("employee")
-    assert "/trade-in/**" in paths
-    assert "/trade-in/api/v1/**" in paths
-    assert "/admin/**" in deny
-    assert "/api/v1/admin/**" in deny
-
-
-def test_get_db_role_scope_manager_adds_team_path() -> None:
-    paths, deny = svc.get_db_role_scope("manager")
-    assert "/trade-in/**" in paths
-    assert "/api/v1/team/**" in paths
-    assert "/admin/**" in deny
-
-
-def test_get_db_role_scope_admin_full_access() -> None:
-    paths, deny = svc.get_db_role_scope("admin")
-    assert paths == ["/**"]
-    assert deny == []
-
-
-def test_get_db_role_scope_unknown_role_denies_all() -> None:
-    paths, deny = svc.get_db_role_scope("ghost")
-    assert paths == []
-    assert deny == ["/**"]