Merge pull request 'feat(tradein/auth): auth-core — login/logout, sessions, dual-mode rbac (#2552)' (#2561) from feat/tradein-auth-core into main

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,6 +152,37 @@ async def rbac_guard(
     if path in _PUBLIC_PATHS:
         return await call_next(request)
 
+    username: str | None = None
+    role: str | None = None
+    from_session = False
+
+    # #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": "valid session required"},
+            )
+
+        # ---- legacy trusted-header path — BIT-FOR-BIT как было до #2552 ----
         username = request.headers.get("X-Authenticated-User")
         if not username:
             return JSONResponse(
@@ -94,6 +218,9 @@ async def rbac_guard(
                 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)
         return JSONResponse(
@@ -101,15 +228,14 @@ 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
+        if from_session:
+            allowed = _db_role_path_allowed(role, external_path)
+        else:
             try:
                 allowed = is_path_allowed(role, external_path)
             except Exception:

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/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 == ["/**"]