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

import json
from pathlib import Path

from fastapi import APIRouter, Depends, status
from fastapi.responses import JSONResponse

from ..config import CONFIG_FILE, CRYPT_CONTEXT, DB, BaseConfig
from ..db import Certificate, DistinguishedName, User, UserCapability

router = APIRouter(prefix="/install")


async def get_default_config() -> BaseConfig:
    return BaseConfig()


async def get_config() -> BaseConfig:
    with open(CONFIG_FILE, "r") as kv:
        return BaseConfig.parse_obj(json.load(kv))


async def is_configured() -> bool:
    return Path(CONFIG_FILE).is_file()


@router.get("/config/get_default", response_model=BaseConfig)
async def config_get_default(config: BaseConfig = Depends(get_default_config)):
    return config


@router.get(
    "/config/get",
    response_model=BaseConfig,
    responses={
        status.HTTP_404_NOT_FOUND: {
            "description": "Not configured",
            "content": None,
        },
    },
)
async def config_get(
    config: BaseConfig = Depends(get_config),
    is_configured: bool = Depends(is_configured),
):
    if not is_configured:
        return JSONResponse(status_code=status.HTTP_404_NOT_FOUND)

    return config


@router.post(
    "/config/set",
    responses={
        status.HTTP_200_OK: {
            "content": None,
        },
        status.HTTP_403_FORBIDDEN: {
            "description": "Must be admin",
            "content": None,
        },
    },
)
async def config_set(
    config: BaseConfig,
    is_configured: bool = Depends(is_configured),
):
    if is_configured:
        return JSONResponse(status_code=status.HTTP_403_FORBIDDEN)

    config.save(CONFIG_FILE)


async def is_installed():
    return DB.table_exists(User)


@router.get("/check_installed", responses={
    status.HTTP_200_OK: {
        "model": bool,
    },
})
async def check_installed(is_installed: bool = Depends(is_installed)):
    return is_installed


@router.get(
    "/create_db",
    responses={
        status.HTTP_200_OK: {
            "content": None,
        },
        status.HTTP_400_BAD_REQUEST: {
            "description": "Could not create Database",
            "content": None,
        },
    },
)
async def create_db(is_installed: bool = Depends(is_installed)):
    if is_installed:
        return JSONResponse(
            status_code=status.HTTP_400_BAD_REQUEST,
        )

    DB.create_tables([Certificate, DistinguishedName, User, UserCapability])

    admin = User.create(name="admin", password=CRYPT_CONTEXT.hash("secret"))
    UserCapability.create(user=admin, capability="admin")

    User.create(name="johndoe", password=CRYPT_CONTEXT.hash("secret"))

    DB.close()