"""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, ([], ["/**"]))