kiwi-vpn/api/kiwi_vpn_api/routers/user.py

from __future__ import annotations

from typing import Optional

from fastapi import APIRouter, Depends, HTTPException, status
from jose import JWTError, jwt
from pydantic import BaseModel

from ..config import ALGORITHM, SECRET_KEY
from ..db import User as db_User
from .auth import SCHEME

router = APIRouter(prefix="/user")


class User(BaseModel):
    name: str
    capabilities: list[str]

    @classmethod
    def from_db(cls, username: str) -> Optional[User]:
        user = db_User.get_by_name(username)

        if not user:
            return None

        return cls(
            name=user.name,
            capabilities=[cap.capability for cap in user.capabilities],
        )


async def get_current_user(token: str = Depends(SCHEME)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user = User.from_db(username)
    if user is None:
        raise credentials_exception
    return user


@router.get("/current_user/get", response_model=User)
async def get_current_user(current_user: User = Depends(get_current_user)):
    return current_user


async def is_admin(current_user: User = Depends(get_current_user)):
    return ("admin" in current_user.capabilities)


@router.get("/current_user/is_admin")
async def current_user_is_admin(is_admin: bool = Depends(is_admin)):
    return {"is_admin": is_admin}