refactor: system implementation decoupling

refs #143

app/repositories/system.py

@@ -1,11 +1,9 @@
 import asyncio
-import re
-from typing import Dict
+from typing import Dict, Optional
 
 from decouple import config
 from fastapi import HTTPException, Request, status
 
-from app.auth.auth_handler import sign_jwt
 from app.core_utils import SSE, broadcast_sse_msg
 from app.models.system import (
     APIPlatform,
@@ -17,104 +15,39 @@ from app.models.system import (
 
 PLATFORM = config("platform", default=APIPlatform.RASPIBLITZ)
 if PLATFORM == APIPlatform.RASPIBLITZ:
-    from app.core_utils import call_script, call_sudo_script, parse_key_value_text
     from app.repositories.hardware_impl.raspiblitz import (
         HW_INFO_YIELD_TIME,
         get_hardware_info_impl,
     )
-    from app.repositories.system_impl.raspiblitz import (
-        get_connection_info_impl,
-        get_system_info_impl,
-        match_password,
-        shutdown_impl,
-    )
+
+    from .system_impl.raspiblitz import RaspiBlitzSystem as SystemImpl
 elif PLATFORM == APIPlatform.NATIVE_PYTHON:
     from app.repositories.hardware_impl.native_python import (
         HW_INFO_YIELD_TIME,
         get_hardware_info_impl,
     )
-    from app.repositories.system_impl.native_python import (
-        get_connection_info_impl,
-        get_system_info_impl,
-        match_password,
-        shutdown_impl,
-    )
-else:
+
+    from .system_impl.native_python import NativePythonSystem as SystemImpl
+
+
+system = SystemImpl()
+
+if system is None:
     raise RuntimeError(f"Unknown platform {PLATFORM}")
 
 
-def password_valid(password: str):
-    # TODO: remove this once RaspiBlitz is fully refactored
-    #       into its own implementation file
-
-    if len(password) < 8:
-        return False
-    if password.find(" ") >= 0:
-        return False
-    return re.match("^[a-zA-Z0-9]*$", password)
-
-
-def name_valid(password: str):
-    if len(password) < 3:
-        return False
-    if password.find(" ") >= 0:
-        return False
-    return re.match("^[\.a-zA-Z0-9-_]*$", password)
-
-
-async def password_change(type: str, old_password: str, new_password: str):
-
-    # check just allowed type values
-    type = type.lower()
-    if not type in ["a", "b", "c"]:
-        raise HTTPException(status.HTTP_400_BAD_REQUEST, detail="unknown password type")
-
-    # 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"
-        )
-
-    if PLATFORM == APIPlatform.RASPIBLITZ:
-
-        # first check if old password is correct
-        result = await call_script(
-            f'/home/admin/config.scripts/blitz.passwords.sh check {type} "{old_password}"'
-        )
-        data = parse_key_value_text(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}"'
-        result = await call_sudo_script(script_call)
-        data = parse_key_value_text(result)
-        print(str(data))
-        if "error" in data.keys() and len(data["error"]) > 0:
-            raise HTTPException(status.HTTP_501_NOT_IMPLEMENTED, detail=data["error"])
-        return
-
-    else:
-        raise HTTPException(
-            status.HTTP_501_NOT_IMPLEMENTED,
-            detail="endpoint just works on raspiblitz so far",
-        )
+async def change_password(type: Optional[str], old_password: str, new_password: str):
+    try:
+        return await system.change_password(type, old_password, new_password)
+    except HTTPException as r:
+        raise
+    except NotImplementedError as r:
+        raise HTTPException(status.HTTP_501_NOT_IMPLEMENTED, detail=r.args[0])
 
 
 async def get_system_info() -> SystemInfo:
     try:
-        return await get_system_info_impl()
+        return await system.get_system_info()
     except HTTPException as r:
         raise
     except NotImplementedError as r:
@@ -122,11 +55,21 @@ async def get_system_info() -> SystemInfo:
 
 
 async def get_hardware_info() -> map:
-    return await get_hardware_info_impl()
+    try:
+        return await get_hardware_info_impl()
+    except HTTPException as r:
+        raise
+    except NotImplementedError as r:
+        raise HTTPException(status.HTTP_501_NOT_IMPLEMENTED, detail=r.args[0])
 
 
 async def get_connection_info() -> ConnectionInfo:
-    return await get_connection_info_impl()
+    try:
+        return await system.get_connection_info()
+    except HTTPException as r:
+        raise
+    except NotImplementedError as r:
+        raise HTTPException(status.HTTP_501_NOT_IMPLEMENTED, detail=r.args[0])
 
 
 async def shutdown(reboot: bool) -> bool:
@@ -135,7 +78,12 @@ async def shutdown(reboot: bool) -> bool:
     else:
         await broadcast_sse_msg(SSE.SYSTEM_SHUTDOWN_NOTICE, {"shutdown": True})
 
-    return await shutdown_impl(reboot=reboot)
+    try:
+        return await system.shutdown(reboot=reboot)
+    except HTTPException as r:
+        raise
+    except NotImplementedError as r:
+        raise HTTPException(status.HTTP_501_NOT_IMPLEMENTED, detail=r.args[0])
 
 
 async def subscribe_hardware_info(request: Request):
@@ -148,31 +96,12 @@ async def subscribe_hardware_info(request: Request):
 
 
 async def get_debug_logs_raw() -> 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.",
-    )
+    try:
+        return await system.get_debug_logs_raw()
+    except HTTPException as r:
+        raise
+    except NotImplementedError as r:
+        raise HTTPException(status.HTTP_501_NOT_IMPLEMENTED, detail=r.args[0])
 
 
 async def _handle_gather_hardware_info():
