auth.py - master - ring0/support-system

auth.py
80 строк3 KB
ring0
Init project
22 июн 2026, 17:13
22 июн 2026, 17:139df53b7
import os
from datetime import datetime, timedelta

import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from passlib.context import CryptContext
from sqlalchemy.orm import Session

from db import get_db
from models import User

# Настройки JWT берутся из переменных окружения (.env)
SECRET_KEY = os.getenv("SECRET_KEY", "change-me-in-production")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_HOURS = 8  # время жизни токена — 8 часов

# Контекст хэширования паролей на основе алгоритма bcrypt
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

security = HTTPBearer()


def hash_password(password: str) -> str:
    """Вычисление хэша пароля алгоритмом bcrypt (со встроенной солью)"""
    return pwd_context.hash(password)


def verify_password(plain_password: str, password_hash: str) -> bool:
    """Сравнение введённого пароля с сохранённым хэшем"""
    return pwd_context.verify(plain_password, password_hash)


def create_access_token(user: User) -> str:
    """Генерация JWT-токена с идентификатором пользователя, ролью и сроком действия"""
    expire = datetime.utcnow() + timedelta(hours=ACCESS_TOKEN_EXPIRE_HOURS)
    payload = {
        "sub": str(user.id),
        "username": user.username,
        "role": user.role,
        "exp": expire,
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)


def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
    db: Session = Depends(get_db),
) -> User:
    """Извлечение и верификация JWT-токена, получение текущего пользователя"""
    token = credentials.credentials
    invalid = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Недействительный или просроченный токен",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id = int(payload.get("sub"))
    except (jwt.PyJWTError, TypeError, ValueError):
        raise invalid

    user = db.query(User).filter(User.id == user_id).first()
    if user is None:
        raise invalid
    return user


def require_roles(*allowed_roles: str):
    """Фабрика зависимостей для разграничения доступа по ролям пользователей"""

    def checker(current_user: User = Depends(get_current_user)) -> User:
        if current_user.role not in allowed_roles:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Недостаточно прав для выполнения операции",
            )
        return current_user

    return checker