114 lines
2.6 KiB
Python
114 lines
2.6 KiB
Python
import json
|
|
from pathlib import Path
|
|
from secrets import token_hex
|
|
|
|
from fastapi import APIRouter, Depends, status
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from ..config import CRYPT_CONTEXT, DB, BaseConfig
|
|
from ..db import Certificate, DistinguishedName, User, UserCapability
|
|
|
|
router = APIRouter(prefix="/install")
|
|
|
|
|
|
CONFIG_FILE = "tmp/config.json"
|
|
|
|
|
|
async def has_config() -> bool:
|
|
return Path(CONFIG_FILE).is_file()
|
|
|
|
|
|
async def load_config() -> BaseConfig:
|
|
try:
|
|
with open(CONFIG_FILE, "r") as kv:
|
|
return BaseConfig.parse_obj(json.load(kv))
|
|
|
|
except FileNotFoundError:
|
|
return BaseConfig()
|
|
|
|
|
|
@router.get(
|
|
"/config",
|
|
response_model=BaseConfig,
|
|
responses={
|
|
status.HTTP_403_FORBIDDEN: {
|
|
"description": "Must be admin",
|
|
"content": None,
|
|
},
|
|
},
|
|
)
|
|
async def get_config(
|
|
config: BaseConfig = Depends(load_config),
|
|
has_config: bool = Depends(has_config),
|
|
):
|
|
if has_config:
|
|
return JSONResponse(status_code=status.HTTP_403_FORBIDDEN)
|
|
|
|
return config
|
|
|
|
|
|
@router.put(
|
|
"/config",
|
|
responses={
|
|
status.HTTP_200_OK: {
|
|
"content": None,
|
|
},
|
|
status.HTTP_403_FORBIDDEN: {
|
|
"description": "Must be admin",
|
|
"content": None,
|
|
},
|
|
},
|
|
)
|
|
async def set_config(
|
|
config: BaseConfig,
|
|
has_config: bool = Depends(has_config),
|
|
):
|
|
if has_config:
|
|
return JSONResponse(status_code=status.HTTP_403_FORBIDDEN)
|
|
|
|
if config.jwt.secret is None:
|
|
config.jwt.secret = token_hex(32)
|
|
|
|
with open(CONFIG_FILE, "w") as kv:
|
|
kv.write(config.json(indent=2))
|
|
|
|
|
|
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()
|