121 lines
3.9 KiB
Python
121 lines
3.9 KiB
Python
"""Безопасность: 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)
|