#!/usr/bin/env python3 """ComputerArtz Remote Support - Linux/macOS Agent (Konsole) Verbindet sich mit dem Support-Server und ermöglicht Remote-Befehle. Keine GUI, keine externen Abhängigkeiten - nur Python 3 stdlib. Funktioniert auf Linux und macOS. Nutzung: python3 ca-support-linux.py python3 ca-support-linux.py --server https://support.computerartz.de python3 ca-support-linux.py --session ABCD (Reconnect mit alter Session-ID) """ import argparse import json import os import platform import signal import socket import subprocess import sys import tempfile import threading import time import urllib.request import urllib.error # === Konfiguration === DEFAULT_SERVER = "https://support.computerartz.de" POLL_INTERVAL = 2 # Sekunden zwischen Polls HEARTBEAT_INTERVAL = 10 # Sekunden zwischen Heartbeats CMD_TIMEOUT = 300 # Max 5 Minuten pro Befehl SESSION_FILE = os.path.expanduser("~/.ca-support-session") # === Farben (ANSI) === BOLD = "\033[1m" GREEN = "\033[32m" RED = "\033[31m" YELLOW = "\033[33m" CYAN = "\033[36m" RESET = "\033[0m" DIM = "\033[2m" def get_system_info(): """Sammelt Systeminformationen""" info = { "hostname": socket.gethostname(), "username": os.environ.get("USER", os.environ.get("LOGNAME", "unknown")), "os": "", "ip": "", "ram_gb": 0, "cpu": "", "disk": "", "domain": "" } is_macos = platform.system() == "Darwin" # OS-Version try: if is_macos: ver = platform.mac_ver()[0] info["os"] = f"macOS {ver}" if ver else "macOS" try: prod = subprocess.run(["sw_vers", "-productName"], capture_output=True, text=True, timeout=5) pver = subprocess.run(["sw_vers", "-productVersion"], capture_output=True, text=True, timeout=5) if prod.returncode == 0 and pver.returncode == 0: info["os"] = f"{prod.stdout.strip()} {pver.stdout.strip()}" except Exception: pass elif os.path.exists("/etc/os-release"): with open("/etc/os-release") as f: lines = f.read() for line in lines.splitlines(): if line.startswith("PRETTY_NAME="): info["os"] = line.split("=", 1)[1].strip('"') break if not info["os"]: info["os"] = f"{platform.system()} {platform.release()}" except Exception: info["os"] = f"{platform.system()} {platform.release()}" # IP-Adresse (bevorzugt nicht-localhost) try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(("8.8.8.8", 53)) info["ip"] = s.getsockname()[0] s.close() except Exception: info["ip"] = "127.0.0.1" # RAM try: if is_macos: r = subprocess.run(["sysctl", "-n", "hw.memsize"], capture_output=True, text=True, timeout=5) if r.returncode == 0: info["ram_gb"] = round(int(r.stdout.strip()) / (1024**3), 1) else: with open("/proc/meminfo") as f: for line in f: if line.startswith("MemTotal:"): kb = int(line.split()[1]) info["ram_gb"] = round(kb / 1024 / 1024, 1) break except Exception: pass # CPU try: if is_macos: r = subprocess.run(["sysctl", "-n", "machdep.cpu.brand_string"], capture_output=True, text=True, timeout=5) if r.returncode == 0 and r.stdout.strip(): info["cpu"] = r.stdout.strip() else: # Apple Silicon hat kein machdep.cpu.brand_string chip = subprocess.run(["sysctl", "-n", "hw.model"], capture_output=True, text=True, timeout=5) info["cpu"] = chip.stdout.strip() if chip.returncode == 0 else platform.machine() else: with open("/proc/cpuinfo") as f: for line in f: if line.startswith("model name"): info["cpu"] = line.split(":", 1)[1].strip() break if not info["cpu"]: info["cpu"] = platform.processor() or platform.machine() except Exception: info["cpu"] = platform.machine() # Festplatte (Root-Partition) try: st = os.statvfs("/") total = st.f_blocks * st.f_frsize free = st.f_bavail * st.f_frsize total_gb = total / (1024**3) free_gb = free / (1024**3) info["disk"] = f"{free_gb:.0f} GB frei von {total_gb:.0f} GB" except Exception: pass # Domain try: fqdn = socket.getfqdn() if "." in fqdn: info["domain"] = fqdn.split(".", 1)[1] else: info["domain"] = "WORKGROUP" except Exception: info["domain"] = "WORKGROUP" return info def api_request(url, data=None, method=None, timeout=30): """HTTP-Request an den Server (nur stdlib)""" headers = {"Content-Type": "application/json"} if data is not None: body = json.dumps(data).encode("utf-8") req = urllib.request.Request(url, data=body, headers=headers, method=method or "POST") else: req = urllib.request.Request(url, headers=headers, method=method or "GET") # SSL-Kontext (Self-Signed Certs akzeptieren) import ssl ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE try: with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp: return json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as e: body = e.read().decode("utf-8", errors="replace") raise Exception(f"HTTP {e.code}: {body[:200]}") except urllib.error.URLError as e: raise Exception(f"Verbindungsfehler: {e.reason}") def save_session(sid): """Session-ID speichern fuer Reconnect""" try: with open(SESSION_FILE, "w") as f: f.write(sid) except Exception: pass def load_session(): """Gespeicherte Session-ID laden""" try: with open(SESSION_FILE) as f: sid = f.read().strip() if len(sid) == 4 and sid.isalnum(): return sid except Exception: pass return None def execute_command(cmd, timeout=CMD_TIMEOUT): """Fuehrt einen Befehl per bash aus und gibt stdout, stderr, exit_code zurueck""" try: proc = subprocess.run( ["bash", "-c", cmd], capture_output=True, text=True, timeout=timeout, env={**os.environ, "LANG": "C.UTF-8", "LC_ALL": "C.UTF-8"} ) return proc.stdout, proc.stderr, proc.returncode except subprocess.TimeoutExpired: return "", f"Befehl abgebrochen nach {timeout} Sekunden (Timeout)", 124 except Exception as e: return "", str(e), 1 def print_banner(session_id, system_info): """Zeigt das Banner mit Session-ID""" width = 52 print() print(f"{CYAN}{'=' * width}{RESET}") os_label = "macOS" if platform.system() == "Darwin" else "Linux" print(f"{CYAN} ComputerArtz Remote Support - {os_label} Agent{RESET}") print(f"{CYAN}{'=' * width}{RESET}") print() print(f" {BOLD}Session-ID: {GREEN} {session_id} {RESET}") print() print(f" {DIM}Host: {RESET}{system_info['hostname']}") print(f" {DIM}User: {RESET}{system_info['username']}") print(f" {DIM}OS: {RESET}{system_info['os']}") print(f" {DIM}IP: {RESET}{system_info['ip']}") print(f" {DIM}CPU: {RESET}{system_info['cpu']}") print(f" {DIM}RAM: {RESET}{system_info['ram_gb']} GB") print(f" {DIM}Disk: {RESET}{system_info['disk']}") print() print(f" {DIM}Server: {RESET}{server_url}") print() print(f" {YELLOW}Warte auf Befehle... (Strg+C zum Beenden){RESET}") print(f"{CYAN}{'-' * width}{RESET}") print() def print_status(msg, color=DIM): """Status-Zeile ausgeben""" ts = time.strftime("%H:%M:%S") print(f" {DIM}[{ts}]{RESET} {color}{msg}{RESET}") # === Globale Variablen === server_url = DEFAULT_SERVER running = True cmd_count = 0 def heartbeat_loop(sid): """Heartbeat-Thread: Sendet regelmaessig Lebenszeichen""" while running: try: api_request( f"{server_url}/api/heartbeat/{sid}", data={"features": {"persist": False, "admin_user": False}}, timeout=10 ) except Exception: pass time.sleep(HEARTBEAT_INTERVAL) def main(): global server_url, running, cmd_count parser = argparse.ArgumentParser(description="ComputerArtz Remote Support - Linux Agent") parser.add_argument("--server", default=DEFAULT_SERVER, help="Server-URL") parser.add_argument("--session", default=None, help="Vorherige Session-ID fuer Reconnect") parser.add_argument("--no-save", action="store_true", help="Session-ID nicht speichern") args = parser.parse_args() server_url = args.server.rstrip("/") # Signal-Handler fuer sauberes Beenden def signal_handler(sig, frame): global running running = False print(f"\n {YELLOW}Beende...{RESET}\n") sys.exit(0) signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) # Systeminformationen sammeln print(f"\n {DIM}Sammle Systeminformationen...{RESET}") system_info = get_system_info() # Session-ID (Reconnect oder gespeichert) session_id = args.session or load_session() # Registrierung print(f" {DIM}Verbinde mit {server_url}...{RESET}") retry_count = 0 max_retries = 30 # 90 Sekunden while running: try: reg_data = {"system_info": system_info} if session_id: reg_data["session_id"] = session_id result = api_request(f"{server_url}/api/register", data=reg_data) session_id = result["session_id"] if result.get("reconnected"): print(f" {GREEN}Reconnect erfolgreich!{RESET}") break except Exception as e: retry_count += 1 if retry_count >= max_retries: print(f"\n {RED}Konnte keine Verbindung herstellen: {e}{RESET}") sys.exit(1) if retry_count == 1: print(f" {YELLOW}Server nicht erreichbar, versuche erneut...{RESET}") time.sleep(3) # Session-ID speichern if not args.no_save: save_session(session_id) # Banner anzeigen print_banner(session_id, system_info) # Heartbeat-Thread starten hb_thread = threading.Thread(target=heartbeat_loop, args=(session_id,), daemon=True) hb_thread.start() # === Polling-Loop === consecutive_errors = 0 while running: try: resp = api_request(f"{server_url}/api/poll/{session_id}", timeout=10) consecutive_errors = 0 cmd = resp.get("command") if cmd is None: time.sleep(POLL_INTERVAL) continue cmd_id = cmd["id"] command = cmd["command"] cmd_count += 1 print_status(f"Befehl #{cmd_count}: {command[:80]}{'...' if len(command) > 80 else ''}", CYAN) # Befehl ausfuehren stdout, stderr, exit_code = execute_command(command) # Ergebnis senden api_request( f"{server_url}/api/result/{session_id}/{cmd_id}", data={ "stdout": stdout, "stderr": stderr, "exit_code": exit_code } ) status = f"{GREEN}OK{RESET}" if exit_code == 0 else f"{RED}Exit {exit_code}{RESET}" out_lines = len(stdout.splitlines()) if stdout else 0 print_status(f" -> {status} ({out_lines} Zeilen)") except Exception as e: consecutive_errors += 1 if consecutive_errors == 1: print_status(f"Verbindungsfehler: {e}", RED) elif consecutive_errors == 5: print_status("Mehrere Fehler - versuche weiter...", YELLOW) elif consecutive_errors >= 30: print_status("Verbindung verloren. Beende.", RED) break # Exponentielles Backoff, max 30s wait = min(POLL_INTERVAL * (2 ** min(consecutive_errors, 4)), 30) time.sleep(wait) print(f"\n {DIM}Agent beendet.{RESET}\n") # Session-Datei aufraeumen try: os.remove(SESSION_FILE) except Exception: pass if __name__ == "__main__": main()