"""Безопасность: JWT, хеширование паролей.""" from datetime import datetime, timedelta, timezone from typing import Any import jwt from passlib.context import CryptContext from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer from app.config import settings pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login") def hash_password(password: str) -> str: return pwd_context.hash(password) def verify_password(plain: str, hashed: str) -> bool: return pwd_context.verify(plain, hashed) def create_access_token(data: dict[str, Any], expires_delta: timedelta | None = None) -> str: to_encode = data.copy() expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)) to_encode.update({"exp": expire}) return jwt.encode(to_encode, settings.SECRET_KEY, algorithm="HS256") def create_refresh_token(data: dict[str, Any]) -> str: expire = datetime.now(timezone.utc) + timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS) data_copy = data.copy() data_copy.update({"exp": expire, "type": "refresh"}) return jwt.encode(data_copy, settings.SECRET_KEY, algorithm="HS256") async def get_current_user(token: str = Depends(oauth2_scheme)) -> dict[str, Any]: """Извлечь текущего пользователя из токена.""" try: payload = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"]) user_id = payload.get("sub") role = payload.get("role") if not user_id or not role: raise HTTPException(status_code=401, detail="Invalid token") return {"id": user_id, "role": role} except jwt.PyJWTError: raise HTTPException(status_code=401, detail="Invalid or expired token")