blitz_api/app/utils.py

import json
import logging
import asyncio
import re
import os
from types import coroutine
from typing import Dict

import aiohttp
import grpc
import requests
from decouple import config
from fastapi.encoders import jsonable_encoder
from fastapi_plugins import redis_plugin
from starlette import status

import app.repositories.ln_impl.protos.lightning_pb2_grpc as lnrpc
import app.repositories.ln_impl.protos.router_pb2_grpc as routerrpc
import app.repositories.ln_impl.protos.walletunlocker_pb2_grpc as unlockerrpc
from app.models.bitcoind import BlockRpcFunc


class BitcoinConfig:
    def __init__(self) -> None:
        self.network = config("network")
        self.zmq_block_rpc = BlockRpcFunc.from_string(config("bitcoind_zmq_block_rpc"))

        if self.network == "testnet":
            self.ip = config("bitcoind_ip_testnet")
            self.rpc_port = config("bitcoind_port_rpc_testnet")
            self.zmq_port = config("bitcoind_zmq_block_port_testnet")
        else:
            self.ip = config("bitcoind_ip_mainnet")
            self.rpc_port = config("bitcoind_port_rpc_mainnet")
            self.zmq_port = config("bitcoind_zmq_block_port_mainnet")

        self.rpc_url = f"http://{self.ip}:{self.rpc_port}"
        self.zmq_url = f"tcp://{self.ip}:{self.zmq_port}"

        self.username = config("bitcoind_user")
        self.pw = config("bitcoind_pw")


bitcoin_config = BitcoinConfig()


class LightningConfig:
    def __init__(self) -> None:
        self.network = config("network")
        self.ln_node = config("ln_node")

        if self.ln_node == "lnd":
            # Due to updated ECDSA generated tls.cert we need to let gprc know that
            # we need to use that cipher suite otherwise there will be a handshake
            # error when we communicate with the lnd rpc server.
            os.environ["GRPC_SSL_CIPHER_SUITES"] = "HIGH+ECDSA"

            # Uncomment to see full gRPC logs
            # os.environ["GRPC_TRACE"] = "all"
            # os.environ["GRPC_VERBOSITY"] = "DEBUG"

            self.lnd_macaroon = config("lnd_macaroon")
            self._lnd_cert = bytes.fromhex(config("lnd_cert"))
            self._lnd_grpc_ip = config("lnd_grpc_ip")
            self._lnd_grpc_port = config("lnd_grpc_port")
            self._lnd_rest_port = config("lnd_rest_port")
            self._lnd_grpc_url = self._lnd_grpc_ip + ":" + self._lnd_grpc_port

            auth_creds = grpc.metadata_call_credentials(self.metadata_callback)
            ssl_creds = grpc.ssl_channel_credentials(self._lnd_cert)
            combined_creds = grpc.composite_channel_credentials(ssl_creds, auth_creds)

            self._channel = grpc.aio.secure_channel(self._lnd_grpc_url, combined_creds)
            self.lnd_stub = lnrpc.LightningStub(self._channel)
            self.router_stub = routerrpc.RouterStub(self._channel)
            self.wallet_unlocker = unlockerrpc.WalletUnlockerStub(self._channel)
        elif self.ln_node == "clightning":
            # TODO: implement c-lightning
            pass
        elif self.ln_node == "":
            # its ok to run raspiblitz also without lightning
            pass
        else:
            raise NameError(
                f'Node type "{self.ln_node}" is unknown. Use "lnd" or "clightning"'
            )

    def metadata_callback(self, context, callback):
        # for more info see grpc docs
        callback([("macaroon", self.lnd_macaroon)], None)


lightning_config = LightningConfig()


def bitcoin_rpc(method: str, params: list = []) -> requests.Response:
    """Make an RPC request to the Bitcoin daemon

    Connection parameters are read from the .env file.

    Parameters
    ----------
    method : str
        The method to call.
    params : list, optional
        Any parameters to include with the call
    """
    auth = (bitcoin_config.username, bitcoin_config.pw)
    headers = {"Content-type": "text/plain"}
    data = (
        '{"jsonrpc": "2.0", "method": "'
        + method
        + '", "id":"0", "params":'
        + json.dumps(params)
        + "}"
    )
    return requests.post(bitcoin_config.rpc_url, auth=auth, headers=headers, data=data)


