From 723c49fc9d8b54fb991ed8e041ef072d404ff4e0 Mon Sep 17 00:00:00 2001 From: ChaoticByte Date: Fri, 28 Feb 2025 21:23:28 +0100 Subject: [PATCH] Moved mixins back into system.py to avoid circular imports, add SSHMixin and SSHControllablePingableWOLSystem (Doggo) --- dashboard/mixins.py | 44 ----------------- dashboard/system.py | 114 ++++++++++++++++++++++++++++++++++++++++++-- 2 files changed, 110 insertions(+), 48 deletions(-) delete mode 100644 dashboard/mixins.py diff --git a/dashboard/mixins.py b/dashboard/mixins.py deleted file mode 100644 index baa677d..0000000 --- a/dashboard/mixins.py +++ /dev/null @@ -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})") diff --git a/dashboard/system.py b/dashboard/system.py index 0dc5a61..7432822 100644 --- a/dashboard/system.py +++ b/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