Merge pull request 'feat(tradein/ui): login-форма + session-aware RouteGuard (#2555)' (#2562) from feat/tradein-login-ui into main

PR #2562 review, 3 однострочника:
1. sanitizeNext обходился: WHATWG URL-парсер (router.push) вырезает ASCII
   tab/CR/LF из ВСЕЙ строки перед парсингом, так что "/\t//evil" проходил
   regex (позиция 1 — таб, не "/"/"\\"), а после навигации резолвился в
   protocol-relative "//evil" → чужой origin. Теперь сначала strip
   [\t\r\n], потом валидация — regex видит ту же строку, что увидит парсер.
2. next=/login (или /login?...) кидал юзера обратно на форму входа
   (RouteGuard не гейтит /login) — dead-end. Фолбэк на "/".
3. RouteGuard брал next= только из usePathname(), без query — сессия,
   истёкшая на deep-link (/v2?id=<uuid>), теряла отчёт после релогина.
   Добавлен window.location.search в next (effect всегда client-side).

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

@@ -0,0 +1,156 @@
+"""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,16 +7,26 @@ 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
+from typing import Annotated, Any
 
-from fastapi import APIRouter, Header, HTTPException
+from fastapi import APIRouter, Depends, Header, HTTPException, Request
+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__)
 
@@ -25,9 +35,40 @@ 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:
-    """Return the current user's RBAC scope (role + allowed/deny paths)."""
+) -> UserScope | dict[str, Any]:
+    """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,12 +7,16 @@ 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 with no DB/lifespan/scheduler side effects
-(only ``app.core.auth`` + ``app.core.config``, both side-effect-free at
-import time beyond requiring ``DATABASE_URL`` in the environment for
-``Settings()``). ``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.
+This module holds the real guard. Historically it had "no DB/lifespan/scheduler
+side effects" beyond ``app.core.auth``/``app.core.config`` (both side-effect-free
+at import time). #2552 (dual-mode DB-session auth) adds a conditional per-request
+DB round trip via ``app.core.db.SessionLocal`` — но ТОЛЬКО когда запрос реально
+несёт session-cookie (``request.cookies.get(settings.session_cookie_name)``);
+без cookie (весь существующий тестовый трафик, legacy Caddy trusted-header
+запросы) ветка не выполняется — ноль новых 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
@@ -21,12 +25,15 @@ 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__)
 
@@ -40,7 +47,22 @@ logger = logging.getLogger(__name__)
 # Public paths без auth (/health, /docs, /openapi.json) пропускаем —
 # X-Authenticated-User там не приходит из Caddy.
 _ADMIN_API_RE = re.compile(r"^/api/v1/admin/")
-_PUBLIC_PATHS = frozenset({"/health", "/docs", "/redoc", "/openapi.json"})
+# #2552: /api/v1/auth/login + /logout — по определению вызываются ДО того, как
+# клиент аутентифицирован (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-проверки восстанавливаем внешний путь.
@@ -51,6 +73,77 @@ _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]],
@@ -59,40 +152,74 @@ async def rbac_guard(
     if path in _PUBLIC_PATHS:
         return await call_next(request)
 
-    username = request.headers.get("X-Authenticated-User")
-    if not username:
-        return JSONResponse(
-            status_code=401,
-            content={"detail": "no authenticated user (Caddy basic_auth required)"},
-        )
+    username: str | None = None
+    role: str | None = None
+    from_session = False
 
-    # #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,
-            )
+    # #2552: session-cookie резолвится ПЕРВЫМ. Если cookie нет вообще —
+    # request.cookies.get() возвращает None без единого похода в БД (ноль
+    # side-effects для всего существующего трафика без cookie).
+    token = request.cookies.get(settings.session_cookie_name)
+    if token:
+        session_user: dict[str, Any] | None = None
+        try:
+            with SessionLocal() as db:
+                session_user = get_session_user(db, token)
+        except Exception:
+            logger.exception("RBAC: session lookup failed for %s", path)
+        if session_user is not None:
+            username = session_user["username"]
+            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": "invalid or missing internal auth secret"},
+                content={"detail": "valid session required"},
             )
 
-    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"},
-        )
+        # ---- legacy trusted-header path — BIT-FOR-BIT как было до #2552 ----
+        username = request.headers.get("X-Authenticated-User")
+        if not username:
+            return JSONResponse(
+                status_code=401,
+                content={"detail": "no authenticated user (Caddy basic_auth required)"},
+            )
+
+        # #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)
@@ -101,25 +228,24 @@ async def rbac_guard(
             content={"detail": "admin only"},
         )
 
-    # #R2-H3: энфорсим roles.yaml scope (paths/deny) для ВСЕХ non-admin путей, а не
-    # только /admin/*. Иначе revoked (role=expired, paths:[] deny:/**) или узко-
-    # скоупленный аккаунт достаёт non-admin API (напр. POST /api/v1/search —
-    # экспорт листингов), который roles.yaml ему запрещает. Bootstrap-пути (/me,
-    # /brand) исключены выше по списку. roles.yaml globs внешние → восстанавливаем
-    # внешний путь (Caddy срезал /trade-in). На сбой парса — fail-open + громкий
-    # лог: не лочим платящего pilot из-за конфиг-бага (admin-гейт выше остаётся).
+    # #R2-H3: энфорсим scope (paths/deny) для ВСЕХ non-admin путей, а не
+    # только /admin/*. Bootstrap-пути (/me, /brand) исключены — иначе revoked/
+    # scope-narrowed юзер не смог бы получить свою роль вовсе.
     if not path.startswith(_RBAC_BOOTSTRAP_EXEMPT):
         external_path = _EXTERNAL_PREFIX + path
-        try:
-            allowed = is_path_allowed(role, external_path)
-        except Exception:
-            logger.exception(
-                "RBAC scope-check raised for %s %s (ext=%s) — fail-open",
-                username,
-                path,
-                external_path,
-            )
-            allowed = True
+        if from_session:
+            allowed = _db_role_path_allowed(role, external_path)
+        else:
+            try:
+                allowed = is_path_allowed(role, external_path)
+            except Exception:
+                logger.exception(
+                    "RBAC scope-check raised for %s %s (ext=%s) — fail-open",
+                    username,
+                    path,
+                    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,6 +23,7 @@ from sentry_sdk.integrations.starlette import StarletteIntegration
 from app.api.v1 import (
     admin,
     audit,
+    auth,
     brand,
     buildings,
     geocode,
@@ -106,6 +107,19 @@ 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:
@@ -158,6 +172,7 @@ 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

@@ -0,0 +1,221 @@
+"""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

