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

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from pydantic import BaseModel
from sqlalchemy.orm import Session

from ..config import Config
from ..db import Connection
from ..db.schemas import User, UserCapability, UserCreate
from . import _deps

router = APIRouter(prefix="/user")


class Token(BaseModel):
    access_token: str
    token_type: str


@router.post("/auth", response_model=Token)
async def login(
    form_data: OAuth2PasswordRequestForm = Depends(),
    current_config: Config | None = Depends(Config.load),
    db: Session | None = Depends(Connection.get),
):
    if current_config is None:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST)

    user = User.login(
        db=db,
        name=form_data.username,
        password=form_data.password,
        crypt_context=await current_config.crypto.crypt_context,
    )

    if user is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Could not validate credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )

    access_token = await current_config.jwt.create_token(user.name)
    return {"access_token": access_token, "token_type": "bearer"}


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


@router.post(
    "/new",
    responses={
        status.HTTP_200_OK: _deps.Responses.ok,
        status.HTTP_400_BAD_REQUEST: _deps.Responses.not_installed,
        status.HTTP_401_UNAUTHORIZED: _deps.Responses.needs_user,
        status.HTTP_403_FORBIDDEN: _deps.Responses.needs_admin,
        status.HTTP_409_CONFLICT: _deps.Responses.entry_exists,
    },
    response_model=User,
)
async def add_user(
    user: UserCreate,
    current_config: Config | None = Depends(Config.load),
    current_user: User | None = Depends(_deps.get_current_user),
    db: Session | None = Depends(Connection.get),
):
    if current_config is None:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST)

    if (current_user is None
            or UserCapability.admin not in current_user.capabilities):
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)

    new_user = User.create(
        db=db,
        user=user,
        crypt_context=await current_config.crypto.crypt_context,
    )

    if new_user is None:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT)

    return new_user