ipam_control/app/security.py

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


_EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc)


def _password_version(user: User) -> int:
    """Версия пароля в токене: смена пароля меняет её, и выданные ранее токены перестают действовать (без зависимости от точности часов).
    Целочисленно (изменение 024, находка №7): без float, хотя результат и был детерминирован."""
    changed = user.password_changed_at
    return (changed - _EPOCH) // timedelta(microseconds=1) if changed else 0


def create_token(user: User) -> str:
    now = datetime.now(timezone.utc)
    exp = now + timedelta(minutes=settings.jwt_ttl_minutes)
    # iat — информационное поле (не проверяется при разборе токена); отзыв выданных токенов работает по pv (версия пароля)
    return jwt.encode({"sub": user.username, "iat": now, "exp": exp, "pv": _password_version(user)}, 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:
        payload = jwt.decode(cred.credentials, settings.jwt_secret, algorithms=["HS256"])
        username = payload["sub"]
    except (jwt.PyJWTError, KeyError):
        raise HTTPException(401, "Недействительный токен")
    user = db.scalar(select(User).where(User.username == username, User.is_active))
    if user is None:
        raise HTTPException(401, "Пользователь не найден")
    if payload.get("pv", 0) != _password_version(user):
        raise HTTPException(401, "Пароль изменён, войдите заново")  # токен выдан до смены пароля
    return user


def admin_user(user: User = Depends(current_user)) -> User:
    if user.role != Role.admin:
        raise HTTPException(403, "Недостаточно прав")
    return user