@@ -192,8 +121,9 @@ async def register_hardware_info_gatherer():
 
 
 async def login(i: LoginInput) -> Dict[str, str]:
-    matches = await match_password(i)
-    if matches:
-        return sign_jwt()
-
-    raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail="Password is incorrect")
+    try:
+        return await system.login(i)
+    except HTTPException as r:
+        raise
+    except NotImplementedError as r:
+        raise HTTPException(status.HTTP_501_NOT_IMPLEMENTED, detail=r.args[0])

app/repositories/system_impl/native_python.py

@@ -1,51 +1,70 @@
 import logging
 import secrets
+from typing import Dict
 
 from decouple import config
+from fastapi import HTTPException, status
 
+from app.auth.auth_handler import sign_jwt
 from app.constants import API_VERSION
-from app.models.system import APIPlatform, ConnectionInfo, LoginInput, SystemInfo
+from app.models.system import (
+    APIPlatform,
+    ConnectionInfo,
+    LoginInput,
+    RawDebugLogData,
+    SystemInfo,
+)
 from app.repositories.lightning import get_ln_info
+from app.repositories.system_impl.system_base import SystemBase
 
 
-async def get_system_info_impl() -> SystemInfo:
-    lninfo = await get_ln_info()
+class NativePythonSystem(SystemBase):
+    async def get_system_info(self) -> SystemInfo:
+        lninfo = await get_ln_info()
 
-    version = config("np_version", default="")
+        version = config("np_version", default="")
 
-    tor_api = config("np_tor_address_api_endpoint", default="")
-    tor_api_docs = config("np_tor_address_api_docs", default="")
+        tor_api = config("np_tor_address_api_endpoint", default="")
+        tor_api_docs = config("np_tor_address_api_docs", default="")
 
-    lan_api = config("np_local_address_api_endpoint", default="")
-    lan_api_docs = config("np_local_address_api_docs", default="")
+        lan_api = config("np_local_address_api_endpoint", default="")
+        lan_api_docs = config("np_local_address_api_docs", default="")
 
-    ssh_address = config("np_ssh_address", default="")
+        ssh_address = config("np_ssh_address", default="")
 
