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

70 lines
1.8 KiB
Python

"""
/admin endpoints.
"""
from fastapi import APIRouter, Depends, HTTPException, status
from ..config import Config
from ..db import Capability, Connection, User, UserCreate
from ._common import Responses, get_current_user_if_admin
router = APIRouter(prefix="/admin", tags=["admin"])
@router.put(
"/install",
responses={
status.HTTP_200_OK: Responses.OK,
status.HTTP_400_BAD_REQUEST: Responses.INSTALLED,
},
)
async def install(
config: Config,
# admin_user: UserCreate,
current_config: Config | None = Depends(Config.load),
):
"""
PUT ./install: Install `kiwi-vpn`.
"""
# fail if already installed
if current_config is not None:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST)
# create config file, connect to database
await config.save()
Connection.connect("sqlite:///tmp/v2.db")
# # create an administrative user
# new_user = User.create(**admin_user)
# assert new_user is not None
# new_user.set_capabilities([Capability.login, Capability.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_ADMIN,
},
)
async def set_config(
new_config: Config,
current_config: Config | None = Depends(Config.load),
_: User | None = Depends(get_current_user_if_admin),
):
"""
PUT ./config: Edit `kiwi-vpn` main config.
"""
# fail if not installed
if current_config is None:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST)
# update config file, reconnect to database
await new_config.save()
Connection.connect("sqlite:///tmp/v2.db")