diff --git a/tradein-mvp/backend/app/api/v1/auth.py b/tradein-mvp/backend/app/api/v1/auth.py new file mode 100644 index 00000000..3c03987d --- /dev/null +++ b/tradein-mvp/backend/app/api/v1/auth.py @@ -0,0 +1,134 @@ +"""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 защита). + - Rate-limit по (username, IP) — ЖЁСТЧЕ общего `RateLimitMiddleware` + (`/api/*`), т.к. login — типичная brute-force поверхность. Использует + `SlidingWindowLimiter` (тот же примитив, что и общий rate-limit). + - Raw-пароль НИКОГДА не логируется и не попадает в user_events payload — + только username/ip/user_agent/path/method (см. schedule_event ниже). +""" + +from __future__ import annotations + +import logging +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 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, +) + +_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"{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) + credentials_ok = ( + user is not None + and user["is_active"] + and user["password_hash"] is not None + and verify_password(body.password, user["password_hash"]) + ) + + 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} diff --git a/tradein-mvp/backend/app/api/v1/me.py b/tradein-mvp/backend/app/api/v1/me.py index 4b7da093..f5e74b22 100644 --- a/tradein-mvp/backend/app/api/v1/me.py +++ b/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: ` через `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, diff --git a/tradein-mvp/backend/app/core/rbac.py b/tradein-mvp/backend/app/core/rbac.py index 0596de39..2f0aed52 100644 --- a/tradein-mvp/backend/app/core/rbac.py +++ b/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,63 @@ _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`` — только формы ``/**`` и + ``/**``, не нужна полная 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()``) видели сессионного DB-юзера так же, как + Caddy trusted-header юзера — без правок в каждом из этих мест по + отдельности (минимально инвазивный способ). + + Механизм: ``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()`` + отработал весь внутренний стек, включая эту мутацию. + + Известное ограничение: ``RateLimitMiddleware`` тоже внешний относительно + rbac_guard, но читает заголовок ДО вызова call_next() (до того, как этот + guard успевает отработать) — для ЭТОГО конкретного запроса сессионный + юзер лимитируется по IP, а не по username (per-user множитель не + применяется). Не регрессия (IP-лимит применялся бы и раньше — до + добавления session-auth такие запросы вообще были 401), просто более + строгий бюджет специфично для session-cookie-запросов; при необходимости + точного per-user квотинга для DB-юзеров — переносить резолв сессии выше + RateLimit в app/main.py отдельным issue. + """ + if request.headers.get("X-Authenticated-User"): + return + request.scope["headers"] = [ + *request.scope.get("headers", []), + (b"x-authenticated-user", username.encode("latin-1")), + ] + + async def rbac_guard( request: Request, call_next: Callable[[Request], Awaitable[Response]], @@ -59,40 +138,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 +214,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)", diff --git a/tradein-mvp/backend/app/main.py b/tradein-mvp/backend/app/main.py index 65fb5098..35ca453e 100644 --- a/tradein-mvp/backend/app/main.py +++ b/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"]) diff --git a/tradein-mvp/backend/app/services/auth_session.py b/tradein-mvp/backend/app/services/auth_session.py new file mode 100644 index 00000000..35b27c9c --- /dev/null +++ b/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, ([], ["/**"])) diff --git a/tradein-mvp/backend/tests/test_auth_api.py b/tradein-mvp/backend/tests/test_auth_api.py new file mode 100644 index 00000000..0eb828e2 --- /dev/null +++ b/tradein-mvp/backend/tests/test_auth_api.py @@ -0,0 +1,424 @@ +"""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 Any + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +import pytest +from fastapi import FastAPI +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} + + 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_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") + assert denied.status_code in (401, 403, 404) diff --git a/tradein-mvp/backend/tests/test_auth_session.py b/tradein-mvp/backend/tests/test_auth_session.py new file mode 100644 index 00000000..650186fa --- /dev/null +++ b/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 == ["/**"]