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, schemas
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 = schemas.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=schemas.User)
async def get_current_user(
    current_user: schemas.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=schemas.User,
)
async def add_user(
    user: schemas.UserCreate,
    current_config: Config | None = Depends(Config.load),
    current_user: schemas.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 schemas.UserCapability.admin not in current_user.capabilities):
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)

    new_user = schemas.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