from datetime import datetime, timedelta, timezone import jwt from argon2 import PasswordHasher from argon2.exceptions import VerificationError from fastapi import Depends, HTTPException from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from sqlalchemy import select from sqlalchemy.orm import Session from app.config import settings from app.db import get_db from app.models import Role, User _ph = PasswordHasher() _bearer = HTTPBearer(auto_error=False) def hash_password(password: str) -> str: return _ph.hash(password) def verify_password(password: str, hashed: str) -> bool: try: return _ph.verify(hashed, password) except VerificationError: # несовпадение пароля или повреждённый хэш — не 500, а 401/403 return False def create_token(user: User) -> str: exp = datetime.now(timezone.utc) + timedelta(minutes=settings.jwt_ttl_minutes) return jwt.encode({"sub": user.username, "exp": exp}, settings.jwt_secret, algorithm="HS256") def current_user( cred: HTTPAuthorizationCredentials | None = Depends(_bearer), db: Session = Depends(get_db) ) -> User: if cred is None: raise HTTPException(401, "Требуется авторизация") try: username = jwt.decode(cred.credentials, settings.jwt_secret, algorithms=["HS256"])["sub"] except jwt.PyJWTError: raise HTTPException(401, "Недействительный токен") user = db.scalar(select(User).where(User.username == username, User.is_active)) if user is None: raise HTTPException(401, "Пользователь не найден") return user def admin_user(user: User = Depends(current_user)) -> User: if user.role != Role.admin: raise HTTPException(403, "Недостаточно прав") return user