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

from fastapi import APIRouter, Depends, HTTPException, status

from ..config import Config
from ..db import Connection, schemas
from . import _deps

router = APIRouter(prefix="/admin")


@router.put(
    "/install",
    responses={
        status.HTTP_200_OK: _deps.Responses.ok,
        status.HTTP_400_BAD_REQUEST: _deps.Responses.installed,
    },
)
async def install(
    config: Config,
    admin_user: schemas.UserCreate,
    current_config: Config | None = Depends(Config.load),
):
    if current_config is not None:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST)

    await config.save()
    Connection.connect(await config.db.db_engine)

    async for db in Connection.get():
        admin_user = schemas.User.create(
            db=db,
            user=admin_user,
            crypt_context=await config.crypto.crypt_context,
        )

        admin_user.add_capabilities(
            db=db,
            capabilities=[schemas.UserCapability.admin],
        )


@router.put(
    "/config",
    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,
    },
)
async def set_config(
    new_config: Config,
    current_config: Config | None = Depends(Config.load),
    current_user: schemas.User | None = Depends(_deps.get_current_user),
):
    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)

    await new_config.save()
    Connection.connect(await new_config.db.db_engine)