-    return SystemInfo(
-        alias=lninfo.alias,
-        color=lninfo.color,
-        platform=APIPlatform.NATIVE_PYTHON,
-        platform_version=version,
-        api_version=API_VERSION,
-        tor_web_ui=tor_api_docs,
-        tor_api=tor_api,
-        lan_web_ui=lan_api_docs,
-        lan_api=lan_api,
-        ssh_address=ssh_address,
-        chain=lninfo.chains[0].network,
-    )
+        return SystemInfo(
+            alias=lninfo.alias,
+            color=lninfo.color,
+            platform=APIPlatform.NATIVE_PYTHON,
+            platform_version=version,
+            api_version=API_VERSION,
+            tor_web_ui=tor_api_docs,
+            tor_api=tor_api,
+            lan_web_ui=lan_api_docs,
+            lan_api=lan_api,
+            ssh_address=ssh_address,
+            chain=lninfo.chains[0].network,
+        )
 
+    async def shutdown(self, reboot: bool) -> bool:
+        logging.info("Shutdown / reboot not supported in native_python mode.")
+        return False
 
-async def shutdown_impl(reboot: bool) -> bool:
-    logging.info("Shutdown / reboot not supported in native_python mode.")
-    return False
+    async def get_connection_info(self) -> ConnectionInfo:
+        # return an empty connection info object for now
+        return ConnectionInfo()
 
+    async def login(self, i: LoginInput) -> Dict[str, str]:
+        matches = secrets.compare_digest(i.password, config("login_password", cast=str))
+        if matches:
+            return sign_jwt()
 
-async def get_connection_info_impl() -> ConnectionInfo:
+        raise HTTPException(
+            status.HTTP_401_UNAUTHORIZED, detail="Password is incorrect"
+        )
 
-    # return an empty connection info object for now
-    return ConnectionInfo()
+    async def change_password(self, type: str, old_password: str, new_password: str):
+        raise NotImplementedError()
 
-
-async def match_password(i: LoginInput) -> bool:
-    return secrets.compare_digest(i.password, config("login_password", cast=str))
+    async def get_debug_logs_raw(self) -> RawDebugLogData:
+        raise NotImplementedError()

app/repositories/system_impl/raspiblitz.py

@@ -1,10 +1,12 @@
 import asyncio
 import logging
 import os
-import re
+from typing import Dict
 
 from decouple import config
+from fastapi import HTTPException, status
 
