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 create_mfa_token(user: User) -> str:
    """Промежуточный токен второго шага входа (изменение 037): typ="mfa", короткий срок жизни, не работает как токен доступа."""
    now = datetime.now(timezone.utc)
    exp = now + timedelta(minutes=5)
    return jwt.encode({"sub": user.username, "iat": now, "exp": exp, "pv": _password_version(user), "typ": "mfa"}, 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, "Недействительный токен")
    if payload.get("typ"):  # изменение 037: mfa_token (typ="mfa") не работает как токен доступа; токены доступа typ не имеют
        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 decode_mfa_token(token: str, db: Session) -> User:
    """Разбор и проверка mfa_token второго шага входа (изменение 037): typ == "mfa", пользователь активен,
    версия пароля совпадает (иначе токен выдан до смены пароля — второй шаг не должен переживать смену пароля)."""
    try:
        payload = jwt.decode(token, settings.jwt_secret, algorithms=["HS256"])
        username = payload["sub"]
    except (jwt.PyJWTError, KeyError):
        raise HTTPException(401, "Недействительный или истёкший токен, войдите заново")
    if payload.get("typ") != "mfa":
        raise HTTPException(401, "Недействительный токен")
    user = db.scalar(select(User).where(User.username == username, User.is_active))
    if user is None or payload.get("pv", 0) != _password_version(user):
        raise HTTPException(401, "Недействительный или истёкший токен, войдите заново")
    return user


def admin_user(user: User = Depends(current_user)) -> User:
    # изменение 032: admin_user проверяет роль != viewer (пропускает admin И superadmin)
    if user.role == Role.viewer:
        raise HTTPException(403, "Недостаточно прав")
    return user


def superadmin_user(user: User = Depends(current_user)) -> User:
    # изменение 032: строгая проверка на superadmin
    if user.role != Role.superadmin:
        raise HTTPException(403, "Недостаточно прав")
    return user