Moved mixins back into system.py to avoid circular imports, add SSHMixin and SSHControllablePingableWOLSystem (Doggo)

dashboard/mixins.py

@@ -1,44 +0,0 @@
-# Copyright (c) 2025, Julian Müller (ChaoticByte)
-
-
-import platform
-import socket
-import subprocess
-
-from typing import Tuple, List
-
-from nicegui import ui
-
-
-class PingableMixin:
-
-    def ping(self) -> Tuple[bool, str, str]:
-        ''' requires the following attributes:
-            - self.host: str    Host ip address
-        '''
-        if platform.system().lower() == "windows": p = "-n"
-        else: p = "-c"
-        s = subprocess.run(
-            ["ping", p, '1', self.host],
-            stdout=subprocess.PIPE, stderr=subprocess.PIPE,
-            env={"LC_ALL": "C"} # don't translate
-        )
-        return s.returncode == 0, s.stdout.decode(), s.stderr.decode()
-
-
-class WakeOnLanMixin:
-
-    def wakeonlan(self):
-        ''' requires the following attributes:
-            - self.name: str        System.name
-            - self.host_mac: str    host mac address
-        '''
-        assert hasattr(self, "host_mac")
-        assert hasattr(self, "name")
-        host_mac_bin = bytes.fromhex(self.host_mac.replace(':', '').replace('-', '').lower())
-        magic_packet = (b'\xff' * 6) + (host_mac_bin * 16)
-        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
-            s.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
-            for port in [7, 9]: # we send to both port 7 and 9 to be compatible with most of the systems
-                s.sendto(magic_packet, ("255.255.255.255", port))
-            ui.notify(f"Magic packet sent to wake up '{self.name}' ({self.host_mac})")

dashboard/system.py

@@ -4,17 +4,19 @@
 # additional libraries
 
 
+import platform
 import re
+import socket
+import subprocess
 import time
 
 from enum import Enum
-from typing import List
+from typing import Dict, List, Tuple
 
 import requests, urllib3
-
 from nicegui import ui
-
-from .mixins import PingableMixin, WakeOnLanMixin
+from paramiko.client import SSHClient
+from paramiko.pkey import PKey
 
 
 # don't need the warning, bc. ssl verification needs to be disabled explicitly
@@ -73,6 +75,77 @@ class System:
         self.state_verbose = ""
 
 
+# Mixins
+
+
+class PingableMixin:
+
+    def ping(self) -> Tuple[bool, str, str]:
+        ''' requires the following attributes:
+            - self.host: str    Host ip address
+        '''
+        if platform.system().lower() == "windows": p = "-n"
+        else: p = "-c"
+        s = subprocess.run(
+            ["ping", p, '1', self.host],
+            stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+            env={"LC_ALL": "C"} # don't translate
+        )
+        return s.returncode == 0, s.stdout.decode(), s.stderr.decode()
+
+
+class WakeOnLanMixin:
+
+    def wakeonlan(self):
+        ''' requires the following attributes:
+            - self.name: str        System.name
+            - self.host_mac: str    host mac address
+        '''
+        assert hasattr(self, "host_mac")
+        assert hasattr(self, "name")
+        host_mac_bin = bytes.fromhex(self.host_mac.replace(':', '').replace('-', '').lower())
+        magic_packet = (b'\xff' * 6) + (host_mac_bin * 16)
+        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
+            s.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
+            for port in [7, 9]: # we send to both port 7 and 9 to be compatible with most of the systems
+                s.sendto(magic_packet, ("255.255.255.255", port))
+            ui.notify(f"Magic packet sent to wake up '{self.name}' ({self.host_mac})")
+
+
+class SSHMixin:
+
+    '''This mixin allows specifying SSH commands as actions.
+    Required attributes:
+        - self.host: str
+        - self.host_mac
+        - self.ssh_commands
+        - self.ssh_user
+        - self.ssh_key_file
+        - self.ssh_key_passphrase   Can be None
+        - self.ssh_port
+    '''
+
+    def ssh_exec(self, action_name: str, cmd: str):
+        ui.notify(f"Executing '{action_name}' on {self.name} ({self.host}) via SSH")
+        with SSHClient() as client:
+            client.load_system_host_keys()
+            client.connect(
+                self.host, port=self.ssh_port,
+                username=self.ssh_user,
+                key_filename=self.ssh_key_file, passphrase=self.ssh_key_passphrase)
+            chan = client.get_transport().open_session()
+            chan.set_combine_stderr(True)
+            chan.exec_command(cmd)
+            last_output_chunk = ""
+            while not chan.exit_status_ready(): # if we don't do that, we might deadlock
+                last_output_chunk = chan.recv(1024)
+            if not chan.exit_status == 0:
+                raise Exception(f"Exit status is {chan.exit_status}, last stdout: {last_output_chunk.decode()}")
+
+    def actions_from_ssh_commands(self):
+        return [Action(name, self.ssh_exec, name, cmd) for name, cmd in self.ssh_commands.items()]
+
+
 # Pingable System
 
 
@@ -120,6 +193,39 @@ class PingableWOLSystem(WakeOnLanMixin, PingableSystem):
         return actions
 
 
+# Pingable + WOL + SSH
+
+
+class SSHControllablePingableWOLSystem(SSHMixin, PingableWOLSystem):
+
+    def __init__(
+        self, name, description,
+        host_ip: str, host_mac: str,
+        ssh_commands: Dict[str, str], # dict containing "action name": "command ..."
+        ssh_user: str,
+        ssh_key_file: str,
+        ssh_key_passphrase: str = None,
+        ssh_port: int = 22,
+    ):
+        super().__init__(name, description, host_ip, host_mac)
+        self.host = host_ip
+        self.host_mac = host_mac
+        self.ssh_commands = ssh_commands
+        self.ssh_user = ssh_user
+        self.ssh_key_file = ssh_key_file
+        self.ssh_key_passphrase = ssh_key_passphrase
+        self.ssh_port = ssh_port
+
+    def get_actions(self) -> List[Action]:
+        actions = super().get_actions()
+        if not self.state == SystemState.FAILED:
+            actions.extend(self.actions_from_ssh_commands())
+        return actions
+
+# alias :)
+Doggo = SSHControllablePingableWOLSystem
+
+
 # HTTP Server System