blitz_api/app/system/impl/raspiblitz.py

import asyncio
import logging
import os
import time
from typing import Dict

from fastapi import HTTPException, status
from loguru import logger

from app.api.config import config
from app.api.constants import API_VERSION
from app.api.error_report.report import Report
from app.api.utils import (
    Event,
    broadcast_msg,
    exec_bash_command,
    parse_key_value_text,
    redis_get,
)
from app.auth.auth_handler import sign_jwt
from app.external.result_type.src.result.result import Err, Ok, Result
from app.lightning.service import get_ln_info
from app.system.impl.raspiblitz_utils import password_valid
from app.system.impl.system_base import SystemBase
from app.system.models import (
    APIPlatform,
    ConnectionInfo,
    LoginInput,
    RawDebugLogData,
    SystemInfo,
)

_HW_INFO_YIELD_TIME = 2

SHELL_SCRIPT_PATH = config("BAPI_RB_SHELL_SCRIPT_PATH")

if not SHELL_SCRIPT_PATH:
    logger.critical("BAPI_RB_SHELL_SCRIPT_PATH is not set or is empty.")
    exit(1)

if type(SHELL_SCRIPT_PATH) is not str:
    logger.critical("BAPI_RB_SHELL_SCRIPT_PATH is not a string.")
    exit(1)

GET_DEBUG_LOG_SCRIPT = os.path.join(
    SHELL_SCRIPT_PATH, "config.scripts", "blitz.debug.sh"
)

GET_PASSWORD_CHECK_SCRIPT = os.path.join(
    SHELL_SCRIPT_PATH, "config.scripts", "blitz.passwords.sh"
)


os.environ["TERM"] = "xterm"


def _safe_int(value: str, default: int = 0) -> int:
    try:
        return int(value)
    except (TypeError, ValueError):
        return default


def _safe_float(value: str, default: float = 0.0) -> float:
    try:
        return float(value)
    except (TypeError, ValueError):
        return default