@@ -0,0 +1,88 @@
+-- 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

@@ -0,0 +1,536 @@
+"""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

@@ -0,0 +1,306 @@
+"""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 == ["/**"]

tradein-mvp/frontend/src/app/login/page.tsx

@@ -0,0 +1,288 @@
+"use client";
+
+/**
+ * #2555 (эпик #2549) — login-форма для новой DB-backed session auth
+ * (POST /api/v1/auth/login, см. `tradein-mvp/backend/app/api/v1/auth.py`).
+ *
+ * Не гейтится RouteGuard'ом (см. `components/auth/RouteGuard.tsx` —
+ * `isLoginPage` bypass) — иначе редирект-петля: 401 от /me на /login тоже
+ * пытался бы увести на /login.
+ *
+ * `next=` читаем вручную из `window.location.search` (SSR-guard), а НЕ
+ * через `next/navigation` `useSearchParams()` — тот форсит Suspense boundary
+ * и ломает `next build` (см. `app/v2/page.tsx: readUrlId` — тот же паттерн,
+ * уже принятый в этом репо).
+ */
+
+import { useState } from "react";
+import type { CSSProperties, FormEvent } from "react";
+import { useRouter } from "next/navigation";
+import { useMutation, useQueryClient } from "@tanstack/react-query";
+
+import { apiFetch, HTTPError } from "@/lib/api";
+import { ME_QUERY_KEY } from "@/lib/useMe";
+
+interface LoginInput {
+  username: string;
+  password: string;
+}
+
+async function loginRequest(input: LoginInput): Promise<void> {
+  await apiFetch<{ ok: boolean }>("/api/v1/auth/login", {
+    method: "POST",
+    body: JSON.stringify(input),
+  });
+}
+
+function readNextParam(): string | null {
+  if (typeof window === "undefined") return null;
+  return new URLSearchParams(window.location.search).get("next");
+}
+
+/**
+ * Open-redirect guard: принимаем только внутренний путь, начинающийся
+ * ровно с одного "/" — не "//host" (protocol-relative URL) и не "/\host"
+ * (браузеры местами трактуют backslash как forward slash в URL-парсинге).
+ *
+ * PR #2562 review finding 1: WHATWG URL-парсер (который `router.push`
+ * использует под капотом) убирает ВСЕ ASCII tab/CR/LF из строки ПЕРЕД
+ * парсингом — так `"/\t//evil"` для наивного regex выглядит как безопасный
+ * путь с одним leading slash (символ в позиции 1 — таб, не "/" и не "\"),
+ * а после навигации превращается в `"//evil"` (protocol-relative → чужой
+ * origin). Убираем те же символы ДО валидации, чтобы regex видел ту же
+ * строку, что увидит парсер.
+ *
+ * PR #2562 review finding 2: `next=/login` (или `/login?...`) после успешного
+ * логина кидал бы юзера обратно на форму входа (RouteGuard не гейтит
+ * `/login`) — dead-end. Фолбэк на "/" в этом случае.
+ */
+function sanitizeNext(next: string | null): string {
+  if (!next) return "/";
+  const cleaned = next.replace(/[\t\r\n]/g, "");
+  if (!/^\/(?!\/|\\)/.test(cleaned)) return "/";
+  if (
+    cleaned === "/login" ||
+    cleaned.startsWith("/login?") ||
+    cleaned.startsWith("/login#")
+  ) {
+    return "/";
+  }
+  return cleaned;
+}
+
+function loginErrorMessage(error: unknown): string {
+  if (error instanceof HTTPError) {
+    if (error.status === 401) return "Неверный логин или пароль";
+    if (error.status === 429) {
+      return "Слишком много попыток. Попробуйте через несколько минут";
+    }
+  }
+  return "Не удалось войти. Проверьте подключение и попробуйте ещё раз";
+}
+
+const cardStyle: CSSProperties = {
+  background: "var(--bg-card)",
+  border: "1px solid var(--border-card)",
+  borderRadius: 12,
+  padding: "32px 28px",
+  maxWidth: 380,
+  width: "100%",
+};
+
+const labelStyle: CSSProperties = {
+  display: "block",
+  fontSize: 12,
+  fontWeight: 500,
+  color: "var(--fg-secondary)",
+  marginBottom: 6,
+};
+
+const inputStyle: CSSProperties = {
+  width: "100%",
+  boxSizing: "border-box",
+  height: 40,
+  padding: "0 12px",
+  border: "1px solid var(--border-card)",
+  borderRadius: 8,
+  fontSize: 14,
+  color: "var(--fg-primary)",
+  background: "var(--bg-card)",
+  fontFamily: "inherit",
+};
+
+export default function LoginPage() {
+  const router = useRouter();
+  const queryClient = useQueryClient();
+  const [username, setUsername] = useState("");
+  const [password, setPassword] = useState("");
+
+  const loginMutation = useMutation({
+    mutationFn: loginRequest,
+    onSuccess: () => {
+      queryClient.invalidateQueries({ queryKey: ME_QUERY_KEY });
+      router.push(sanitizeNext(readNextParam()));
+    },
+  });
+
+  function handleSubmit(e: FormEvent<HTMLFormElement>) {
+    e.preventDefault();
+    if (loginMutation.isPending) return;
+    loginMutation.mutate({ username: username.trim(), password });
+  }
+
+  return (
+    <main
+      style={{
+        minHeight: "100vh",
+        background: "var(--bg-app)",
+        display: "flex",
+        alignItems: "center",
+        justifyContent: "center",
+        padding: 24,
+        fontFamily: "Inter, -apple-system, 'Segoe UI', system-ui, sans-serif",
+      }}
+    >
+      <style>{`
+        @keyframes login-spin { to { transform: rotate(360deg); } }
+        .login-spinner { animation: login-spin .7s linear infinite; }
+        .login-input:focus-visible {
+          outline: none;
+          border-color: var(--accent);
+          box-shadow: 0 0 0 2px var(--accent-soft);
+        }
+        .login-submit:hover:not(:disabled) { background: var(--accent-hover); }
+      `}</style>
+
+      <form onSubmit={handleSubmit} style={cardStyle} aria-label="Вход в Меру">
+        <h1
+          style={{
+            margin: "0 0 8px",
+            fontSize: 22,
+            fontWeight: 600,
+            color: "var(--fg-primary)",
+            lineHeight: 1.25,
+          }}
+        >
+          Вход
+        </h1>
+        <p
+          style={{
+            margin: "0 0 24px",
+            fontSize: 14,
+            color: "var(--fg-secondary)",
+            lineHeight: 1.5,
+          }}
+        >
+          Войдите, чтобы продолжить работу с Мерой.
+        </p>
+
+        <div style={{ marginBottom: 16 }}>
+          <label style={labelStyle} htmlFor="login-username">
+            Логин
+          </label>
+          <input
+            id="login-username"
+            name="username"
+            type="text"
+            autoComplete="username"
+            required
+            autoFocus
+            className="login-input"
+            style={inputStyle}
+            value={username}
+            onChange={(e) => setUsername(e.target.value)}
+            disabled={loginMutation.isPending}
+          />
+        </div>
+
+        <div style={{ marginBottom: 24 }}>
+          <label style={labelStyle} htmlFor="login-password">
+            Пароль
+          </label>
+          <input
+            id="login-password"
+            name="password"
+            type="password"
+            autoComplete="current-password"
+            required
+            className="login-input"
+            style={inputStyle}
+            value={password}
+            onChange={(e) => setPassword(e.target.value)}
+            disabled={loginMutation.isPending}
+          />
+        </div>
+
+        {loginMutation.isError ? (
+          <p
+            role="alert"
+            style={{
+              margin: "0 0 16px",
+              padding: "8px 12px",
+              borderRadius: 8,
+              background: "var(--danger-soft)",
+              color: "var(--danger)",
+              fontSize: 13,
+              lineHeight: 1.4,
+            }}
+          >
+            {loginErrorMessage(loginMutation.error)}
+          </p>
+        ) : null}
+
+        <button
+          type="submit"
+          className="login-submit"
+          disabled={loginMutation.isPending}
+          style={{
+            width: "100%",
+            display: "flex",
+            alignItems: "center",
+            justifyContent: "center",
+            gap: 8,
+            background: "var(--accent)",
+            color: "#FFFFFF",
+            border: "none",
+            borderRadius: 8,
+            padding: "10px 16px",
+            fontSize: 14,
+            fontWeight: 500,
+            cursor: loginMutation.isPending ? "wait" : "pointer",
+            opacity: loginMutation.isPending ? 0.75 : 1,
+            fontFamily: "inherit",
+          }}
+        >
+          {loginMutation.isPending ? (
+            <>
+              <svg
+                className="login-spinner"
+                width={16}
+                height={16}
+                viewBox="0 0 24 24"
+                fill="none"
+                aria-hidden="true"
+              >
+                <circle
+                  cx="12"
+                  cy="12"
+                  r="9"
+                  stroke="rgba(255,255,255,0.35)"
+                  strokeWidth="3"
+                />
+                <path
+                  d="M21 12a9 9 0 0 0-9-9"
+                  stroke="#FFFFFF"
+                  strokeWidth="3"
+                  strokeLinecap="round"
+                />
+              </svg>
+              Входим…
+            </>
+          ) : (
+            "Войти"
+          )}
+        </button>
+      </form>
+    </main>
+  );
+}