async def bitcoin_rpc_async(method: str, params: list = []) -> coroutine:
    auth = aiohttp.BasicAuth(bitcoin_config.username, bitcoin_config.pw)
    headers = {"Content-type": "text/plain"}
    data = (
        '{"jsonrpc": "2.0", "method": "'
        + method
        + '", "id":"0", "params":'
        + json.dumps(params)
        + "}"
    )

    async with aiohttp.ClientSession(auth=auth, headers=headers) as session:
        async with session.post(bitcoin_config.rpc_url, data=data) as resp:
            if resp.status == status.HTTP_200_OK:
                return await resp.json()
            elif resp.status == status.HTTP_401_UNAUTHORIZED:
                return {
                    "error": "Access denied to Bitcoin Core RPC. Check if username and password is correct",
                    "status": status.HTTP_403_FORBIDDEN,
                }
            elif resp.status == status.HTTP_403_FORBIDDEN:
                return {
                    "error": "Access denied to Bitcoin Core RPC. If this is a remote node, check if 'network.rpcallowip=0.0.0.0/0' is set.",
                    "status": status.HTTP_403_FORBIDDEN,
                }
            else:
                return {
                    "error": f"Unknown answer from Bitcoin Core. Reason: {resp.reason}",
                    "status": resp.status,
                }


async def send_sse_message(id: str, json_data: Dict):
    """Send a message to any SSE connections

    Parameters
    ----------
    id : str
        ID String von SSE class
    data : list, optional
        The data to include
    """

    await redis_plugin.redis.publish_json(
        "default", {"event": id, "data": json.dumps(jsonable_encoder(json_data))}
    )


async def redis_get(key: str) -> str:
    v = await redis_plugin.redis.get(key)

    if not v:
        logging.warning(f"Key '{key}' not found in Redis DB.")
        return ""

    return v.decode("utf-8")


# TODO
# idea is to have a second redis channel called system, that the API subscribes to. If for example
# the 'state' value gets changed by the _cache.sh script, it should publish this to this channel
# so the API can forward the change to thru the SSE to the WebUI


class SSE:
    SYSTEM_INFO = "system_info"
    SYSTEM_SHUTDOWN_NOTICE = "system_shutdown_initiated"
    SYSTEM_SHUTDOWN_ERROR = "system_shutdown_error"
    SYSTEM_REBOOT_NOTICE = "system_reboot_initiated"
    SYSTEM_REBOOT_ERROR = "system_reboot_error"
    HARDWARE_INFO = "hardware_info"

    INSTALL_APP = "install"
    INSTALLED_APP_STATUS = "installed_app_status"

    BTC_NETWORK_STATUS = "btc_network_status"
    BTC_MEMPOOL_STATUS = "btc_mempool_status"
    BTC_NEW_BLOC = "btc_new_bloc"
    BTC_INFO = "btc_info"

    LN_INFO = "ln_info"
    LN_INFO_LITE = "ln_info_lite"
    LN_INVOICE_STATUS = "ln_invoice_status"
    LN_PAYMENT_STATUS = "ln_payment_status"
    LN_ONCHAIN_PAYMENT_STATUS = "ln_onchain_payment_status"
    LN_FEE_REVENUE = "ln_fee_revenue"
    LN_FORWARD_SUCCESSES = "ln_forward_successes"
    WALLET_BALANCE = "wallet_balance"
    WALLET_LOCK_STATUS = "wallet_lock_status"


async def call_script(scriptPath) -> str:
    cmd = f"bash {scriptPath}"
    proc = await asyncio.create_subprocess_shell(
        cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    stdout, stderr = await proc.communicate()
    if stdout:
        return stdout.decode()
    if stderr:
        logging.error(stderr.decode())
    return ""


def parse_key_value_lines(lines: list) -> dict:
    Dict = {}
    for line in lines:
        line=line.strip()
        if len(line) == 0:
            continue
        if not re.match("^[a-zA-Z0-9]*=", line):
            continue
        key, value = line.strip().split("=", 1)
        Dict[key] = value.strip('"').strip("'")
    return Dict


def parse_key_value_text(text: str) -> dict:
    return parse_key_value_lines(text.splitlines())