class RaspiBlitzSystem(SystemBase):
    def __init__(self) -> None:
        self._check_shell_scripts_status()
        super().__init__()

    async def get_system_info(self) -> SystemInfo:
        lightning = await redis_get("lightning")
        if lightning == "" or lightning == "none":
            data_chain = await redis_get("chain")
            data_chain = f"{data_chain}net"
            data_alias = await redis_get("hostname")
            data_color = "#FF9900"
        else:
            lninfo = await get_ln_info()
            data_chain = lninfo.chains[0].network
            data_alias = lninfo.alias
            data_color = lninfo.color

        lan = await redis_get("internet_localip")
        tor = await redis_get("tor_web_addr")

        return SystemInfo(
            alias=data_alias,
            color=data_color,
            platform=APIPlatform.RASPIBLITZ,
            platform_version=await redis_get("raspiBlitzVersion"),
            code_version=await redis_get("codeVersion"),
            api_version=API_VERSION,
            tor_web_ui=tor,
            tor_api=f"{tor}/api",
            lan_web_ui=f"http://{lan}/",
            lan_api=f"http://{lan}/api",
            ssh_address=f"admin@{lan}",
            chain=data_chain,
        )

    async def shutdown(self, reboot: bool) -> bool:
        params = ""
        if reboot:
            params = "reboot"

        script = os.path.join(SHELL_SCRIPT_PATH, "config.scripts", "blitz.shutdown.sh")
        cmd = f"sudo bash {script} {params}"

        proc = await asyncio.create_subprocess_shell(
            cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )

        stdout, stderr = await proc.communicate()

        logging.info(f"[{cmd!r} exited with {proc.returncode}]")
        if stdout:
            logging.info(f"[stdout]\n{stdout.decode()}")
        if stderr:
            logging.error(f"[stderr]\n{stderr.decode()}")

        if proc.returncode > 0:
            err = stderr.decode()
            if reboot:
                await broadcast_msg(Event.SYSTEM_REBOOT_ERROR, {"error_message": err})
            else:
                await broadcast_msg(
                    Event.SYSTEM_SHUTDOWN_ERROR, {"error_message": err}
                )

            return False

        return True

    async def get_connection_info(self) -> Result[ConnectionInfo, Report]:
        lightning = await redis_get("lightning")

        # Bitcoin RPC
        # seems to be local network that also needs open ports
        # or tor that needs hidden service

        # LND MACAROONS & TLS
        data_lnd_rest_onion = ""
        data_lnd_admin_macaroon = ""
        data_lnd_invoice_macaroon = ""
        data_lnd_readonly_macaroon = ""
        data_lnd_tls_cert = ""

        if lightning == "lnd":
            result = await exec_bash_command(
                "/home/admin/config.scripts/lnd.export.sh hexstring key-value"
            )
            key_value = {}
            match result:
                case Ok(data):
                    key_value = parse_key_value_text(data.stdout)
                case Err(_):
                    return result

            if "adminMacaroon" in key_value.keys():
                data_lnd_admin_macaroon = key_value["adminMacaroon"]
            if "invoiceMacaroon" in key_value.keys():
                data_lnd_invoice_macaroon = key_value["invoiceMacaroon"]
            if "readonlyMacaroon" in key_value.keys():
                data_lnd_readonly_macaroon = key_value["readonlyMacaroon"]
            if "tlsCert" in key_value.keys():
                data_lnd_tls_cert = key_value["tlsCert"]
            if "restTor" in key_value.keys():
                data_lnd_rest_onion = key_value["restTor"]
            if "error" in key_value.keys():
                logging.warning(f"Error from script call: {key_value['error']}")

        # ZEUS-Wallet (LND)
        data_lnd_zeus_connection_string = ""
        if lightning == "lnd":
            result = await exec_bash_command(
                "/home/admin/config.scripts/bonus.lndconnect.sh zeus-android tor key-value"  # noqa: E501
            )
            key_value = {}
            match result:
                case Ok(data):
                    key_value = parse_key_value_text(data.stdout)
                case Err(_):
                    return result

            if "lndconnect" in key_value.keys():
                data_lnd_zeus_connection_string = key_value["lndconnect"]
            if "error" in key_value.keys():
                logging.warning(f"Error from script call: {key_value['error']}")

        # ZEUS-Wallet (Core Lightning)
        data_cl_rest_zeus_connection_string = ""
        data_cl_rest_macaroon = ""
        data_cl_rest_onion = ""
        if lightning == "cl":
            result = await exec_bash_command(
                "/home/admin/config.scripts/cl.rest.sh connect mainnet key-value"
            )
            key_value = {}
            match result:
                case Ok(data):
                    key_value = parse_key_value_text(data.stdout)
                case Err(_):
                    return result

            if "connectstring" in key_value.keys():
                data_cl_rest_zeus_connection_string = key_value["connectstring"]
            if "macaroon" in key_value.keys():
                data_cl_rest_macaroon = key_value["macaroon"]
            if "toraddress" in key_value.keys():
                data_cl_rest_onion = key_value["toraddress"]
            if "error" in key_value.keys():
                logging.warning(f"Error from script call: {key_value['error']}")

        # BTC PAY CONNECTION STRING
        data_lnd_btcpay_connection_string = ""
        if lightning == "lnd":
            result = await exec_bash_command(
                "/home/admin/config.scripts/lnd.export.sh btcpay key-value"
            )
            key_value = {}
            match result:
                case Ok(data):
                    key_value = parse_key_value_text(data.stdout)
                case Err(_):
                    return result

            if "connectionString" in key_value.keys():
                data_lnd_btcpay_connection_string = key_value["connectionString"]
            if "error" in key_value.keys():
                logging.warning(f"Error from script call: {key_value['error']}")

        return Ok(
            ConnectionInfo(
                lnd_admin_macaroon=data_lnd_admin_macaroon,
                lnd_invoice_macaroon=data_lnd_invoice_macaroon,
                lnd_readonly_macaroon=data_lnd_readonly_macaroon,
                lnd_rest_onion=data_lnd_rest_onion,
                lnd_tls_cert=data_lnd_tls_cert,
                lnd_zeus_connection_string=data_lnd_zeus_connection_string,
                lnd_btcpay_connection_string=data_lnd_btcpay_connection_string,
                cl_rest_zeus_connection_string=data_cl_rest_zeus_connection_string,
                cl_rest_macaroon=data_cl_rest_macaroon,
                cl_rest_onion=data_cl_rest_onion,
            )
        )

    async def login(self, i: LoginInput) -> Result[Dict[str, str], Report]:
        result = await self._match_password(i)
        match result:
            case Ok(matches):
                if matches:
                    return Ok(sign_jwt())
                else:
                    raise HTTPException(
                        status.HTTP_401_UNAUTHORIZED, detail="Password is incorrect"
                    )

            case Err(report):
                raise HTTPException(
                    status.HTTP_500_INTERNAL_SERVER_ERROR,
                    detail=report.format_verbose(),
                )

    async def change_password(self, type: str, old_password: str, new_password: str):
        # check just allowed type values
        if not type or type.lower() not in ["a", "b", "c"]:
            raise HTTPException(
                status.HTTP_400_BAD_REQUEST, detail=f"unknown password type: {type}"
            )
        type = type.lower()

        # check password formatting
        if not password_valid(old_password):
            raise HTTPException(
                status.HTTP_400_BAD_REQUEST, detail="old password format invalid"
            )
        if not password_valid(new_password):
            raise HTTPException(
                status.HTTP_400_BAD_REQUEST, detail="new password format invalid"
            )

        # first check if old password is correct
        result = await exec_bash_command(
            f'/home/admin/config.scripts/blitz.passwords.sh check {type} "{old_password}"',  # noqa: E501
            sensitive=True,
        )
        data = {}
        match result:
            case Ok(in_data):
                data = parse_key_value_text(in_data.stdout)
            case Err(_):
                return result

        if not data["correct"] == "1":
            raise HTTPException(
                status.HTTP_406_NOT_ACCEPTABLE, detail="old password not correct"
            )

        # second set new password
        script_call = (
            f'/home/admin/config.scripts/blitz.passwords.sh set {type} "{new_password}"'
        )
        if type == "c":
            # will set password c of both lnd & core lightning if installed/activated
            script_call = f'/home/admin/config.scripts/blitz.passwords.sh set c "{old_password}" "{new_password}"'  # noqa: E501
        result = await exec_bash_command(script_call, sensitive=True)
        data = {}
        match result:
            case Ok(in_data):
                data = parse_key_value_text(in_data.stdout)
            case Err(_):
                return result

        if "error" in data.keys() and len(data["error"]) > 0:
            raise HTTPException(status.HTTP_501_NOT_IMPLEMENTED, detail=data["error"])

        return

    async def _match_password(self, i: LoginInput) -> Result[bool, Report]:
        if password_valid(i.password):
            result = await exec_bash_command(
                f'{GET_PASSWORD_CHECK_SCRIPT} check a "{i.password}"', sensitive=True
            )

            match result:
                case Ok(data):
                    data = parse_key_value_text(data.stdout)
                    if data["correct"] == "1":
                        return Ok(True)
                case Err(_):
                    return result

        return Ok(False)

    def _check_shell_scripts_status(self):
        if not os.path.exists(SHELL_SCRIPT_PATH):
            logger.critical(f"invalid shell script path: {SHELL_SCRIPT_PATH}")
            exit(1)

        if not os.path.isfile(GET_DEBUG_LOG_SCRIPT):
            logger.critical(f"required file does not exist: {GET_DEBUG_LOG_SCRIPT}")
            exit(1)

        if not os.path.isfile(GET_PASSWORD_CHECK_SCRIPT):
            logger.critical(
                f"required file does not exist: {GET_PASSWORD_CHECK_SCRIPT}"
            )
            exit(1)

    async def get_debug_logs_raw(self) -> RawDebugLogData:
        cmd = f"bash {GET_DEBUG_LOG_SCRIPT}"
        proc = await asyncio.create_subprocess_shell(
            cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )

        stdout, stderr = await proc.communicate()

        if stderr:
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail=f"""
    f"[{cmd!r} exited with {proc.returncode}]"\n
    [stderr]\n{stderr.decode()}
            """,
            )

        if stdout:
            return RawDebugLogData(raw_data=f"[stdout]\n{stdout.decode()}")

        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"{cmd} returned no error and no output.",
        )

    async def get_hardware_info(self) -> map:
        info = {}

        # On VM setups (e.g. Proxmox) the RaspiBlitz monitor scripts may not
        # populate these keys, so redis returns empty strings. Fall back to
        # zero values instead of crashing the hardware-info gatherer.
        raw_load = await redis_get("system_cpu_load")
        loads = raw_load.split(",") if raw_load else []
        iloads = [_safe_float(load) for load in loads]
        info["cpu_overall_percent"] = (
            round(sum(iloads) / len(iloads), 2) if iloads else 0
        )
        info["cpu_per_cpu_percent"] = iloads

        info["vram_total_bytes"] = (
            _safe_int(await redis_get("system_ram_mb")) * 1000 * 1000
        )

        info["vram_available_bytes"] = (
            _safe_int(await redis_get("system_ram_available_mb")) * 1000 * 1000
        )

        info["vram_used_bytes"] = (
            info["vram_total_bytes"] - info["vram_available_bytes"]
        )
        info["vram_usage_percent"] = (
            round((100 / info["vram_total_bytes"]) * info["vram_used_bytes"], 2)
            if info["vram_total_bytes"]
            else 0
        )

        info["temperatures_celsius"] = {
            "system_temp": _safe_float(await redis_get("system_temp_celsius")),
            "coretemp": [],
        }

        now = time.time()
        boot = _safe_float(await redis_get("system_up"))
        info["boot_time_timestamp"] = now - boot

        info["networks"] = {
            "internet_online": await redis_get("internet_online"),
            "tor_web_addr": await redis_get("tor_web_addr"),
            "internet_localip": await redis_get("internet_localip"),
            "internet_localiprange": await redis_get("internet_localiprange"),
        }

        # the following is just available when setup is done
        setup_phase = await redis_get("setupPhase")
        info["disks"] = []
        if setup_phase == "done":
            res = await redis_get("hdd_capacity_bytes")
            total = int(res if len(res) > 0 else 0)
            res = await redis_get("hdd_free_bytes")
            free = int(res if len(res) > 0 else 0)
            info["disks"] = [
                {
                    "device": "/",
                    "mountpoint": "/",
                    "filesystem_type": "ext4",
                    "partition_total_bytes": total,
                    "partition_used_bytes": total - free,
                    "partition_free_bytes": free,
                    "partition_percent": (
                        0 if total == 0 else round((100 / total) * free, 2)
                    ),
                }
            ]

        return info

    def get_hardware_info_yield_time(self) -> float:
        return _HW_INFO_YIELD_TIME