"""Безопасность: bcrypt, JWT, cookie сессии и CSRF (double-submit).""" from __future__ import annotations import secrets from datetime import datetime, timedelta, timezone import bcrypt import jwt from fastapi import Response from app.core.config import settings USER_COOKIE = "fs_session" ADMIN_COOKIE = "fs_admin" CSRF_COOKIE = "csrf_token" CSRF_HEADER = "x-csrf-token" AUDIENCE_USER = "user" AUDIENCE_ADMIN = "admin" _ADMIN_PATH = "/api/admin" _USER_PATH = "/" # ─── Пароли ────────────────────────────────────────────────────────────────── def hash_password(password: str) -> str: return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8") def verify_password(password: str, password_hash: str) -> bool: try: return bcrypt.checkpw(password.encode("utf-8"), password_hash.encode("utf-8")) except (ValueError, TypeError): return False # ─── JWT ───────────────────────────────────────────────────────────────────── def create_token(subject: str | int, audience: str, ttl_minutes: int, provider: str = "") -> str: now = datetime.now(timezone.utc) payload = { "sub": str(subject), "aud": audience, "iat": int(now.timestamp()), "exp": int((now + timedelta(minutes=ttl_minutes)).timestamp()), "jti": secrets.token_hex(8), "provider": provider, } return jwt.encode(payload, settings.secret_key, algorithm=settings.jwt_algorithm) def decode_token(token: str, audience: str) -> dict: return jwt.decode( token, settings.secret_key, algorithms=[settings.jwt_algorithm], audience=audience, ) # ─── Cookie сессии + CSRF ──────────────────────────────────────────────────── def generate_csrf_token() -> str: return secrets.token_urlsafe(24) def _set_csrf_cookie(response: Response, max_age: int) -> str: csrf = generate_csrf_token() response.set_cookie( key=CSRF_COOKIE, value=csrf, max_age=max_age, httponly=False, # должен читаться JS, чтобы продублировать в заголовок secure=settings.cookie_secure, samesite="lax", path="/", domain=settings.cookie_domain_value, ) return csrf def set_user_session(response: Response, user_id: int, provider: str) -> None: ttl = settings.jwt_user_ttl_minutes token = create_token(user_id, AUDIENCE_USER, ttl, provider) response.set_cookie( key=USER_COOKIE, value=token, max_age=ttl * 60, httponly=True, secure=settings.cookie_secure, samesite="lax", path=_USER_PATH, domain=settings.cookie_domain_value, ) _set_csrf_cookie(response, ttl * 60) def set_admin_session(response: Response, admin_id: int) -> None: ttl = settings.jwt_admin_ttl_minutes token = create_token(admin_id, AUDIENCE_ADMIN, ttl, "local") response.set_cookie( key=ADMIN_COOKIE, value=token, max_age=ttl * 60, httponly=True, secure=settings.cookie_secure, samesite="lax", path=_ADMIN_PATH, domain=settings.cookie_domain_value, ) _set_csrf_cookie(response, ttl * 60) def clear_user_session(response: Response) -> None: response.delete_cookie(USER_COOKIE, path=_USER_PATH, domain=settings.cookie_domain_value) def clear_admin_session(response: Response) -> None: response.delete_cookie(ADMIN_COOKIE, path=_ADMIN_PATH, domain=settings.cookie_domain_value)