tradein-mvp/frontend/src/app/v2/page.tsx

@@ -66,7 +66,7 @@ import {
 } from "@/lib/trade-in-api";
 import { useQuota } from "@/lib/useQuota";
 import { useMe } from "@/lib/useMe";
-import { logout } from "@/lib/logout";
+import { useLogout } from "@/lib/useLogout";
 
 // OUTER HUD FRAME + 4 corner brackets (design lines 31-37). Decorative,
 // non-interactive overlay drawn over the artboard gradient. The frame has
@@ -524,6 +524,10 @@ export default function TradeInV2Page() {
   // (#2046) — known profile fields fall back to username / brand ?? role / ""
   // when absent (never invented). undefined while loading → TopNav «Гость».
   const me = useMe();
+  // #2555: session-logout (POST /api/v1/auth/logout + local /me-cache
+  // invalidate + redirect на /login) — replaces the legacy basic_auth-cache
+  // -bust `logout()` for the v2 TopNav (new /login-form users).
+  const logoutMutation = useLogout();
 
   // Dashboard sub-hooks — each resolves independently; failure degrades its
   // section via the mappers (null input) rather than blanking the page.
@@ -973,7 +977,7 @@ export default function TradeInV2Page() {
                 onNavigate={setNav}
                 reports={reportsCount ?? 0}
                 user={topNavUser}
-                onLogout={logout}
+                onLogout={() => logoutMutation.mutate()}
               />
             </nav>
             <main

tradein-mvp/frontend/src/components/auth/RouteGuard.tsx

@@ -10,9 +10,17 @@
  * RBAC config (`auth/roles.yaml`) использует абсолютные пути сайта
  * (`/trade-in/**`, `/trade-in/api/v1/admin/**`), поэтому перед проверкой
  * isPathAllowed мы префиксим pathname через NEXT_PUBLIC_BASE_PATH.
+ *
+ * #2555 login redirect: `router.push()` (как и `usePathname()`) работает в
+ * пространстве путей БЕЗ basePath — Next сам префиксит basePath на навигации
+ * (см. `next.config.ts` комментарий `basePath`). Поэтому `next=` в query
+ * строится из `rawPath` (БЕЗ basePath), а не `absolutePath` — иначе
+ * `/login/page.tsx` сделал бы `router.push("/trade-in/history")`, и Next
+ * задвоил бы префикс в `/trade-in/trade-in/history`.
  */
 
-import { usePathname } from "next/navigation";
+import { useRouter, usePathname } from "next/navigation";
+import { useEffect } from "react";
 
 import { NoAccessScreen } from "@/components/auth/NoAccessScreen";
 import { HTTPError } from "@/lib/api";
@@ -29,6 +37,7 @@ interface RouteGuardProps {
 
 export function RouteGuard({ children }: RouteGuardProps) {
   const rawPath = usePathname() ?? "/";
+  const router = useRouter();
   // Абсолютный путь сайта: BASE_PATH + rawPath. Аккуратно с двойным слэшем
   // на `/`: `BASE_PATH = "/trade-in"` + `"/"` → `/trade-in/` (ок).
   const absolutePath = BASE_PATH
@@ -36,21 +45,51 @@ export function RouteGuard({ children }: RouteGuardProps) {
     : rawPath;
   const { data, isLoading, error } = useMe();
 
+  // #2555: /login сам себя не гейтит — иначе редирект-петля (401 на /me →
+  // редирект на /login → RouteGuard на /login опять видит 401 → редирект…).
+  const isLoginPage = rawPath === "/login";
+
+  // Prod-only: сессия истекла/отсутствует → уводим на логин вместо старого
+  // NoAccessScreen variant="session". Редирект — побочный эффект (нельзя
+  // router.push во время рендера), поэтому useEffect; пока он не сработал,
+  // рендерим null (см. return ниже), чтобы не мигал старый contents.
+  const shouldRedirectToLogin =
+    !isLoginPage &&
+    process.env.NODE_ENV === "production" &&
+    error instanceof HTTPError &&
+    error.status === 401;
+
+  useEffect(() => {
+    if (!shouldRedirectToLogin) return;
+    // PR #2562 review finding 3: deep-links carry их state в query (`/v2?id=
+    // <uuid>` — см. next.config.ts redirect comment про restore-by-id). Без
+    // `window.location.search` юзер, чья сессия истекла mid-session на такой
+    // ссылке, после логина попадал бы на голый `/v2` и терял отчёт. Effect
+    // — гарантированно client-side (useEffect тело никогда не бежит на SSR),
+    // поэтому `window` тут безопасен без typeof-guard.
+    const next = `${rawPath}${window.location.search}`;
+    router.push(`/login?next=${encodeURIComponent(next)}`);
+  }, [shouldRedirectToLogin, rawPath, router]);
+
   // #801: preview-страница самодостаточна (свой QueryClient с фейковым me),
   // RBAC к ней не применяем. Только под флагом — в проде по умолчанию выключено.
   if (ENABLE_PREVIEW && rawPath.startsWith("/ui-preview")) {
     return <>{children}</>;
   }
 
+  if (isLoginPage) {
+    return <>{children}</>;
+  }
+
   if (isLoading) return null;
 
   if (error instanceof HTTPError && error.status === 401) {
     // Dev without Caddy: 401 is normal, mount the app so local dev works.
-    // Prod: mounting children on 401 causes TanStack Query re-subscribe storm
-    // (each new observer on errored query triggers a refetch). Show session screen
-    // instead — prevents the subtree from mounting, kills the loop.
     if (process.env.NODE_ENV !== "production") return <>{children}</>;
-    return <NoAccessScreen variant="session" />;
+    // Prod: редирект уже запущен эффектом выше — ничего не рендерим, пока
+    // навигация не завершится (mounting children on 401 causes TanStack
+    // Query re-subscribe storm, см. историю до #2555 в git blame).
+    return null;
   }
 
   if (error instanceof HTTPError && error.status === 403) {

tradein-mvp/frontend/src/lib/useLogout.ts

@@ -0,0 +1,40 @@
+"use client";
+
+/**
+ * #2555: session-logout — POST /api/v1/auth/logout (revoke DB session +
+ * очистка httponly cookie tradein_session), затем чистим локальный
+ * TanStack Query /me-кэш и уходим на /login.
+ *
+ * NB: это НЕ замена legacy `@/lib/logout.ts` (Caddy basic_auth cache-bust +
+ * hard reload) — тот остаётся для страниц/пользователей на старом
+ * trusted-header механизме (см. `app.core.rbac` dual-mode resolver).
+ * useLogout — для юзеров, залогиненных через новую /login форму (#2552).
+ *
+ * Backend logout — best-effort по духу (revoke конкретной сессии), поэтому
+ * локальный logout (кэш + редирект) выполняется в `onSettled`, а не только
+ * `onSuccess`: сетевой сбой / уже-протухшая сессия не должны запирать юзера
+ * на странице без возможности разлогиниться.
+ */
+
+import { useMutation, useQueryClient } from "@tanstack/react-query";
+import { useRouter } from "next/navigation";
+
+import { apiFetch } from "@/lib/api";
+import { ME_QUERY_KEY } from "@/lib/useMe";
+
+async function logoutRequest(): Promise<void> {
+  await apiFetch<{ ok: boolean }>("/api/v1/auth/logout", { method: "POST" });
+}
+
+export function useLogout() {
+  const queryClient = useQueryClient();
+  const router = useRouter();
+
+  return useMutation({
+    mutationFn: logoutRequest,
+    onSettled: () => {
+      queryClient.invalidateQueries({ queryKey: ME_QUERY_KEY });
+      router.push("/login");
+    },
+  });
+}

tradein-mvp/frontend/src/lib/useMe.ts

@@ -14,7 +14,18 @@ import { useQuery } from "@tanstack/react-query";
 
 import { apiFetchWithStatus, HTTPError } from "@/lib/api";
 
-export type Role = "admin" | "pilot" | "expired";
+// #2555: session-auth (POST /api/v1/auth/login) вводит новые роли
+// admin|manager|employee. Legacy Caddy trusted-header роли (pilot|analyst|
+// expired) остаются — backend `/api/v1/me` может отдать любую из обеих
+// групп в зависимости от того, каким механизмом пришёл юзер (dual-mode
+// resolver, см. `tradein-mvp/backend/app/core/rbac.py`).
+export type Role =
+  | "admin"
+  | "manager"
+  | "employee"
+  | "pilot"
+  | "analyst"
+  | "expired";
 
 export interface UserScope {
   username: string;