src/api/auth.py - master - azathd/mutiagent

src/api/auth.py
63 строки2 KB
Your Name
init
15 май 2026, 18:05
15 май 2026, 18:05dce7801
"""JWT auth with roles viewer / operator / admin."""

from __future__ import annotations

from datetime import UTC, datetime, timedelta
from enum import Enum
from typing import Annotated

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from jose import JWTError, jwt

from src.common.settings import settings

security = HTTPBearer(auto_error=False)


class Role(str, Enum):
    VIEWER = "viewer"
    OPERATOR = "operator"
    ADMIN = "admin"


def create_token(sub: str, role: Role, hours: int = 12) -> str:
    payload = {
        "sub": sub,
        "role": role.value,
        "exp": datetime.now(tz=UTC) + timedelta(hours=hours),
    }
    return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)


def decode_token(token: str) -> dict:
    return jwt.decode(token, settings.jwt_secret, algorithms=[settings.jwt_algorithm])


def _dev_user() -> dict:
    return {"sub": "dev", "role": Role.ADMIN.value}


async def get_current_user(
    creds: Annotated[HTTPAuthorizationCredentials | None, Depends(security)],
) -> dict:
    if settings.auth_disabled:
        return _dev_user()
    if creds is None:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing token")
    try:
        return decode_token(creds.credentials)
    except JWTError as exc:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token") from exc


def require_role(min_role: Role):
    order = {Role.VIEWER: 0, Role.OPERATOR: 1, Role.ADMIN: 2}

    async def _dep(user: Annotated[dict, Depends(get_current_user)]) -> dict:
        role = Role(user.get("role", Role.VIEWER.value))
        if order[role] < order[min_role]:
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Forbidden")
        return user

    return _dep