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

"""
/admin endpoints.
"""

from fastapi import APIRouter, Depends, HTTPException, status
from sqlmodel import select

from ..config import Config
from ..db import Connection, TagValue, User, UserCreate
from ._common import Responses, get_current_config, get_current_user

router = APIRouter(prefix="/admin", tags=["admin"])


@router.put(
    "/install/config",
    responses={
        status.HTTP_200_OK: Responses.OK,
        status.HTTP_400_BAD_REQUEST: Responses.INSTALLED,
    },
)
async def initial_configure(
    config: Config,
    current_config: Config | None = Depends(Config.load),
):
    """
    PUT ./install/config: Configure `kiwi-vpn`.
    """

    # fail if already configured
    if current_config is not None:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST)

    # create config file, connect to database
    config.save()
    Connection.connect(config.db.uri)


@router.put(
    "/install/admin",
    responses={
        status.HTTP_200_OK: Responses.OK,
        status.HTTP_400_BAD_REQUEST: Responses.NOT_INSTALLED,
        status.HTTP_409_CONFLICT: Responses.ENTRY_EXISTS,
    },
)
async def create_initial_admin(
    admin_user: UserCreate,
    _: Config = Depends(get_current_config),
):
    """
    PUT ./install/admin: Create the first administrative user.
    """

    # fail if any user exists
    with Connection.session as db:
        if db.exec(select(User).limit(1)).first() is not None:
            raise HTTPException(status_code=status.HTTP_409_CONFLICT)

    # create an administrative user
    if (new_user := User.create(user=admin_user)) is None:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT)

    new_user.add_tags([TagValue.admin])
    new_user.update()


@router.put(
    "/config",
    responses={
        status.HTTP_200_OK: Responses.OK,
        status.HTTP_400_BAD_REQUEST: Responses.NOT_INSTALLED,
        status.HTTP_401_UNAUTHORIZED: Responses.NEEDS_USER,
        status.HTTP_403_FORBIDDEN: Responses.NEEDS_PERMISSION,
    },
)
async def set_config(
    config: Config,
    current_user: User = Depends(get_current_user),
):
    """
    PUT ./config: Edit `kiwi-vpn` main config.
    """

    # check permissions
    if not current_user.has_tag(TagValue.admin):
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)

    # update config file, reconnect to database
    config.save()
    Connection.connect(config.db.uri)