+from app.auth.auth_handler import sign_jwt
 from app.constants import API_VERSION
 from app.core_utils import (
     SSE,
@@ -14,8 +16,16 @@ from app.core_utils import (
     parse_key_value_text,
     redis_get,
 )
-from app.models.system import APIPlatform, ConnectionInfo, LoginInput, SystemInfo
+from app.models.system import (
+    APIPlatform,
+    ConnectionInfo,
+    LoginInput,
+    RawDebugLogData,
+    SystemInfo,
+)
 from app.repositories.lightning import get_ln_info
+from app.repositories.system_impl.system_base import SystemBase
+from app.repositories.utils.raspiblitz import password_valid
 
 SHELL_SCRIPT_PATH = config("shell_script_path")
 GET_DEBUG_LOG_SCRIPT = os.path.join(
@@ -23,187 +33,258 @@ GET_DEBUG_LOG_SCRIPT = os.path.join(
 )
 
 
-def _password_valid(password: str):
-    if len(password) < 8:
-        return False
-    if password.find(" ") >= 0:
-        return False
-    return re.match("^[a-zA-Z0-9]*$", password)
+class RaspiBlitzSystem(SystemBase):
+    def __init__(self) -> None:
+        self._check_shell_scripts_status()
+        super().__init__()
 
+    async def get_system_info(self) -> SystemInfo:
 
-def _check_shell_scripts_status():
-    if not os.path.exists(SHELL_SCRIPT_PATH):
-        raise Exception(f"invalid shell script path: {SHELL_SCRIPT_PATH}")
-
-    if not os.path.isfile(GET_DEBUG_LOG_SCRIPT):
-        raise Exception(f"Required file does not exist: {GET_DEBUG_LOG_SCRIPT}")
-
-
-_check_shell_scripts_status()
-
-
-async def get_system_info_impl() -> 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"),
-        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_impl(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_sse_msg(SSE.SYSTEM_REBOOT_ERROR, {"error_message": err})
+        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:
-            await broadcast_sse_msg(SSE.SYSTEM_SHUTDOWN_ERROR, {"error_message": err})
+            lninfo = await get_ln_info()
+            data_chain = lninfo.chains[0].network
+            data_alias = lninfo.alias
+            data_color = lninfo.color
 
-        return False
+        lan = await redis_get("internet_localip")
+        tor = await redis_get("tor_web_addr")
 
-    return True
-
-
-async def get_connection_info_impl() -> ConnectionInfo:
-
-    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":
-        key_value_text = await call_script(
-            "/home/admin/config.scripts/lnd.export.sh hexstring key-value"
+        return SystemInfo(
+            alias=data_alias,
+            color=data_color,
+            platform=APIPlatform.RASPIBLITZ,
+            platform_version=await redis_get("raspiBlitzVersion"),
+            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,
         )
-        key_value = parse_key_value_text(key_value_text)
-        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":
-        key_value_text = await call_script(
-            "/home/admin/config.scripts/bonus.lndconnect.sh zeus-android tor key-value"
+    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,
         )
-        key_value = parse_key_value_text(key_value_text)
-        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":
-        key_value_text = await call_sudo_script(
-            "/home/admin/config.scripts/cl.rest.sh connect mainnet key-value"
+        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_sse_msg(SSE.SYSTEM_REBOOT_ERROR, {"error_message": err})
+            else:
+                await broadcast_sse_msg(
+                    SSE.SYSTEM_SHUTDOWN_ERROR, {"error_message": err}
+                )
+
+            return False
+
+        return True
+
+    async def get_connection_info(self) -> ConnectionInfo:
+
+        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":
+            key_value_text = await call_script(
+                "/home/admin/config.scripts/lnd.export.sh hexstring key-value"
+            )
+            key_value = parse_key_value_text(key_value_text)
+            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":
+            key_value_text = await call_script(
+                "/home/admin/config.scripts/bonus.lndconnect.sh zeus-android tor key-value"
+            )
+            key_value = parse_key_value_text(key_value_text)
+            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":
+            key_value_text = await call_sudo_script(
+                "/home/admin/config.scripts/cl.rest.sh connect mainnet key-value"
+            )
+            key_value = parse_key_value_text(key_value_text)
+            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":
+            key_value_text = await call_script(
+                "/home/admin/config.scripts/lnd.export.sh btcpay key-value"
+            )
+            key_value = parse_key_value_text(key_value_text)
+            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 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,
         )
-        key_value = parse_key_value_text(key_value_text)
-        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":
-        key_value_text = await call_script(
-            "/home/admin/config.scripts/lnd.export.sh btcpay key-value"
+    async def login(self, i: LoginInput) -> Dict[str, str]:
+        matches = await self._match_password(i)
+        if matches:
+            return sign_jwt()
+
+        raise HTTPException(
+            status.HTTP_401_UNAUTHORIZED, detail="Password is incorrect"
         )
-        key_value = parse_key_value_text(key_value_text)
-        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 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 change_password(self, type: str, old_password: str, new_password: str):
 
+        # check just allowed type values
+        type = type.lower()
+        if not type in ["a", "b", "c"]:
+            raise HTTPException(
+                status.HTTP_400_BAD_REQUEST, detail=f"unknown password type: {type}"
+            )
 
-async def match_password(i: LoginInput) -> bool:
-    if _password_valid(i.password):
+        # 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 call_script(
-            f'/home/admin/config.scripts/blitz.passwords.sh check a "{i.password}"'
+            f'/home/admin/config.scripts/blitz.passwords.sh check {type} "{old_password}"'
         )
         data = parse_key_value_text(result)
-        if data["correct"] == "1":
-            return True
+        if not data["correct"] == "1":
+            raise HTTPException(
+                status.HTTP_406_NOT_ACCEPTABLE, detail="old password not correct"
+            )
 
-    return False
+        # 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}"'
+        result = await call_sudo_script(script_call)
+        data = parse_key_value_text(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) -> bool:
+        if password_valid(i.password):
+            result = await call_script(
+                f'/home/admin/config.scripts/blitz.passwords.sh check a "{i.password}"'
+            )
+            data = parse_key_value_text(result)
+            if data["correct"] == "1":
+                return True
+
+        return False
+
+    def _check_shell_scripts_status(self):
+        if not os.path.exists(SHELL_SCRIPT_PATH):
+            raise Exception(f"invalid shell script path: {SHELL_SCRIPT_PATH}")
+
+        if not os.path.isfile(GET_DEBUG_LOG_SCRIPT):
+            raise Exception(f"Required file does not exist: {GET_DEBUG_LOG_SCRIPT}")
+
+    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.",
+        )

app/repositories/system_impl/system_base.py

@@ -0,0 +1,30 @@
+from abc import abstractmethod
+from typing import Dict
+
+from app.models.system import ConnectionInfo, LoginInput, RawDebugLogData, SystemInfo
+
+
+class SystemBase:
+    @abstractmethod
+    async def get_system_info(self) -> SystemInfo:
+        raise NotImplementedError()
+
+    @abstractmethod
+    async def shutdown(self, reboot: bool) -> bool:
+        raise NotImplementedError()
+
+    @abstractmethod
+    async def get_connection_info(self) -> ConnectionInfo:
+        raise NotImplementedError()
+
+    @abstractmethod
+    async def login(self, i: LoginInput) -> Dict[str, str]:
+        raise NotImplementedError()
+
+    @abstractmethod
+    async def change_password(self, type: str, old_password: str, new_password: str):
+        raise NotImplementedError()
+
+    @abstractmethod
+    async def get_debug_logs_raw(self) -> RawDebugLogData:
+        raise NotImplementedError()

app/repositories/utils/raspiblitz.py

@@ -1,3 +1,5 @@
+import re
+
 from decouple import config
 
 SHELL_SCRIPT_PATH = config("shell_script_path")
@@ -12,3 +14,19 @@ available_app_ids = {
     "mempool",
     "thunderhub",
 }
+
+
+def password_valid(password: str):
+    if len(password) < 8:
+        return False
+    if password.find(" ") >= 0:
+        return False
+    return re.match("^[a-zA-Z0-9]*$", password)
+
+
+def name_valid(password: str):
+    if len(password) < 3:
+        return False
+    if password.find(" ") >= 0:
+        return False
+    return re.match("^[\.a-zA-Z0-9-_]*$", password)

app/routers/system.py

@@ -1,5 +1,7 @@
+from typing import Optional
+
 from fastapi import APIRouter, HTTPException, Request, status
-from fastapi.params import Depends
+from fastapi.params import Depends, Query
 
 from app.auth.auth_bearer import JWTBearer
 from app.auth.auth_handler import sign_jwt
@@ -8,12 +10,12 @@ from app.external.sse_starlette import EventSourceResponse
 from app.models.system import ConnectionInfo, LoginInput, RawDebugLogData, SystemInfo
 from app.repositories.system import (
     HW_INFO_YIELD_TIME,
+    change_password,
     get_connection_info,
     get_debug_logs_raw,
     get_hardware_info,
     get_system_info,
     login,
-    password_change,
     shutdown,
     subscribe_hardware_info,
 )
@@ -59,12 +61,19 @@ def refresh_token():
 @router.post(
     "/change-password",
     name=f"{_PREFIX}.change-password",
-    summary="Endpoint to change your password a, b or c",
+    summary="Endpoint to change your password",
     response_description="if 200 OK - password change worked",
     dependencies=[Depends(JWTBearer())],
 )
-async def change_password(type: str, old_password: str, new_password: str):
-    return await password_change(type, old_password, new_password)
+async def change_password(
+    old_password: str,
+    new_password: str,
+    type: Optional[str] = Query(
+        None,
+        description=' ℹ️ Used in **RaspiBlitz only**. Password A, B or C. Must be one of `["a", "b", "c"]`',
+    ),
+):
+    return await change_password(type, old_password, new_password)
 
 
 @router.get(