blitz_api/app/system/router.py

from fastapi import APIRouter, HTTPException, Response, status
from fastapi.params import Depends, Query

from app.api.utils import Event
from app.auth.auth_bearer import JWTBearer
from app.auth.auth_handler import sign_jwt
from app.system.docs import (
    get_debug_logs_raw_desc,
    get_debug_logs_raw_resp_desc,
    get_debug_logs_raw_summary,
    get_hw_info_json,
)
from app.system.models import (
    ChangePasswordInput,
    ConnectionInfo,
    LoginInput,
    RawDebugLogData,
    SystemHealthInfo,
    SystemInfo,
)
from app.system.service import (
    change_password,
    get_connection_info,
    get_debug_logs_raw,
    get_hardware_info,
    get_system_info,
    login,
    shutdown,
    system_health,
)

_PREFIX = "system"

router = APIRouter(prefix=f"/{_PREFIX}", tags=["System"])


@router.post(
    "/login",
    name=f"{_PREFIX}.login",
    summary="Logs the user in with the current password",
    response_description="JWT token for the current session.",
    status_code=status.HTTP_200_OK,
)
async def login_path(i: LoginInput, response: Response):
    try:
        token = await login(i)
        response.set_cookie("access_token", token)
        return token
    except HTTPException:
        raise
    except NotImplementedError as r:
        raise HTTPException(status.HTTP_501_NOT_IMPLEMENTED, detail=r.args[0])


@router.post(
    "/refresh-token",
    name=f"{_PREFIX}.refresh-token",
    summary="Endpoint to refresh an authentication token",
    response_description="Returns a fresh JWT token.",
    dependencies=[Depends(JWTBearer())],
)
def refresh_token():
    return sign_jwt()


@router.post(
    "/change-password",
    name=f"{_PREFIX}.change-password",
    summary="Endpoint to change your password",
    response_description="if 200 OK - password change worked",
    dependencies=[Depends(JWTBearer())],
)
async def change_password_impl(data: ChangePasswordInput):
    return await change_password(
        data.type, data.old_password, data.new_password
    )


@router.get(
    "/get-system-info",
    name=f"{_PREFIX}.get-system-info",
    summary="Get system status information",
    dependencies=[Depends(JWTBearer())],
    response_model=SystemInfo,
    responses={
        423: {"description": "Wallet is locked. Unlock via /lightning/unlock-wallet"}
    },
)
async def get_system_info_path():
    try:
        return await get_system_info()
    except HTTPException:
        raise
    except NotImplementedError as r:
        raise HTTPException(status.HTTP_501_NOT_IMPLEMENTED, detail=r.args[0])


@router.get(
    "/hardware-info",
    name=f"{_PREFIX}.hardware-info",
    summary="Get hardware status information.",
    response_description="Returns a JSON string with hardware information:\n"
    + get_hw_info_json,
    dependencies=[Depends(JWTBearer())],
    status_code=status.HTTP_200_OK,
)
async def hw_info():
    return await get_hardware_info()


@router.get(
    "/connection-info",
    name=f"{_PREFIX}.connection-info",
    summary="Get credential information to connect external apps.",
    response_description="Returns a JSON string with credential information.",
    response_model=ConnectionInfo,
    dependencies=[Depends(JWTBearer())],
    status_code=status.HTTP_200_OK,
)
async def connection_info():
    return await get_connection_info()


@router.get(
    "/get-debug-logs-raw",
    name=f"{_PREFIX}.get-debug-logs-raw",
    summary=get_debug_logs_raw_summary,
    description=get_debug_logs_raw_desc,
    response_description=get_debug_logs_raw_resp_desc,
    response_model=RawDebugLogData,
    dependencies=[Depends(JWTBearer())],
)
async def get_debug_logs_raw_route() -> RawDebugLogData:
    return await get_debug_logs_raw()


@router.get(
    "/health",
    name=f"{_PREFIX}.health",
    summary="Returns info about the system's health",
    response_model=SystemHealthInfo,
    responses={503: {"description": "One or more subsystems are not ready"}},
)
async def get_system_health(
    response: Response,
    verbose: bool = Query(
        False,
        description="If true, include a per-subsystem (api, bitcoind, lightning) health breakdown.",
    ),
) -> SystemHealthInfo:
    result = await system_health(verbose)
    if not result.healthy:
        response.status_code = status.HTTP_503_SERVICE_UNAVAILABLE
    return result


@router.post(
    "/reboot",
    name=f"{_PREFIX}.reboot",
    summary="Reboots the system",
    description=f"""Attempts to reboot the system.
    Will send a `{Event.SYSTEM_REBOOT_NOTICE}` SSE message immediately to
    all connected clients.
    """,
    response_description=f"""True if successful. False on failure.
    A failure will also send an error message with id `{Event.SYSTEM_REBOOT_ERROR}`
    to all connected clients.
    """,
    dependencies=[Depends(JWTBearer())],
)
async def reboot_system() -> bool:
    return await shutdown(reboot=True)


@router.post(
    "/shutdown",
    name=f"{_PREFIX}.shutdown",
    summary="Shuts the system down",
    description=f"""Attempts to shutdown the system.
    Will send a `{Event.SYSTEM_SHUTDOWN_NOTICE}` SSE message immediately to all
    connected clients.
    """,
    response_description=f"""True if successful. False on failure.
    A failure will also send an error message with id {Event.SYSTEM_SHUTDOWN_ERROR}
    to all connected clients.
    """,
    dependencies=[Depends(JWTBearer())],
)
async def shutdown_path() -> bool:
    return await shutdown(False)