""" ARIA Satellit — Info-/Gateway-Aussenposten in einem fremden Netz. Laeuft eigenstaendig (z.B. im Buero) und verbindet sich als RVS-Client in Stefans Raum (gleicher Token). Gibt ARIA damit Augen + Haende in DIESEM Netz: Augen: entdeckt Geraete (mDNS/Zeroconf, SSDP/UPnP + DIAL, ARP-Tabelle) und meldet ein Inventar → sat_devices. Haende: steuert Geraete (DIAL-App-Launch z.B. YouTube auf Fire TV, Wake-on- LAN, generisches HTTP) → sat_command / sat_result. Nur wenn CONTROL_ENABLED=true, Aktion in der Allowlist, alles geloggt. Adressierung: mehrere Satelliten haengen im selben RVS-Raum. Jeder hat eine SATELLITE_ID (technisch, eindeutig) + SATELLITE_LOCATION (menschlich, "Buero"). ARIA spricht einen Satelliten ueber seine ID/Location an. Message-Typen (RVS, Base64/JSON-Relay wie der Rest): raus: sat_hello {id, location, caps, ts} sat_devices {requestId, satellite, devices:[...]} sat_result {requestId, satellite, ok, result|error} rein: sat_discover {satellite?, requestId} sat_command {satellite?, requestId, device, action, params} Sicherheit: reagiert nur auf den eigenen RVS-Raum (Token). Commands brauchen CONTROL_ENABLED + Allowlist. Discovery ist read-only. Keine offenen Ports. """ from __future__ import annotations import asyncio import base64 import json import logging import os import re import socket import struct import threading import time from pathlib import Path from typing import Optional import websockets logging.basicConfig( level=logging.INFO, format="%(asctime)s [satellite] %(levelname)s %(message)s", ) logger = logging.getLogger("satellite") def _load_dotenv() -> None: """Laedt eine .env neben dem Script (oder im CWD) in os.environ — fuer den NATIVEN Start (`python satellite.py`). In Docker sind die Variablen via env_file schon gesetzt; bereits gesetzte Werte gewinnen (werden NICHT ueberschrieben). Kein python-dotenv noetig.""" here = os.path.dirname(os.path.abspath(__file__)) for path in (os.path.join(here, ".env"), os.path.join(os.getcwd(), ".env")): if not os.path.isfile(path): continue try: with open(path, "r", encoding="utf-8") as f: for line in f: line = line.strip() if not line or line.startswith("#") or "=" not in line: continue key, _, val = line.partition("=") key = key.strip() if key.startswith("export "): key = key[len("export "):].strip() val = val.strip() if val[:1] in ("'", '"'): # Gequotet: Inhalt bis zum schliessenden Quote, Rest (Kommentar) egal. q = val[0] end = val.find(q, 1) val = val[1:end] if end != -1 else val[1:] else: # Ungequotet: Inline-Kommentar (Whitespace + #) abschneiden. m = re.search(r"\s+#", val) if m: val = val[:m.start()] val = val.strip() if key and key not in os.environ: os.environ[key] = val except Exception as exc: logging.getLogger("satellite").warning(".env laden fehlgeschlagen (%s): %s", path, exc) break # erste gefundene .env gewinnt _load_dotenv() # ─── Konfiguration ────────────────────────────────────────────────── def _env_bool(name: str, default: bool) -> bool: v = os.environ.get(name) if v is None: return default return v.strip().lower() in ("1", "true", "yes", "on", "ja") def _default_id() -> str: host = socket.gethostname() or "satellite" slug = re.sub(r"[^a-zA-Z0-9_-]+", "-", host).strip("-").lower() return slug or "satellite" RVS_HOST = os.environ.get("RVS_HOST", "") RVS_PORT = int(os.environ.get("RVS_PORT", "443") or "443") RVS_TLS = _env_bool("RVS_TLS", True) # Bei TLS-Fehlschlag einmal auf ws:// zurueckfallen (wie die Compute-Bridges). RVS_TLS_FALLBACK = _env_bool("RVS_TLS_FALLBACK", True) RVS_TOKEN = os.environ.get("RVS_TOKEN", "") # TLS-Hostname (SNI + Zertifikatspruefung), falls RVS_HOST eine IP ist — z.B. ein # Satellit im selben Netz wie der RVS, der direkt auf die interne IP verbindet # (das Caddy-Zertifikat gilt aber fuer den Namen). Leer = SNI = RVS_HOST. RVS_SNI = os.environ.get("RVS_SNI", "").strip() # In-Process-DNS-Override: wenn RVS_SNI gesetzt ist, verbindet die URI ueber den # HOSTNAMEN (Host-Header + SNI + Cert stimmen), waehrend getaddrinfo den Namen auf # die echte IP in RVS_HOST aufloest. Versionsunabhaengig — host=/port= kollidiert # in der Legacy-websockets-API mit dem aus der URI abgeleiteten Host. if RVS_TLS and RVS_SNI: import socket as _socket _orig_getaddrinfo = _socket.getaddrinfo def _sni_getaddrinfo(host, *a, **k): return _orig_getaddrinfo(RVS_HOST if host == RVS_SNI else host, *a, **k) _socket.getaddrinfo = _sni_getaddrinfo SATELLITE_ID = (os.environ.get("SATELLITE_ID") or _default_id()).strip() SATELLITE_LOCATION = (os.environ.get("SATELLITE_LOCATION") or SATELLITE_ID).strip() CONTROL_ENABLED = _env_bool("CONTROL_ENABLED", False) CONTROL_ALLOWLIST = [ a.strip() for a in os.environ.get("CONTROL_ALLOWLIST", "dial.launch,wol,http.get,snmp.get,snmp.walk,snmp.printer," "snmp.ports,snmp.info,fritzbox.*,ssh.exec," "adb.devices,adb.connect,adb.install,adb.shell,adb.provision,adb.status").split(",") if a.strip() ] # ── Port-Tunnel ────────────────────────────────────────────────────── # # ARIA sitzt woanders und kommt nicht ins fremde Netz. Der Satellit kann aber # beides: er sieht das LAN und haelt eine ausgehende Verbindung zum RVS. Also # reicht er eine TCP-Verbindung durch — wie der VNC-Tunnel, nur fuer beliebige # Ports. Damit erreicht ARIA z.B. einen FreeCAD-MCP oder eine Web-App, die nur # im Buero-Netz lauscht. # # DEFAULT AUS. Das ist ein Loch in ein fremdes Netz — das schaltet man bewusst # ein, nicht aus Versehen. TUNNEL_ENABLED = _env_bool("TUNNEL_ENABLED", False) # Soll der Rechner, auf dem der Satellit laeuft, selbst in der Geraeteliste # auftauchen? Default JA — meist ist er ein ganz normaler Host mit weiteren # Diensten. Wer den Satelliten-Rechner bewusst heraushalten will, setzt das # auf false. LIST_SELF = _env_bool("LIST_SELF", True) # Welche Ziele? Leer = nur Geraete, die der Satellit bei der Suche auch # wirklich GEFUNDEN hat (beliebiger Port). Damit kann der Tunnel nichts # ansteuern, was nicht ohnehin im LAN sichtbar ist — insbesondere nichts im # Internet. Sonst Liste aus "host" oder "host:port", * als Platzhalter: # TUNNEL_ALLOW=192.168.1.50:8080,192.168.1.60,*:5000 TUNNEL_ALLOW = [a.strip() for a in os.environ.get("TUNNEL_ALLOW", "").split(",") if a.strip()] TUNNEL_MAX = int(os.environ.get("TUNNEL_MAX", "8") or 8) TUNNEL_IDLE_SEC = int(os.environ.get("TUNNEL_IDLE_SEC", "900") or 900) # adb.*: Android-Geraete (v.a. Fire TV) im LAN per Netzwerk-Debugging erreichen. # install dauert lange (APK laden + uebertragen + installieren) -> grosszuegiger # Timeout. APK-Cache in /data (persistentes Volume), damit N Sticks nur EINEN # Download brauchen. ADB_PORT = int(os.environ.get("ADB_PORT", "5555") or "5555") ADB_TIMEOUT_SEC = float(os.environ.get("ADB_TIMEOUT_SEC", "180") or "180") # Beim ERSTEN Verbinden zeigt das Geraet "USB-Debugging zulassen?" auf dem # Bildschirm. Statt mit Fehler abzubrechen (und den Nutzer nochmal klicken zu # lassen) wartet der Satellit so lange auf die Bestaetigung. ADB_AUTH_WAIT_SEC = float(os.environ.get("ADB_AUTH_WAIT_SEC", "90") or "90") APK_CACHE_DIR = os.environ.get("APK_CACHE_DIR", "/data/apk") APK_MAX_BYTES = int(os.environ.get("APK_MAX_BYTES", str(64 * 1024 * 1024)) or str(64 * 1024 * 1024)) SCAN_INTERVAL_SEC = int(os.environ.get("SCAN_INTERVAL_SEC", "300") or "300") DISCOVER_TIMEOUT_SEC = float(os.environ.get("DISCOVER_TIMEOUT_SEC", "6") or "6") # Obergrenze fuer einen ganzen Scan. Haengt ein Schritt (mDNS, SNMP, ADB …), # kommt nach dieser Zeit die letzte bekannte Liste zurueck statt gar nichts. SCAN_MAX_SEC = float(os.environ.get("SCAN_MAX_SEC", "90") or "90") DEVICE_CACHE_TTL_SEC = int(os.environ.get("DEVICE_CACHE_TTL_SEC", "120") or "120") # http.get/http.post: Body-Ausschnitt. Default grosszuegig (ganze Statusseiten # passen), mit hartem Deckel gegen Riesen-Payloads durchs RVS. offset/max_chars # pro Request ueberschreibbar; contains-Filter zieht nur relevante Zeilen. HTTP_TIMEOUT_SEC = float(os.environ.get("HTTP_TIMEOUT_SEC", "10") or "10") HTTP_MAX_CHARS = int(os.environ.get("HTTP_MAX_CHARS", "20000") or "20000") HTTP_MAX_CHARS_HARD = int(os.environ.get("HTTP_MAX_CHARS_HARD", "200000") or "200000") # SSH (ssh.exec): Verbindungs-/Kommando-Timeout. Auth aus dem Credential-Store # (Typ 'ssh': user + pass ODER key). Default-Port 22, pro Request/Cred ueberschreibbar. SSH_TIMEOUT_SEC = float(os.environ.get("SSH_TIMEOUT_SEC", "20") or "20") # SNMP (net-snmp-CLI): Default-Community/Version + Timeout. Drucker antworten # i.d.R. auf community 'public', v2c. SNMP_COMMUNITY = os.environ.get("SNMP_COMMUNITY", "public") or "public" SNMP_VERSION = os.environ.get("SNMP_VERSION", "2c") or "2c" SNMP_TIMEOUT_SEC = float(os.environ.get("SNMP_TIMEOUT_SEC", "5") or "5") # Printer-MIB (RFC 3805) prtMarkerSuppliesEntry-Spalten (numerisch, ohne MIB-Files): SNMP_SUPPLY_DESC = "1.3.6.1.2.1.43.11.1.1.6.1" # Beschreibung (z.B. "Black Ink") SNMP_SUPPLY_MAX = "1.3.6.1.2.1.43.11.1.1.8.1" # Max-Kapazitaet SNMP_SUPPLY_LVL = "1.3.6.1.2.1.43.11.1.1.9.1" # aktueller Fuellstand # SNMP-Anreicherung bei der Discovery: jedes entdeckte Geraet mit IP wird kurz # nach seiner System-Group (RFC 1213) gefragt. Switches/Router/APs/NAS geben so # Name, Beschreibung, Standort & Uptime preis -> im Inventar (satellite_devices) # sichtbar. Abschaltbar; kurzer Timeout + parallel, damit der Scan flott bleibt. SNMP_DISCOVERY = _env_bool("SNMP_DISCOVERY", True) SNMP_DISCOVERY_CONCURRENCY = int(os.environ.get("SNMP_DISCOVERY_CONCURRENCY", "16") or "16") SNMP_DISCOVERY_TIMEOUT = float(os.environ.get("SNMP_DISCOVERY_TIMEOUT", "2") or "2") # System-Group (RFC 1213) .0-Instanzen: SNMP_SYS_OIDS = { "descr": "1.3.6.1.2.1.1.1.0", # sysDescr "objectid": "1.3.6.1.2.1.1.2.0", # sysObjectID "uptime": "1.3.6.1.2.1.1.3.0", # sysUpTime "contact": "1.3.6.1.2.1.1.4.0", # sysContact "name": "1.3.6.1.2.1.1.5.0", # sysName "location": "1.3.6.1.2.1.1.6.0", # sysLocation } # ─── Geraete-Credential-Store (verschluesselt, pro IP) ───────────── # Diagnostic legt via sat_creds_set pro Geraet Zugangsdaten ab (SNMP-Community/ # v3, HTTP-Basic, FritzBox-Login). Der Satellit nutzt sie automatisch bei snmp.*/ # http/fritzbox. Persistiert verschluesselt (Fernet) in einem Bind-Volume. CREDS_PATH = os.environ.get("CREDS_PATH", "/data/credentials.json.enc") CREDS_KEY_PATH = os.environ.get("CREDS_KEY_PATH", "/data/creds.key") _CREDS: dict = {} # {ip: {snmp:{...}, http:{...}, fritzbox:{...}}} _creds_fernet = None # Fernet-Instanz (lazy) def _creds_cipher(): """Fernet-Instanz; Schluessel aus CREDS_KEY (env) oder Schluesseldatei im Volume (wird beim ersten Start erzeugt, 0600).""" global _creds_fernet if _creds_fernet is not None: return _creds_fernet from cryptography.fernet import Fernet key = os.environ.get("CREDS_KEY", "").strip().encode() or None if not key: kp = Path(CREDS_KEY_PATH) if kp.exists(): key = kp.read_bytes().strip() else: key = Fernet.generate_key() kp.parent.mkdir(parents=True, exist_ok=True) kp.write_bytes(key) try: os.chmod(kp, 0o600) except OSError: pass logger.info("[creds] neuer Verschluesselungs-Schluessel erzeugt: %s", CREDS_KEY_PATH) _creds_fernet = Fernet(key) return _creds_fernet def _creds_load() -> None: global _CREDS p = Path(CREDS_PATH) if not p.exists(): _CREDS = {} return try: blob = p.read_bytes() raw = _creds_cipher().decrypt(blob) _CREDS = json.loads(raw.decode("utf-8")) or {} logger.info("[creds] %d Geraete-Eintraege geladen", len(_CREDS)) except Exception as exc: logger.warning("[creds] laden fehlgeschlagen (%s) — starte leer", exc) _CREDS = {} def _creds_save() -> None: p = Path(CREDS_PATH) p.parent.mkdir(parents=True, exist_ok=True) blob = _creds_cipher().encrypt(json.dumps(_CREDS).encode("utf-8")) p.write_bytes(blob) try: os.chmod(p, 0o600) except OSError: pass def _creds_for(ip: str) -> dict: return _CREDS.get((ip or "").strip(), {}) if ip else {} _CRED_SECRETS = ("pass", "key", "community", "authKey", "privKey", "backupPass") def _strong_password_problem(pw: str) -> str: """"" wenn stark genug, sonst was fehlt. Regel fuer Sicherungs-Kennwoerter: mind. 12 Zeichen, Gross- und Kleinbuchstabe, Ziffer, Sonderzeichen.""" pw = str(pw or "") missing = [] if len(pw) < 12: missing.append("mindestens 12 Zeichen") if not re.search(r"[A-Z]", pw): missing.append("ein Grossbuchstabe") if not re.search(r"[a-z]", pw): missing.append("ein Kleinbuchstabe") if not re.search(r"\d", pw): missing.append("eine Ziffer") if not re.search(r"[^A-Za-z0-9]", pw): missing.append("ein Sonderzeichen") return ", ".join(missing) def _creds_detail(d: dict, reveal: bool) -> dict: """Ein Cred-Eintrag fuer die Diagnostic: Benutzer, Port, Version … immer; Geheimes nur als has_=True — ausser reveal (Auge im Formular).""" out = {} for k, v in (d or {}).items(): if isinstance(v, dict): out[k] = _creds_detail(v, reveal) elif k in _CRED_SECRETS: out["has_" + k] = bool(v) if reveal: out[k] = v else: out[k] = v return out def _creds_keep_secrets(new: dict, old: dict) -> dict: """Beim Speichern: fehlt ein Geheimnis im neuen Eintrag (Feld unangetastet), bleibt das alte. So muss man ein Passwort nicht neu tippen, um nur den Benutzer zu aendern.""" out = dict(new) for k, v in (old or {}).items(): if isinstance(v, dict) and isinstance(out.get(k), dict): out[k] = _creds_keep_secrets(out[k], v) elif k in _CRED_SECRETS and k not in out: out[k] = v return out def _creds_public_summary(ip: str = "", reveal: bool = False) -> list: """Fuer sat_creds_list: welche Geraete welche Cred-Typen haben, dazu die Felder OHNE Secrets (Benutzer, Port, SNMP-Version). Mit ip + reveal auch die Secrets dieses einen Geraets.""" out = [] for dev_ip, entry in sorted(_CREDS.items()): if ip and dev_ip != ip: continue types = [t for t in ("snmp", "http", "fritzbox", "ssh") if entry.get(t)] out.append({"ip": dev_ip, "types": types, "detail": {t: _creds_detail(entry[t], reveal and bool(ip)) for t in types}}) return out HEARTBEAT_SEC = 25 # mDNS-Servicetypen, die fuer ARIA interessant sind. MDNS_TYPES = [ "_googlecast._tcp.local.", # Chromecast / Google TV / Nest "_airplay._tcp.local.", # Apple TV / AirPlay "_raop._tcp.local.", # AirPlay-Audio "_spotify-connect._tcp.local.", # Spotify-Geraete "_sonos._tcp.local.", # Sonos "_hap._tcp.local.", # HomeKit "_printer._tcp.local.", # Drucker "_ipp._tcp.local.", # Drucker (IPP) "_smb._tcp.local.", # NAS / Fileshares "_workstation._tcp.local.", # generische Hosts "_http._tcp.local.", # Web-UIs (Router, NAS, IoT) "_amzn-wplay._tcp.local.", # Amazon Whisperplay — zuverlaessigster Fire-TV-Marker "_adb-tls-connect._tcp.local.", # Android 11+ Wireless Debugging "_androidtvremote2._tcp.local.", # Android TV Remote ] CAPABILITIES = ["discover", "name"] if CONTROL_ENABLED: CAPABILITIES += CONTROL_ALLOWLIST if TUNNEL_ENABLED: # Macht in Diagnostic/ARIA sichtbar, dass dieser Satellit Ports durchreicht. CAPABILITIES.append("tunnel") # ─── Netz-Kontext / Selbstdiagnose ────────────────────────────────── def _in_docker_bridge(ip: str) -> bool: # Docker-Default-Bridge-Range 172.16.0.0/12 try: a, b = ip.split(".")[:2] return a == "172" and 16 <= int(b) <= 31 except Exception: return False def _net_context() -> dict: """Ermittelt in welchem Netz der Satellit LAeUFT — und warnt, wenn das ein Docker-/NAT-Netz ist (dann erreicht Discovery das echte LAN nicht).""" ips: list[str] = [] primary = "" try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.settimeout(1) s.connect(("8.8.8.8", 80)) primary = s.getsockname()[0] s.close() except Exception: pass try: for info in socket.getaddrinfo(socket.gethostname(), None, socket.AF_INET): ip = info[4][0] if ip and not ip.startswith("127.") and ip not in ips: ips.append(ip) except Exception: pass if primary and primary not in ips: ips.insert(0, primary) p = primary or (ips[0] if ips else "") warning = "" if p.startswith("192.168.65.") or _in_docker_bridge(p): warning = (f"Satellit laeuft in einem Docker-/NAT-Netz ({p}), NICHT im echten LAN. " "mDNS/SSDP erreichen die realen Geraete so nicht. Auf Docker Desktop " "(Mac/Windows) geht LAN-Discovery nicht — den Satelliten NATIV (python " "satellite.py) oder auf einem Linux-Host im Ziel-LAN betreiben.") return {"primary_ip": p, "ips": ips, "mac": _own_mac(p), "warning": warning} def _own_mac(ip: str) -> str: """MAC der eigenen (primaeren) Netzwerkkarte — identifiziert die Kiste eindeutig (DHCP-Reservierung, WOL). Liest /sys/class/net (Linux); faellt auf uuid.getnode() zurueck, wenn das Interface nicht zuzuordnen ist.""" import subprocess try: base = "/sys/class/net" if os.path.isdir(base): # Interface zur primaeren IP finden (ueber die Adressen im /proc-Netz). want = "" if ip: try: out = subprocess.run(["ip", "-o", "-4", "addr", "show"], capture_output=True, text=True, timeout=3).stdout for line in out.splitlines(): parts = line.split() if len(parts) >= 4 and parts[3].split("/")[0] == ip: want = parts[1] break except Exception: pass names = [want] if want else [] names += [n for n in sorted(os.listdir(base)) if n != "lo" and n not in names] for n in names: try: with open(os.path.join(base, n, "address"), "r") as f: mac = f.read().strip() if mac and mac != "00:00:00:00:00:00": return mac.lower() except Exception: continue except Exception: pass try: import uuid node = uuid.getnode() return ":".join(f"{(node >> e) & 0xFF:02x}" for e in range(40, -8, -8)) except Exception: return "" NET = _net_context() # ─── Discovery ────────────────────────────────────────────────────── def _discover_mdns(timeout: float) -> list[dict]: """Blockierend (im Executor): mDNS/Zeroconf-Sweep ueber MDNS_TYPES.""" out: dict[str, dict] = {} try: from zeroconf import Zeroconf, ServiceBrowser except Exception as exc: logger.warning("zeroconf nicht verfuegbar: %s", exc) return [] class _Listener: def add_service(self, zc, type_, name): try: info = zc.get_service_info(type_, name, timeout=2000) except Exception: info = None if not info: return ips = [] try: for addr in info.parsed_addresses(): ips.append(addr) except Exception: pass props = {} try: for k, v in (info.properties or {}).items(): try: props[k.decode("utf-8", "ignore")] = ( v.decode("utf-8", "ignore") if isinstance(v, (bytes, bytearray)) else v) except Exception: pass except Exception: pass friendly = name.split("." + type_.split(".", 1)[0])[0].strip(".") fn = props.get("fn") or props.get("friendlyName") or friendly dev_id = _slug(f"{fn}-{ips[0] if ips else name}") out[dev_id] = { "id": dev_id, "name": fn, "type": _mdns_kind(type_), "ip": ips[0] if ips else "", "port": info.port, "via": "mdns", "service": type_, "model": props.get("md") or props.get("model") or "", } def update_service(self, *a): pass def remove_service(self, *a): pass zc = None browsers = [] try: zc = Zeroconf() listener = _Listener() for t in MDNS_TYPES: try: browsers.append(ServiceBrowser(zc, t, listener)) except Exception: pass time.sleep(timeout) except Exception as exc: logger.warning("mDNS-Sweep-Fehler: %s", exc) finally: # Jeder ServiceBrowser ist ein eigener Thread, den zc.close() NICHT # beendet — ohne cancel() blieben pro Scan 14 Threads liegen. Nach # ~5 Tagen war das Thread-Limit des Containers voll (18.841), danach # scheiterte jeder Scan mit "[Errno 11] Resource temporarily unavailable". for b in browsers: try: b.cancel() except Exception: pass try: if zc: zc.close() except Exception: pass return list(out.values()) def _mdns_kind(service_type: str) -> str: m = { "_googlecast": "cast", "_airplay": "airplay", "_raop": "airplay-audio", "_spotify-connect": "spotify", "_sonos": "sonos", "_hap": "homekit", "_printer": "printer", "_ipp": "printer", "_smb": "fileshare", "_workstation": "host", "_http": "web", "_amzn-wplay": "firetv", "_adb-tls-connect": "android-adb", "_androidtvremote2": "androidtv", } for k, v in m.items(): if service_type.startswith(k): return v return "unknown" def _discover_ssdp(timeout: float) -> list[dict]: """Blockierend: SSDP M-SEARCH (UPnP + DIAL). Liefert v.a. Smart-TVs / Fire TV mit ihrer DIAL Application-URL (fuer App-Launch wie YouTube).""" out: dict[str, dict] = {} targets = [ "urn:dial-multiscreen-org:service:dial:1", "ssdp:all", ] for st in targets: msg = ( "M-SEARCH * HTTP/1.1\r\n" "HOST: 239.255.255.250:1900\r\n" 'MAN: "ssdp:discover"\r\n' "MX: 2\r\n" f"ST: {st}\r\n\r\n" ).encode("utf-8") sock = None try: sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 2) sock.settimeout(timeout) sock.sendto(msg, ("239.255.255.250", 1900)) deadline = time.time() + timeout while time.time() < deadline: try: data, addr = sock.recvfrom(65507) except socket.timeout: break except Exception: break headers = _parse_http_headers(data.decode("utf-8", "ignore")) location = headers.get("location", "") dial_app = headers.get("application-url", "") ip = addr[0] dev = _fetch_upnp_description(location) if location else {} name = dev.get("name") or headers.get("server", "") or ip dev_id = _slug(f"{name}-{ip}") entry = out.get(dev_id, { "id": dev_id, "name": name, "type": "media-renderer", "ip": ip, "via": "ssdp", }) if dev.get("name"): entry["name"] = dev["name"] if dev.get("model"): entry["model"] = dev["model"] # Router (FritzBox & Co.) melden sich per SSDP als # InternetGatewayDevice — kein Medienabspieler. if re.search(r"InternetGatewayDevice|FRITZ!Box", f"{entry['name']} {dev.get('model', '')}"): entry["type"] = "router" if dev.get("manufacturer"): entry["manufacturer"] = dev["manufacturer"] if dial_app or dev.get("dialAppUrl"): entry["dialAppUrl"] = dial_app or dev.get("dialAppUrl") entry["type"] = "dial" out[dev_id] = entry except Exception as exc: logger.debug("SSDP (%s) Fehler: %s", st, exc) finally: try: if sock: sock.close() except Exception: pass return list(out.values()) def _fetch_upnp_description(location: str) -> dict: try: import requests r = requests.get(location, timeout=3) dial_app = r.headers.get("Application-URL", "") xml = r.text name = _xml_tag(xml, "friendlyName") model = _xml_tag(xml, "modelName") manuf = _xml_tag(xml, "manufacturer") return {"name": name, "model": model, "manufacturer": manuf, "dialAppUrl": dial_app} except Exception: return {} def _discover_arp() -> list[dict]: """Rohe Host-Liste aus der ARP-Tabelle (kein aktiver Scan).""" out = [] try: with open("/proc/net/arp", "r", encoding="utf-8") as f: lines = f.read().splitlines()[1:] for ln in lines: parts = ln.split() if len(parts) < 4: continue ip, _hw, _flags, mac = parts[0], parts[1], parts[2], parts[3] if mac == "00:00:00:00:00:00": continue # Docker-interne Netze (Container auf DIESEM Rechner) sind keine # Geraete — laeuft der Satellit auf einem Docker-Host (z.B. ARIAs # VM), stuenden sonst alle Container in der Liste. iface = parts[5] if len(parts) > 5 else "" if iface.startswith(("docker", "br-", "veth")): continue out.append({ "id": _slug(f"host-{ip}"), "name": ip, "type": "host", "ip": ip, "mac": mac, "via": "arp", }) except Exception: pass return out def _merge_devices(*lists) -> list[dict]: """Fuehrt Geraetelisten zusammen, dedupt per IP (reichere Quelle gewinnt).""" by_ip: dict[str, dict] = {} loose: list[dict] = [] order = {"mdns": 3, "ssdp": 2, "arp": 1} for lst in lists: for d in lst: ip = d.get("ip") or "" if not ip: loose.append(d) continue cur = by_ip.get(ip) if not cur: by_ip[ip] = d else: # bessere Quelle / mehr Felder → mergen merged = {**d, **{k: v for k, v in cur.items() if v}} if order.get(d.get("via"), 0) >= order.get(cur.get("via"), 0): merged.update({k: v for k, v in d.items() if v}) # DIAL-URL / mac aus beiden behalten. isSatellite/satelliteId # muessen mit: der Satellit findet sich selbst auch per mDNS, # und dann duerfte die Kennzeichnung nicht verlorengehen. for key in ("dialAppUrl", "mac", "model", "manufacturer", "isSatellite", "satelliteId", "hostname"): merged[key] = d.get(key) or cur.get(key) or merged.get(key) by_ip[ip] = {k: v for k, v in merged.items() if v not in (None, "")} return list(by_ip.values()) + loose # ─── Geraete-Namen ────────────────────────────────────────────────── # # "duffy-desktop" ist ansprechbar, "192.168.1.47" nicht. Darum kann Stefan # jedem gefundenen Geraet einen Namen geben; ARIA adressiert danach. # # Schluessel ist die MAC, nicht die IP: per DHCP wandert die IP, die MAC # bleibt. Nur wenn keine MAC bekannt ist, faellt es auf die IP zurueck. NAMES_PATH = os.environ.get("DEVICE_NAMES_PATH", "/data/device_names.json") _names: dict = {} def _names_load() -> None: global _names try: with open(NAMES_PATH) as f: data = json.load(f) _names = data if isinstance(data, dict) else {} except FileNotFoundError: _names = {} except Exception as exc: logger.warning("device_names.json unlesbar (%s) — starte leer", exc) _names = {} def _names_save() -> None: try: os.makedirs(os.path.dirname(NAMES_PATH), exist_ok=True) tmp = NAMES_PATH + ".tmp" with open(tmp, "w") as f: json.dump(_names, f, indent=2) os.replace(tmp, NAMES_PATH) except Exception as exc: logger.warning("device_names.json schreiben fehlgeschlagen: %s", exc) def _self_device() -> dict: """Der Rechner, auf dem der Satellit SELBST laeuft, als normales Geraet. Er taucht in der Suche nicht auf (man findet sich nicht per mDNS/ARP selbst), ist aber oft der interessanteste Host im Netz: hier laufen typischerweise noch andere Dienste — ein FreeCAD-MCP, eine Web-App, ein Dateiserver. Ohne diesen Eintrag waere ausgerechnet der Rechner unsichtbar, auf dem ARIA schon eine Verbindung hat. `isSatellite` kennzeichnet ihn; ansonsten ist er ein Geraet wie jedes andere und laesst sich genauso ansprechen. """ ip = (NET or {}).get("primary_ip") or "" if not ip: return {} d = { "ip": ip, "mac": (NET or {}).get("mac") or "", "via": "self", "isSatellite": True, "satelliteId": SATELLITE_ID, "type": "host", } try: d["hostname"] = socket.gethostname() except Exception: pass return {k: v for k, v in d.items() if v not in (None, "")} def _name_key(dev: dict) -> str: mac = (dev.get("mac") or "").strip().lower() return mac if mac else (dev.get("ip") or "").strip() def _apply_names(devices: list[dict]) -> list[dict]: """Haengt den vergebenen Namen an jedes Geraet (Feld 'name').""" for d in devices: n = _names.get(_name_key(d)) if n: d["name"] = n return devices def _resolve_target(target: str, devices: list[dict]) -> str: """Name ODER IP ODER Hostname -> IP. Erlaubt ARIA, "duffy-desktop" zu sagen statt eine Adresse zu kennen. Vergleich case-insensitive; eine direkt angegebene IP geht unveraendert durch. """ t = (target or "").strip() if not t: return "" if re.match(r"^\d{1,3}(\.\d{1,3}){3}$", t): return t tl = t.lower() for d in devices: if (d.get("name") or "").strip().lower() == tl: return d.get("ip") or "" # Zweite Runde: Hostname aus der Suche (mDNS liefert oft einen) for d in devices: for field in ("hostname", "host", "model"): if (d.get(field) or "").strip().lower().rstrip(".") == tl: return d.get("ip") or "" return "" # ─── Port-Tunnel ──────────────────────────────────────────────────── def _tunnel_allowed(ip: str, port: int, devices: list[dict]) -> tuple[bool, str]: """Darf zu ip:port getunnelt werden? Ohne TUNNEL_ALLOW gilt: nur Geraete, die bei der Suche tatsaechlich gefunden wurden. Damit kann der Tunnel nichts ansteuern, was nicht ohnehin im LAN sichtbar ist — und schon gar nichts im Internet. """ if not TUNNEL_ENABLED: return False, "Tunnel ist an diesem Satelliten aus (TUNNEL_ENABLED=false)." if not ip: return False, "Ziel nicht aufloesbar (Name unbekannt?)." if not (0 < port < 65536): return False, f"Port {port} ist ungueltig." if TUNNEL_ALLOW: for rule in TUNNEL_ALLOW: rhost, _, rport = rule.partition(":") if rhost not in ("*", ip): continue if rport and rport != "*" and rport != str(port): continue return True, "" return False, f"{ip}:{port} steht nicht in TUNNEL_ALLOW." known = {d.get("ip") for d in devices if d.get("ip")} if ip in known: return True, "" return False, (f"{ip} wurde im Netz nicht gefunden. Erst 'Geraete suchen', " f"oder {ip} in TUNNEL_ALLOW eintragen.") # ─── Control ──────────────────────────────────────────────────────── def _action_allowed(action: str) -> bool: """Allowlist mit Platzhalter: "fritzbox.*" erlaubt alle fritzbox-Aktionen.""" for a in CONTROL_ALLOWLIST: if a == action or (a.endswith(".*") and action.startswith(a[:-1])): return True return False async def _control(action: str, params: dict, devices: list[dict]) -> dict: """Fuehrt eine Steuer-Aktion aus. Guards: CONTROL_ENABLED + Allowlist.""" if not CONTROL_ENABLED: return {"ok": False, "error": "Steuerung ist an diesem Satelliten deaktiviert (CONTROL_ENABLED=false)."} if not _action_allowed(action): return {"ok": False, "error": f"Aktion '{action}' nicht erlaubt (Allowlist: {', '.join(CONTROL_ALLOWLIST)})."} logger.info("[control] %s params=%s", action, {k: str(v)[:60] for k, v in (params or {}).items()}) loop = asyncio.get_event_loop() try: if action == "dial.launch": return await loop.run_in_executor(None, _do_dial_launch, params, devices) if action == "wol": return await loop.run_in_executor(None, _do_wol, params) if action in ("http.get", "http.post"): return await loop.run_in_executor(None, _do_http, action, params) if action in ("snmp.get", "snmp.walk"): return await loop.run_in_executor(None, _do_snmp, action, params) if action == "snmp.printer": return await loop.run_in_executor(None, _do_snmp_printer, params) if action == "snmp.ports": return await loop.run_in_executor(None, _do_snmp_ports, params) if action == "snmp.info": return await loop.run_in_executor(None, _do_snmp_info, params) if action == "fritzbox.backup_all": return await loop.run_in_executor(None, _fb_backup_all, params, devices) if action.startswith("fritzbox."): # Geraetename ("Fritzbox Keller") → IP, wie bei den anderen Aktionen t = _snmp_target(params) ip = _resolve_target(t, devices) or t extra = {"ip": ip} if t and t != ip and not params.get("name"): extra["name"] = t # z.B. Dateiname der Sicherung: fritzbox-keller_… return await loop.run_in_executor(None, _do_fritzbox, action, {**params, **extra}) if action == "ssh.exec": return await loop.run_in_executor(None, _do_ssh, params) if action == "adb.devices": return await loop.run_in_executor(None, _do_adb_devices, params) if action == "adb.connect": return await loop.run_in_executor(None, _do_adb_connect, params) if action == "adb.install": return await loop.run_in_executor(None, _do_adb_install, params) if action == "adb.shell": return await loop.run_in_executor(None, _do_adb_shell, params) if action == "adb.provision": return await loop.run_in_executor(None, _do_adb_provision, params) if action == "adb.status": return await loop.run_in_executor(None, _do_adb_status, params) if action == "adb.permissions": return await loop.run_in_executor(None, _do_adb_permissions, params) return {"ok": False, "error": f"Aktion '{action}' nicht implementiert."} except Exception as exc: return {"ok": False, "error": f"{action} fehlgeschlagen: {exc}"} def _find_device(devices: list[dict], ref: str) -> Optional[dict]: ref = (ref or "").strip().lower() if not ref: return None for d in devices: if d.get("id", "").lower() == ref or d.get("ip", "") == ref: return d for d in devices: if ref in (d.get("name", "").lower()): return d return None def _do_dial_launch(params: dict, devices: list[dict]) -> dict: """DIAL-App-Launch, z.B. YouTube-Video auf Fire TV / Smart-TV. params: {device, app='YouTube', v= (oder beliebige app-params)}""" import requests ref = params.get("device") or "" dev = _find_device(devices, ref) app_url = (dev or {}).get("dialAppUrl") if dev else params.get("dialAppUrl") if not app_url: return {"ok": False, "error": f"Kein DIAL-Geraet fuer '{ref}' gefunden (oder keine Application-URL)."} app = params.get("app") or "YouTube" # app-Parameter (alles ausser device/app) als form-urlencoded Body. body = {k: v for k, v in (params or {}).items() if k not in ("device", "app", "dialAppUrl")} url = app_url.rstrip("/") + "/" + app r = requests.post(url, data=body, timeout=5) ok = r.status_code in (200, 201) return {"ok": ok, "result": f"DIAL {app} → {(dev or {}).get('name', ref)} (HTTP {r.status_code})" if ok else None, "error": None if ok else f"DIAL-Launch HTTP {r.status_code}: {r.text[:120]}"} def _do_wol(params: dict) -> dict: mac = (params.get("mac") or "").strip() if not re.match(r"^([0-9A-Fa-f]{2}[:-]){5}[0-9A-Fa-f]{2}$", mac): return {"ok": False, "error": f"Ungueltige MAC: {mac!r}"} clean = re.sub(r"[:-]", "", mac) packet = b"\xff" * 6 + bytes.fromhex(clean) * 16 s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) s.sendto(packet, ("255.255.255.255", 9)) s.close() return {"ok": True, "result": f"Wake-on-LAN an {mac} gesendet."} def _window_text(text: str, params: dict) -> dict: """Schneidet grosse Text-Ausgaben zu: optionaler contains-Zeilenfilter, dann offset/max_chars-Fenster. Gibt body + Metadaten (total/returned/truncated).""" total = len(text) contains = params.get("contains") if contains: terms = [contains] if isinstance(contains, str) else list(contains) terms = [str(t).lower() for t in terms if str(t).strip()] if terms: text = "\n".join(ln for ln in text.splitlines() if any(t in ln.lower() for t in terms)) try: offset = max(0, int(params.get("offset", 0))) except (TypeError, ValueError): offset = 0 try: max_chars = int(params.get("max_chars", HTTP_MAX_CHARS)) except (TypeError, ValueError): max_chars = HTTP_MAX_CHARS max_chars = max(1, min(max_chars, HTTP_MAX_CHARS_HARD)) body = text[offset:offset + max_chars] return {"body": body, "total_chars": total, "filtered": bool(contains), "offset": offset, "returned_chars": len(body), "truncated": offset + len(body) < len(text)} def _do_http(action: str, params: dict) -> dict: """HTTP-GET/POST vom Satelliten aus (lokale Webhooks, Geraete-Statusseiten …). params: url Pflicht (http/https). body/headers optional (POST). offset ab welchem Zeichen der Body zurueckgegeben wird (Default 0). max_chars wie viele Zeichen max. (Default HTTP_MAX_CHARS, hart gedeckelt). contains String oder Liste: nur Zeilen, die (case-insensitive) einen der Begriffe enthalten, werden zurueckgegeben. Ideal um aus einer grossen Statusseite nur die relevanten Werte (z.B. Tinte) zu ziehen, ohne die ganze Seite zu paginieren. Antwort enthaelt total_chars + truncated, damit der Aufrufer weiss, ob noch mehr da ist.""" import requests url = params.get("url") or "" if not url.startswith(("http://", "https://")): return {"ok": False, "error": "url (http/https) erforderlich."} method = "GET" if action == "http.get" else "POST" # HTTP-Basic-Auth: explizite params > gespeicherte http-Creds fuer den Host. auth = None hcreds = {} try: from urllib.parse import urlparse host = urlparse(url).hostname or "" hcreds = _creds_for(host).get("http", {}) except Exception: pass user = params.get("user") or hcreds.get("user") pw = params.get("pass") or params.get("password") or hcreds.get("pass") if user or pw: # Benutzer optional — manche Geraete nutzen Password-only-Basic-Auth auth = (str(user or ""), str(pw or "")) r = requests.request(method, url, data=params.get("body"), headers=params.get("headers"), auth=auth, timeout=HTTP_TIMEOUT_SEC) return {"ok": True, "result": {"status": r.status_code, **_window_text(r.text, params)}} def _snmp_run(args: list, timeout: float) -> tuple: """Fuehrt ein net-snmp-CLI-Tool aus. Gibt (ok, stdout|fehlertext).""" import subprocess try: r = subprocess.run(args, capture_output=True, text=True, timeout=timeout) except FileNotFoundError: return False, "snmp-Tools fehlen im Container (Paket 'snmp' im Dockerfile)." except subprocess.TimeoutExpired: return False, "SNMP-Timeout — Geraet antwortet nicht (community/version/IP pruefen)." if r.returncode != 0: return False, (r.stderr or r.stdout or "SNMP-Fehler").strip()[:200] return True, r.stdout def _snmp_base_args(params: dict, ip: str = "") -> list: """Version/Community bzw. v3-Auth. Prioritaet: explizite params > gespeicherte Creds fuer die IP > globale Defaults. OHNE -t/-r (haengt der Aufrufer an).""" creds = _creds_for(ip).get("snmp", {}) if ip else {} version = str(params.get("version") or creds.get("version") or SNMP_VERSION) if version == "3": v3 = creds.get("v3", {}) or {} user = str(params.get("user") or v3.get("user") or "") level = str(params.get("level") or v3.get("level") or "authPriv") args = ["-v", "3", "-u", user, "-l", level] ap = params.get("authProto") or v3.get("authProto") ak = params.get("authKey") or v3.get("authKey") pp = params.get("privProto") or v3.get("privProto") pk = params.get("privKey") or v3.get("privKey") if ap and ak: args += ["-a", str(ap), "-A", str(ak)] if pp and pk: args += ["-x", str(pp), "-X", str(pk)] return args community = str(params.get("community") or creds.get("community") or SNMP_COMMUNITY) return ["-v", version, "-c", community] def _snmp_target(params: dict) -> str: return (params.get("ip") or params.get("host") or params.get("device") or "").strip() def _do_snmp(action: str, params: dict) -> dict: """Generisches snmp.get / snmp.walk. params: {ip|host, oid, community?='public', version?='2c'}.""" ip = _snmp_target(params) if not ip: return {"ok": False, "error": "ip/host erforderlich."} oid = str(params.get("oid") or "").strip() if not oid: return {"ok": False, "error": "oid erforderlich (z.B. 1.3.6.1.2.1.1.5.0 fuer sysName)."} tool = "snmpwalk" if action == "snmp.walk" else "snmpget" # -OQ: OID = Wert, ohne Typannotation; numerische OIDs brauchen keine MIB-Files. args = [tool, "-OQ", *_snmp_base_args(params, ip), "-t", "2", "-r", "1", ip, oid] ok, out = _snmp_run(args, SNMP_TIMEOUT_SEC) if not ok: return {"ok": False, "error": out} lines = [ln.strip() for ln in out.splitlines() if ln.strip()] return {"ok": True, "result": {"ip": ip, "oid": oid, "lines": lines[:200]}} def _snmp_walk_values(ip: str, base: list, oid: str) -> list: """snmpwalk -Oqv (nur Werte, in OID-Index-Reihenfolge).""" ok, out = _snmp_run(["snmpwalk", "-Oqv", *base, ip, oid], SNMP_TIMEOUT_SEC) if not ok: return [] return [ln.strip().strip('"') for ln in out.splitlines() if ln.strip()] def _do_snmp_printer(params: dict) -> dict: """Komfort: liest die Verbrauchsmaterialien (Tinte/Toner) aus der Printer-MIB und rechnet Fuellstaende in Prozent. params: {ip|host, community?, version?}.""" ip = _snmp_target(params) if not ip: return {"ok": False, "error": "ip/host erforderlich."} base = [*_snmp_base_args(params, ip), "-t", "2", "-r", "1"] descs = _snmp_walk_values(ip, base, SNMP_SUPPLY_DESC) if not descs: return {"ok": False, "error": "Keine Printer-MIB-Daten (Geraet unterstuetzt kein SNMP, falsche " "community/version, oder es ist kein Drucker)."} lvls = _snmp_walk_values(ip, base, SNMP_SUPPLY_LVL) maxs = _snmp_walk_values(ip, base, SNMP_SUPPLY_MAX) supplies = [] for i, name in enumerate(descs): lvl = _to_int(lvls[i]) if i < len(lvls) else None mx = _to_int(maxs[i]) if i < len(maxs) else None percent = None if lvl is not None and mx and mx > 0 and lvl >= 0: percent = round(lvl / mx * 100) elif lvl == -3: percent = "vorhanden (Stand unbekannt)" # RFC: some remaining elif lvl in (-1, -2): percent = "unbekannt" supplies.append({"name": name, "level": lvl, "max": mx, "percent": percent}) return {"ok": True, "result": {"ip": ip, "supplies": supplies}} # ifTable (RFC 1213) Spalten: _IF_DESCR = "1.3.6.1.2.1.2.2.1.2" _IF_TYPE = "1.3.6.1.2.1.2.2.1.3" _IF_SPEED = "1.3.6.1.2.1.2.2.1.5" _IF_ADMIN = "1.3.6.1.2.1.2.2.1.7" # up(1) down(2) _IF_OPER = "1.3.6.1.2.1.2.2.1.8" # up(1) down(2) ... _IF_ALIAS = "1.3.6.1.2.1.31.1.1.1.18" # ifAlias (ifXTable, optional) def _do_snmp_ports(params: dict) -> dict: """Interface-Uebersicht eines Switches/Routers: welche Ports sind aktiv (Link), welche frei. params: {ip|host, community?/v3?}. ethernetCsmacd(6)=echte Ports; Loopback/VLAN etc. werden als 'other' markiert, nicht als freier Port gezaehlt.""" ip = _snmp_target(params) if not ip: return {"ok": False, "error": "ip/host erforderlich."} base = [*_snmp_base_args(params, ip), "-t", "2", "-r", "1"] descr = _snmp_walk_values(ip, base, _IF_DESCR) if not descr: return {"ok": False, "error": "Keine Interface-Daten (kein SNMP / falsche Credentials / kein Switch)."} types = _snmp_walk_values(ip, base, _IF_TYPE) opers = _snmp_walk_values(ip, base, _IF_OPER) admins = _snmp_walk_values(ip, base, _IF_ADMIN) speeds = _snmp_walk_values(ip, base, _IF_SPEED) aliases = _snmp_walk_values(ip, base, _IF_ALIAS) ports = [] up = down_free = disabled = 0 for i, name in enumerate(descr): itype = _to_int(types[i]) if i < len(types) else None oper = _to_int(opers[i]) if i < len(opers) else None admin = _to_int(admins[i]) if i < len(admins) else None speed = _to_int(speeds[i]) if i < len(speeds) else None is_eth = (itype == 6) # ethernetCsmacd state = ("up" if oper == 1 else "disabled" if admin == 2 else "down") if is_eth: if state == "up": up += 1 elif state == "disabled": disabled += 1 else: down_free += 1 ports.append({ "name": name.strip('"'), "alias": (aliases[i].strip('"') if i < len(aliases) else ""), "physical": is_eth, "state": state, "speedMbps": round(speed / 1_000_000) if speed else None, }) return {"ok": True, "result": { "ip": ip, "summary": {"physical_ports": up + down_free + disabled, "up": up, "free": down_free, "disabled": disabled}, "ports": ports, }} # entPhysicalTable (RFC 4133) — Modell/Serie/Firmware: _ENT_MODEL = "1.3.6.1.2.1.47.1.1.1.1.13" # entPhysicalModelName _ENT_SERIAL = "1.3.6.1.2.1.47.1.1.1.1.11" # entPhysicalSerialNum _ENT_SWREV = "1.3.6.1.2.1.47.1.1.1.1.10" # entPhysicalSoftwareRev _ENT_FWREV = "1.3.6.1.2.1.47.1.1.1.1.9" # entPhysicalFirmwareRev def _do_snmp_info(params: dict) -> dict: """Geraeteinfo: sysName/sysDescr + (falls vorhanden) Modell, Seriennummer, Firmware-/Software-Version aus der Entity-MIB. Sagt die INSTALLIERTE Version — ob ein Update existiert, weiss SNMP nicht (Hersteller-Sache).""" ip = _snmp_target(params) if not ip: return {"ok": False, "error": "ip/host erforderlich."} sysinfo = _snmp_system(ip, str(params.get("community") or ""), str(params.get("version") or "")) base = [*_snmp_base_args(params, ip), "-t", "2", "-r", "1"] def _first(oid): vals = [v for v in _snmp_walk_values(ip, base, oid) if v and "No Such" not in v] return vals[0] if vals else None result = { "ip": ip, "name": (sysinfo or {}).get("name"), "descr": (sysinfo or {}).get("descr"), "location": (sysinfo or {}).get("location"), "uptime": (sysinfo or {}).get("uptime"), "model": _first(_ENT_MODEL), "serial": _first(_ENT_SERIAL), "firmware": _first(_ENT_FWREV) or _first(_ENT_SWREV), } if not any(result[k] for k in ("name", "descr", "model", "firmware")): return {"ok": False, "error": "Kein SNMP / keine verwertbaren Infos."} return {"ok": True, "result": result} # ─── FritzBox (TR-064) ───────────────────────────────────────────── # TR-064 ist SOAP+Digest-Auth — zu fummelig fuer on-the-fly http.post, daher ein # schlanker Reader. Braucht FritzBox-Login (Credential-Store, Typ 'fritzbox'). def _tr064(ip: str, user: str, pw: str, service: str, control: str, action: str, args: Optional[dict] = None, token: str = "") -> dict: """Ein TR-064-SOAP-Call. Gibt {ok, fields|error}. fields = alle -Tags.""" import requests from requests.auth import HTTPDigestAuth body = "".join(f"<{k}>{v}" for k, v in (args or {}).items()) envelope = ( '' '' # Zusatzbestaetigung: der Token aus X_AVM-DE_Auth SetConfig(start) MUSS # bei GetState und beim Wiederholen der Aktion im SOAP-Header stehen # (AVM "TR-064 Support – Authentication", Kap. 6.4) — sonst ordnet die # Box den Tastendruck nicht zu und meldet "stopped". + (f'{token}' if token else '') + '' f'{body}' '' ) url = f"http://{ip}:49000{control}" try: r = requests.post(url, data=envelope.encode("utf-8"), headers={"Content-Type": 'text/xml; charset="utf-8"', "SOAPAction": f"{service}#{action}"}, auth=HTTPDigestAuth(user, pw), timeout=HTTP_TIMEOUT_SEC) except Exception as exc: return {"ok": False, "error": f"TR-064 nicht erreichbar: {exc}"} if r.status_code == 401: return {"ok": False, "error": "TR-064 Auth fehlgeschlagen (FritzBox-Login pruefen)."} if r.status_code != 200: code = re.search(r"(\d+)", r.text or "") desc = re.search(r"(.*?)", r.text or "", re.DOTALL) if code: return {"ok": False, "upnp_error": code.group(1), "error": f"FritzBox lehnt ab: {code.group(1)} {(desc.group(1).strip() if desc else '')}".strip()} return {"ok": False, "error": f"TR-064 HTTP {r.status_code}"} fields = {m.group(1): m.group(2) for m in re.finditer(r"<(New[^>/]+)>(.*?)", r.text, re.DOTALL)} return {"ok": True, "fields": fields} # ─── FritzBox generisch: alles, was die Firmware per TR-064 anbietet ─── # Jede Box beschreibt sich selbst (tr64desc.xml → Dienste, je Dienst eine # SCPD-Datei → Aktionen mit Argumenten). Beides ist ohne Anmeldung lesbar. # Damit muss niemand pro Modell/Firmware Befehle pflegen: fritzbox.describe # zeigt, was DIESE Box kann, fritzbox.call ruft es auf. _FB_CACHE: dict = {} # (ip, "desc"|scpd-url) → (ts, data) _FB_CACHE_SEC = 3600 # Schreibende Aktionen, die besonders weh tun koennen _FB_DANGEROUS = ("Reboot", "FactoryReset", "DoUpdate", "DoManualUpdate", "SetConfigFile", "AddPortMapping", "DeletePortMapping", "AddPinhole", "SetPassword", "SetSecurityKeys", "SetDefaultWEPKeyIndex", "SetUser", "DeleteUser") def _fb_get(ip: str, path: str) -> str: import requests key = (ip, path) hit = _FB_CACHE.get(key) if hit and time.time() - hit[0] < _FB_CACHE_SEC: return hit[1] r = requests.get(f"http://{ip}:49000{path}", timeout=HTTP_TIMEOUT_SEC) r.raise_for_status() _FB_CACHE[key] = (time.time(), r.text) return r.text def _fb_services(ip: str) -> list: """[{type, short, control, scpd}] aus tr64desc.xml.""" xml = _fb_get(ip, "/tr64desc.xml") out = [] for m in re.finditer(r"(.*?)", xml, re.DOTALL): blk = m.group(1) def _t(tag): mm = re.search(rf"<{tag}>(.*?)", blk, re.DOTALL) return mm.group(1).strip() if mm else "" stype = _t("serviceType") if stype: # "urn:dslforum-org:service:WLANConfiguration:3" → "WLANConfiguration:3" short = ":".join(stype.split(":")[-2:]) out.append({"type": stype, "short": short, "control": _t("controlURL"), "scpd": _t("SCPDURL")}) return out def _fb_actions(ip: str, svc: dict) -> list: """[{name, in:[...], out:[...]}] aus der SCPD-Datei eines Dienstes.""" xml = _fb_get(ip, svc["scpd"]) out = [] for m in re.finditer(r"(.*?)", xml, re.DOTALL): blk = m.group(1) name = re.search(r"(.*?)", blk) if not name: continue args_in, args_out = [], [] for a in re.finditer(r"(.*?)", blk, re.DOTALL): an = re.search(r"(.*?)", a.group(1)) ad = re.search(r"(.*?)", a.group(1)) if an: (args_in if (ad and ad.group(1).strip() == "in") else args_out).append(an.group(1).strip()) out.append({"name": name.group(1).strip(), "in": args_in, "out": args_out}) return out def _fb_is_read(action: str) -> bool: """Lesend = Get… (auch X_AVM-DE_Get…). Alles andere aendert etwas.""" base = action[len("X_AVM-DE_"):] if action.startswith("X_AVM-DE_") else action return base.startswith("Get") def _fb_find_service(ip: str, service: str) -> Optional[dict]: """Voller urn, "WLANConfiguration:3" oder nur "WLANConfiguration" (= :1).""" want = (service or "").strip() svcs = _fb_services(ip) for s_ in svcs: if want in (s_["type"], s_["short"]): return s_ if ":" not in want: return next((s_ for s_ in svcs if s_["short"].split(":")[0] == want), None) return None def _fb_describe(ip: str, params: dict) -> dict: flt = str(params.get("filter") or "").strip().lower() try: svcs = _fb_services(ip) except Exception as exc: return {"ok": False, "error": f"Selbstbeschreibung der Box (:49000/tr64desc.xml) nicht lesbar: {exc}"} lines = [] for svc in svcs: try: acts = _fb_actions(ip, svc) except Exception: continue for a in acts: text = f"{svc['short']} {a['name']}" if flt and flt not in text.lower(): continue kind = "liest" if _fb_is_read(a["name"]) else ( "AENDERT, GEFAEHRLICH" if any(d in a["name"] for d in _FB_DANGEROUS) else "aendert") lines.append(f"{svc['short']} · {a['name']}({', '.join(a['in'])})" + (f" → {', '.join(a['out'])}" if a["out"] and flt else "") + f" [{kind}]") if not lines: return {"ok": True, "result": {"ip": ip, "hinweis": f"Keine Aktion passt zu '{flt}'." if flt else "Keine Aktionen gefunden."}} limit = 120 return {"ok": True, "result": { "ip": ip, "count": len(lines), "aktionen": lines[:limit], "hinweis": ("Aufrufen mit fritzbox.call {service, action, args}. Fuer Ausgabefelder 'filter' setzen." + (f" Gekuerzt auf {limit} — mit 'filter' eingrenzen (z.B. 'WLAN', 'Dect', 'Hosts')." if len(lines) > limit else ""))}} # Manche Aktionen (z.B. Einstellungen sichern) verlangt die Box zusaetzlich am # Geraet bestaetigt: Fehler 866 = "zweiter Faktor noetig". Ablauf laut AVM: # X_AVM-DE_Auth SetConfig(start) → jemand drueckt eine Taste an der Box (oder # waehlt den DTMF-Code am Telefon) → GetState meldet "authenticated" → Aktion # erneut ausfuehren. FB_BUTTON_WAIT_SEC = float(os.environ.get("FB_BUTTON_WAIT_SEC", "120") or "120") _FB_2FA_CODES = ("866",) def _fb_second_factor(ip: str, user: str, pw: str) -> dict: """Startet die Bestaetigung am Geraet und wartet bis FB_BUTTON_WAIT_SEC.""" auth = _fb_find_service(ip, "X_AVM-DE_Auth") if not auth: return {"ok": False, "error": "Die Box bietet keine Bestaetigung per TR-064 an (X_AVM-DE_Auth fehlt)."} st = _tr064(ip, user, pw, auth["type"], auth["control"], "SetConfig", {"NewAction": "start"}) if not st.get("ok"): return {"ok": False, "error": f"Bestaetigung nicht startbar: {st.get('error')}"} methods = st["fields"].get("NewMethods", "") token = st["fields"].get("NewToken", "") logger.info("[fritzbox] %s: warte bis %.0fs auf Bestaetigung am Geraet (%s)", ip, FB_BUTTON_WAIT_SEC, methods) deadline = time.time() + FB_BUTTON_WAIT_SEC state = st["fields"].get("NewState", "") reasons = {"stopped": "die Box hat den Vorgang beendet, ohne Bestaetigung (Zeit abgelaufen?)", "blocked": "zu viele Versuche — die Box sperrt bis zu 60 Minuten", "anotherauthprocess": "es laeuft schon eine Bestaetigung fuer einen anderen Benutzer", "failure": "interner Fehler der Box", "disabled": "die Zusatzbestaetigung ist aus"} while time.time() < deadline: if state == "authenticated": logger.info("[fritzbox] %s: bestaetigt", ip) return {"ok": True, "token": token} if state in reasons and state != "disabled": return {"ok": False, "error": f"Bestaetigung nicht erfolgt — Box meldet '{state}': {reasons[state]}."} time.sleep(2) g = _tr064(ip, user, pw, auth["type"], auth["control"], "GetState", token=token) new = g.get("fields", {}).get("NewState", state) if g.get("ok") else state if new != state: logger.info("[fritzbox] %s: Bestaetigung %s → %s", ip, state, new) state = new _tr064(ip, user, pw, auth["type"], auth["control"], "SetConfig", {"NewAction": "stop"}, token=token) return {"ok": False, "timeout": True, "error": f"Timeout: Die Taste an der Box wurde nicht innerhalb von {FB_BUTTON_WAIT_SEC:.0f} s " f"gedrueckt — abgebrochen, nichts ausgefuehrt." + (f" (Moeglich waeren: {methods})" if methods else "")} def _fb_call(ip: str, user: str, pw: str, params: dict) -> dict: action = str(params.get("action") or "").strip() svc = _fb_find_service(ip, str(params.get("service") or "")) if not action or not svc: return {"ok": False, "error": "service + action noetig (siehe fritzbox.describe), z.B. " "service='WLANConfiguration:3', action='GetInfo'."} if not _fb_is_read(action) and params.get("confirmed") not in (True, "true"): danger = any(d in action for d in _FB_DANGEROUS) return {"ok": False, "needs_confirmation": True, "error": ( f"'{svc['short']} {action}' AENDERT etwas an der FritzBox" + (" und ist GEFAEHRLICH (Neustart/Freigabe/Passwort o.ae.)" if danger else "") + ". Erst Stefan fragen, was genau passiert; nur nach seinem ausdruecklichen Ja " "erneut mit confirmed=true aufrufen.")} args = params.get("args") if isinstance(params.get("args"), dict) else {} args = {(k if k.startswith("New") else "New" + k): v for k, v in args.items()} hidden = [k for k, v in args.items() if "(verborgen)" in str(v)] if hidden: # Sonst setzte die Box den Platzhalter als neuen Schluessel/Kennwort return {"ok": False, "error": ( f"{', '.join(hidden)}: das ist nur der Platzhalter fuer ein verborgenes Geheimnis, " "nicht der echte Wert — so nicht senden. Geheimnisse aendert Stefan selbst " "(FritzBox-Oberflaeche) oder er nennt den neuen Wert ausdruecklich.")} logger.info("[fritzbox] %s %s %s args=%s", ip, svc["short"], action, list(args)) r = _tr064(ip, user, pw, svc["type"], svc["control"], action, args) if not r.get("ok") and r.get("upnp_error") in _FB_2FA_CODES: if params.get("button_wait") not in (True, "true"): return {"ok": False, "needs_button": True, "error": ( f"'{svc['short']} {action}' muss an der FritzBox bestaetigt werden (Taste druecken). " "Frag Stefan, ob er jetzt an der Box steht. Erst nach seinem Ja erneut mit " f"button_wait=true aufrufen — ab dann hat er {FB_BUTTON_WAIT_SEC:.0f} s Zeit, eine Taste " "an der Box zu druecken.")} sf = _fb_second_factor(ip, user, pw) if not sf.get("ok"): return sf # Gleiche Aktion, gleiche Werte — jetzt mit dem Token im SOAP-Header r = _tr064(ip, user, pw, svc["type"], svc["control"], action, args, token=sf.get("token", "")) if not r.get("ok") and r.get("upnp_error") == "867": return {"ok": False, "error": "Die Box sperrt die Bestaetigung (zu viele Versuche) — bis zu 60 Minuten warten."} if not r.get("ok") and r.get("upnp_error") == "868": return {"ok": False, "error": "Die Box ist mit einer anderen Bestaetigung beschaeftigt — in etwa 2 Minuten nochmal."} if not r.get("ok"): return r # Geheimnisse, die die Box auf Lese-Anfragen herausgibt (WLAN-Schluessel, # Kennwoerter, PINs), gehen NICHT an ARIA — ihre Antworten landen beim # Sprachmodell. Stefan sieht sie in der FritzBox-Oberflaeche. fields = {k: ("•••• (verborgen)" if _FB_SECRET_FIELD.search(k) and v else v) for k, v in r["fields"].items()} return {"ok": True, "result": {"ip": ip, "service": svc["short"], "action": action, **fields}} _FB_SECRET_FIELD = re.compile(r"Passphrase|Password|PreSharedKey|WEPKey|KeyPassphrase|Secret|PIN|Passwd|SharedKey", re.IGNORECASE) FB_BACKUP_DIR = os.environ.get("FB_BACKUP_DIR", "/data/fritzbox-backups") FB_BACKUP_KEEP = int(os.environ.get("FB_BACKUP_KEEP", "10") or "10") def _fb_backup_keep(ip: str) -> int: """Wie viele Sicherungen je Box bleiben: Zugangsdaten (Feld "Maximale Sicherungen") > FB_BACKUP_KEEP (Standard 10). Aelteste fliegen zuerst.""" try: n = int(_creds_for(ip).get("fritzbox", {}).get("backupKeep") or 0) except (TypeError, ValueError): n = 0 return max(1, min(100, n)) if n else FB_BACKUP_KEEP def _fb_backup(ip: str, user: str, pw: str, params: dict) -> dict: """Einstellungen der Box sichern (.export) — landet auf dem Satelliten unter FB_BACKUP_DIR, die letzten FB_BACKUP_KEEP je Box bleiben. Kennwort der Sicherung = params.password, sonst das FritzBox-Login (zum Wiederherstellen noetig). Braucht meist die Taste an der Box (button_wait).""" import requests from requests.auth import HTTPDigestAuth # Kennwort der Sicherung: ausdruecklich angegeben > im Satelliten hinterlegt # (Feld "Sicherungs-Kennwort" — ARIA kennt es nicht) > FritzBox-Login. stored = _creds_for(ip).get("fritzbox", {}).get("backupPass") export_pw = str(params.get("password") or stored or pw) weak = _strong_password_problem(export_pw) if weak: src = ("Das angegebene Kennwort" if params.get("password") else "Das hinterlegte Sicherungs-Kennwort" if stored else "Es ist kein Sicherungs-Kennwort hinterlegt, und das FritzBox-Login") return {"ok": False, "error": ( f"{src} ist fuer eine Sicherung zu schwach (es fehlt: {weak}). Stefan soll im " "Diagnostic beim Geraet (🔑) ein Sicherungs-Kennwort hinterlegen: mind. 12 Zeichen, " "Gross-/Kleinbuchstabe, Ziffer, Sonderzeichen.")} r = _fb_call(ip, user, pw, {"service": "DeviceConfig", "action": "X_AVM-DE_GetConfigFile", "args": {"X_AVM-DE_Password": export_pw}, "button_wait": params.get("button_wait")}) if not r.get("ok"): return r url = r["result"].get("NewX_AVM-DE_ConfigFileUrl", "") if not url: return {"ok": False, "error": "Die Box lieferte keine Download-Adresse fuer die Sicherung."} try: d = requests.get(url, auth=HTTPDigestAuth(user, pw), timeout=60, verify=False) d.raise_for_status() except Exception as exc: return {"ok": False, "error": f"Sicherung nicht herunterladbar: {exc}"} os.makedirs(FB_BACKUP_DIR, exist_ok=True) label = _slug(str(params.get("name") or ip)) fname = f"{label}_{time.strftime('%Y-%m-%d_%H%M')}.export" path = os.path.join(FB_BACKUP_DIR, fname) with open(path, "wb") as fh: fh.write(d.content) try: os.chmod(path, 0o600) except OSError: pass info = _tr064(ip, user, pw, "urn:dslforum-org:service:DeviceInfo:1", "/upnp/control/deviceinfo", "GetInfo") fi = info.get("fields", {}) if info.get("ok") else {} meta = {"ip": ip, "name": str(params.get("name") or ip), "created": time.strftime("%Y-%m-%dT%H:%M:%S"), "bytes": len(d.content), "model": fi.get("NewModelName", ""), "firmware": fi.get("NewSoftwareVersion", ""), "serial": fi.get("NewSerialNumber", ""), # nur die QUELLE des Kennworts — zum Wiederherstellen noetig, nie der Wert "kennwort": "angegeben" if params.get("password") else ("backupPass" if stored else "login"), # Das benutzte Kennwort selbst, verschluesselt wie der Zugangsdaten- # Speicher: passt auch noch, wenn das Sicherungs-Kennwort spaeter # geaendert wird (Wiederherstellen + 👁 im Diagnostic). "kennwort_enc": _creds_cipher().encrypt(export_pw.encode()).decode()} with open(path[:-len(".export")] + ".json", "w", encoding="utf-8") as fh: json.dump(meta, fh, ensure_ascii=False, indent=1) keep = _fb_backup_keep(ip) old = sorted(f for f in os.listdir(FB_BACKUP_DIR) if f.startswith(label + "_") and f.endswith(".export")) for f in old[:-keep]: for ext in (".export", ".json"): try: os.remove(os.path.join(FB_BACKUP_DIR, f[:-len(".export")] + ext)) except OSError: pass logger.info("[fritzbox] %s: Sicherung %s (%d Bytes)", ip, fname, len(d.content)) import base64 # Die Datei geht mit zurueck — ARIAs Brain legt sie unter /shared/uploads/ # ab, damit sie in den Chat kann. Sie ist mit dem Kennwort verschluesselt. # no_file: das Diagnostic braucht nur die Meldung, die Datei bleibt auf dem Satelliten file_b64 = (base64.b64encode(d.content).decode() if len(d.content) <= 20 * 1024 * 1024 and not params.get("no_file") else "") return {"ok": True, "result": { "ip": ip, "datei": path, "bytes": len(d.content), "auf": "Satellit", **({"file_b64": file_b64, "filename": fname} if file_b64 else {}), "verschluesselt_mit": ("das angegebene Kennwort" if params.get("password") else "das im Satelliten hinterlegte Sicherungs-Kennwort" if stored else "das FritzBox-Login-Kennwort"), "hinweis": f"Liegt auf dem Satelliten, die letzten {keep} je Box bleiben. " "Zum Wiederherstellen in der FritzBox: System → Sicherung → Wiederherstellen, mit dem Kennwort."}} def _fb_wlan(ip: str, user: str, pw: str, params: dict) -> dict: """Alle WLANs der Box (2,4/5/6 GHz, Gast) mit SSID und an/aus; mit index + enable schalten (braucht confirmed=true).""" wl = sorted((s_ for s_ in _fb_services(ip) if s_["short"].startswith("WLANConfiguration:")), key=lambda s_: int(s_["short"].split(":")[1])) if "enable" in params and params.get("index") is not None: target = next((s_ for s_ in wl if s_["short"] == f"WLANConfiguration:{params['index']}"), None) if not target: return {"ok": False, "error": f"WLAN {params['index']} gibt es nicht (vorhanden: 1–{len(wl)})."} on = str(params.get("enable")).lower() in ("1", "true", "an", "on", "ja") return _fb_call(ip, user, pw, {"service": target["short"], "action": "SetEnable", "args": {"Enable": 1 if on else 0}, "confirmed": params.get("confirmed")}) out = [] for i, s_ in enumerate(wl, 1): r = _tr064(ip, user, pw, s_["type"], s_["control"], "GetInfo") f = r.get("fields", {}) if r.get("ok") else {} w = {"index": int(s_["short"].split(":")[1]), "ssid": f.get("NewSSID", ""), "an": f.get("NewEnable") == "1", "status": f.get("NewStatus", ""), "kanal": f.get("NewChannel", ""), "standard": f.get("NewStandard", ""), "band": f.get("NewX_AVM-DE_FrequencyBand", "")} if not r.get("ok"): w["fehler"] = r.get("error") out.append({k: v for k, v in w.items() if v not in ("", None)}) return {"ok": True, "result": {"ip": ip, "wlans": out, "hinweis": "Das letzte WLAN ist meist das Gast-WLAN. Schalten: index + enable (+ confirmed=true nach Stefans Ja)."}} # ─── Sicherungen verwalten: Liste, Ansicht, Wiederherstellen, Loeschen ─── def _fb_backup_path(name: str) -> str: """Nur Dateien aus FB_BACKUP_DIR — kein Pfad von aussen.""" base = os.path.basename(str(name or "")) if not base.endswith(".export"): base += ".export" path = os.path.join(FB_BACKUP_DIR, base) return path if os.path.isfile(path) else "" def _fb_backup_meta(path: str, with_secret: bool = False) -> dict: try: with open(path[:-len(".export")] + ".json", encoding="utf-8") as fh: m = json.load(fh) except Exception: return {} if not with_secret: m.pop("kennwort_enc", None) return m def _fb_backup_password(ip: str, path: str) -> tuple: """(kennwort, hinweis) fuer genau diese Sicherung. Aeltere Sicherungen ohne gespeichertes Kennwort: das AKTUELL hinterlegte der Quelle (mit Hinweis).""" meta = _fb_backup_meta(path, with_secret=True) if meta.get("kennwort_enc"): try: return _creds_cipher().decrypt(meta["kennwort_enc"].encode()).decode(), "" except Exception: pass fb = _creds_for(ip).get("fritzbox", {}) src = meta.get("kennwort", "") if src == "backupPass" and fb.get("backupPass"): return fb["backupPass"], ("Aeltere Sicherung: das ist das AKTUELL hinterlegte Sicherungs-Kennwort — " "wurde es seitdem geaendert, passt es nicht.") if src == "login" and fb.get("pass"): return fb["pass"], "Aeltere Sicherung: das ist das aktuelle FritzBox-Kennwort." return "", "Fuer diese Sicherung ist kein Kennwort bekannt (eigens genanntes Kennwort)." def _fb_backup_pw_action(ip: str, params: dict) -> dict: path = _fb_backup_path(params.get("file") or params.get("datei") or "") if not path: return {"ok": False, "error": "Sicherung nicht gefunden."} pw, hint = _fb_backup_password(ip, path) if not pw: return {"ok": False, "error": hint} return {"ok": True, "result": {"datei": os.path.basename(path), "kennwort": pw, "hinweis": hint}} def _fb_backups(ip: str, params: dict) -> dict: label = _slug(str(params.get("name") or "")) out = [] for f in sorted(os.listdir(FB_BACKUP_DIR) if os.path.isdir(FB_BACKUP_DIR) else [], reverse=True): if not f.endswith(".export"): continue m = _fb_backup_meta(os.path.join(FB_BACKUP_DIR, f)) if m.get("ip") != ip and not (label and f.startswith(label + "_")): continue out.append({"datei": f, "erstellt": m.get("created", ""), "bytes": m.get("bytes") or os.path.getsize(os.path.join(FB_BACKUP_DIR, f)), "modell": m.get("model", ""), "firmware": m.get("firmware", ""), "kennwort": m.get("kennwort", "")}) return {"ok": True, "result": {"ip": ip, "count": len(out), "sicherungen": out, "behalten": _fb_backup_keep(ip)}} # Lesbare Hinweise aus der Sicherung: AVM-Konfig "schluessel = wert;" _FB_VIEW_KEYS = { "hostname": "Name der Box", "ssid": "WLAN-Name (2,4 GHz)", "ssid_scnd": "WLAN-Name (5 GHz)", "guest_ssid": "Gast-WLAN-Name", "channel": "WLAN-Kanal", "ipaddr": "IP-Adresse", "netmask": "Netzmaske", "dhcpstart": "DHCP von", "dhcpend": "DHCP bis", "username": "Benutzername", "registrar": "SIP-Registrar", "msn": "Rufnummer", "dnsserver1": "DNS-Server 1", "dnsserver2": "DNS-Server 2", "ntp_server": "Zeitserver", } def _fb_backup_view(params: dict) -> dict: path = _fb_backup_path(params.get("file") or params.get("datei") or "") if not path: return {"ok": False, "error": "Sicherung nicht gefunden (Dateiname aus fritzbox.backups)."} text = open(path, encoding="utf-8", errors="replace").read() header, sections, binfiles = {}, {}, [] cur, buf = None, [] for line in text.splitlines(): m = re.match(r"\*{4} (CFGFILE|BINFILE|B64FILE):(\S+)", line) if m: if cur: sections[cur] = "\n".join(buf) cur, buf = (m.group(2) if m.group(1) == "CFGFILE" else None), [] if m.group(1) != "CFGFILE": binfiles.append(m.group(2)) continue if line.startswith("**** END OF FILE"): if cur: sections[cur] = "\n".join(buf) cur, buf = None, [] continue if cur is not None: buf.append(line) elif not sections and not binfiles and "=" in line and not line.startswith("****"): k, v = line.split("=", 1) header[k.strip()] = "«verschluesselt»" if v.strip().startswith("$$$$") else v.strip() secret = re.compile(r"\$\$\$\$\S+") highlights = [] for name, body in sections.items(): for m in re.finditer(r"^\s*(\w+)\s*=\s*\"?([^;\"]*)\"?\s*;", body, re.MULTILINE): key, val = m.group(1), m.group(2).strip() if key in _FB_VIEW_KEYS and val and len(highlights) < 80: highlights.append({"was": _FB_VIEW_KEYS[key], "wert": "«verschluesselt»" if val.startswith("$$$$") else val, "datei": name, "schluessel": key}) res = {"datei": os.path.basename(path), "meta": _fb_backup_meta(path), "kopf": header, "abschnitte": sorted(sections), "anhaenge": binfiles, "wichtig": highlights} if params.get("full"): budget, full = 900_000, {} for name, body in sections.items(): body = secret.sub("«verschluesselt»", body) full[name] = body[:max(0, budget)] budget -= len(body) res["inhalt"] = full return {"ok": True, "result": res} def _fb_serve_once(path: str, seconds: float = 600): """Stellt EINE Datei kurz per HTTP bereit (die Box holt sich die Sicherung beim Wiederherstellen selbst ab). Zufaelliger Pfad, endet nach dem ersten Abruf oder nach `seconds`. Gibt (url, served_event).""" import http.server import secrets as _secrets token = _secrets.token_urlsafe(16) fname = os.path.basename(path) served = threading.Event() class H(http.server.BaseHTTPRequestHandler): def log_message(self, *a): pass def do_GET(self): if self.path != f"/{token}/{fname}": self.send_response(404) self.end_headers() return data = open(path, "rb").read() self.send_response(200) self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) served.set() srv = http.server.ThreadingHTTPServer(("0.0.0.0", 0), H) threading.Thread(target=srv.serve_forever, daemon=True).start() def _stop(): served.wait(seconds) time.sleep(5) srv.shutdown() threading.Thread(target=_stop, daemon=True).start() host = (NET or {}).get("primary_ip") or "127.0.0.1" return f"http://{host}:{srv.server_address[1]}/{token}/{fname}", served def _fb_restore(ip: str, user: str, pw: str, params: dict) -> dict: path = _fb_backup_path(params.get("file") or params.get("datei") or "") if not path: return {"ok": False, "error": "Sicherung nicht gefunden (Dateiname aus fritzbox.backups)."} meta = _fb_backup_meta(path) if meta.get("ip") and meta["ip"] != ip and not params.get("other_box"): return {"ok": False, "error": (f"Diese Sicherung stammt von {meta.get('name') or meta['ip']} ({meta.get('model', '?')}), " f"nicht von {ip}. Auf eine andere Box nur mit other_box=true einspielen.")} if params.get("confirmed") not in (True, "true"): return {"ok": False, "needs_confirmation": True, "error": ( f"Wiederherstellen von {os.path.basename(path)} ist GEFAEHRLICH: die Box uebernimmt ALLE " "Einstellungen dieser Sicherung und startet neu (1–3 Minuten kein Netz/Internet). Erst " "Stefan fragen; nur nach seinem ausdruecklichen Ja erneut mit confirmed=true.")} src = meta.get("kennwort", "") stored = _creds_for(ip).get("fritzbox", {}).get("backupPass") saved_pw, _hint = _fb_backup_password(ip, path) if params.get("password"): export_pw = str(params["password"]) elif saved_pw and _fb_backup_meta(path, with_secret=True).get("kennwort_enc"): export_pw = saved_pw # genau das Kennwort dieser Sicherung elif src == "angegeben": return {"ok": False, "error": "Diese Sicherung wurde mit einem eigens genannten Kennwort erstellt — " "Stefan muss es fuer das Wiederherstellen angeben (password)."} elif src == "backupPass": if not stored: return {"ok": False, "error": "Fuer diese Sicherung ist kein Sicherungs-Kennwort mehr hinterlegt."} export_pw = stored else: export_pw = pw url, served = _fb_serve_once(path) r = _fb_call(ip, user, pw, {"service": "DeviceConfig", "action": "X_AVM-DE_SetConfigFile", "args": {"X_AVM-DE_Password": export_pw, "X_AVM-DE_ConfigFileUrl": url}, "confirmed": True, "button_wait": params.get("button_wait")}) if not r.get("ok"): return r got = served.wait(30) logger.info("[fritzbox] %s: Wiederherstellen %s (abgeholt: %s)", ip, os.path.basename(path), got) return {"ok": True, "result": { "ip": ip, "datei": os.path.basename(path), "abgeholt": got, "hinweis": ("Die Box hat die Sicherung abgeholt, uebernimmt die Einstellungen und startet neu — " "1–3 Minuten nicht erreichbar." if got else "Die Box hat den Auftrag angenommen, die Datei aber (noch) nicht abgeholt. Erreicht sie " "den Satelliten? In ein paar Minuten pruefen, ob die Einstellungen uebernommen sind.")}} def _fb_reachable(ip: str, timeout: float = 3.0) -> bool: try: with socket.create_connection((ip, 49000), timeout=timeout): return True except OSError: return False def _fb_backup_all(params: dict, devices: list) -> dict: """Sichert ALLE FritzBoxen mit hinterlegtem Login, eine nach der anderen. Nicht erreichbar / Fehler / Taste noetig → notieren und WEITER, nie abbrechen.""" names = {d.get("ip"): d.get("name") for d in devices if d.get("ip")} boxes = [ip for ip, e in sorted(_CREDS.items()) if (e.get("fritzbox") or {}).get("pass")] if not boxes: return {"ok": False, "error": "An diesem Satelliten ist fuer keine FritzBox ein Login hinterlegt."} report = [] for ip in boxes: name = names.get(ip) or ip if not _fb_reachable(ip): report.append({"box": name, "ip": ip, "ok": False, "grund": "nicht erreichbar (aus/weg?) — uebersprungen"}) continue fb = _creds_for(ip).get("fritzbox", {}) try: r = _fb_backup(ip, str(fb.get("user") or ""), str(fb.get("pass") or ""), {**params, "name": name, "no_file": True}) except Exception as exc: r = {"ok": False, "error": f"Fehler: {exc}"} if r.get("ok"): report.append({"box": name, "ip": ip, "ok": True, "datei": os.path.basename(r["result"]["datei"])}) else: grund = ("braucht eine Taste an der Box — uebersprungen (mit button_wait einzeln nachholen)" if r.get("needs_button") else r.get("error") or "Fehler") report.append({"box": name, "ip": ip, "ok": False, "grund": grund}) ok = sum(1 for x in report if x["ok"]) return {"ok": True, "result": {"gesichert": ok, "von": len(report), "boxen": report}} # ─── Zusatzbestaetigung (Taste) der Box: Status + Umschalten ─── def _fb_second_factor_info(ip: str, user: str, pw: str) -> dict: """Ist die Tastenbestaetigung an? Und laesst sie sich per TR-064 schalten? Umschalten geht nur, wenn die Firmware dafuer eine Aktion anbietet (eine Set-Aktion des Auth-Dienstes mit einem Enable-Argument) — sonst nur in der FritzBox-Oberflaeche.""" auth = _fb_find_service(ip, "X_AVM-DE_Auth") if not auth: return {"ok": True, "result": {"vorhanden": False, "an": False, "schaltbar": False, "hinweis": "Die Box kennt keine Zusatzbestaetigung per Schnittstelle — Aktionen laufen ohne Taste."}} acts = _fb_actions(ip, auth) setter = next((a for a in acts if not _fb_is_read(a["name"]) and a["name"] != "SetConfig" and any("enable" in x.lower() for x in a["in"])), None) info = _tr064(ip, user, pw, auth["type"], auth["control"], "GetInfo") \ if any(a["name"] == "GetInfo" for a in acts) else {"ok": False} f = info.get("fields", {}) if info.get("ok") else {} en = f.get("NewEnabled", f.get("NewEnable")) return {"ok": True, "result": { "vorhanden": True, "an": None if en is None else en in ("1", "true"), "schaltbar": bool(setter), "aktion": setter["name"] if setter else "", "hinweis": ("" if setter else "Laesst sich ueber die Schnittstelle nicht umschalten (Firmware bietet es nicht an). " "Nur in der FritzBox-Oberflaeche: System → FRITZ!Box-Benutzer → Anmeldung im Heimnetz → " "\"Ausfuehrung bestimmter Einstellungen und Funktionen zusaetzlich bestaetigen\". " "Manche Firmware erlaubt das Abschalten gar nicht mehr.")}} def _fb_second_factor_set(ip: str, user: str, pw: str, params: dict) -> dict: inf = _fb_second_factor_info(ip, user, pw)["result"] if not inf["vorhanden"]: return {"ok": False, "error": inf["hinweis"]} if not inf["schaltbar"]: return {"ok": False, "not_switchable": True, "error": inf["hinweis"]} auth = _fb_find_service(ip, "X_AVM-DE_Auth") act = next(a for a in _fb_actions(ip, auth) if a["name"] == inf["aktion"]) arg = next(x for x in act["in"] if "enable" in x.lower()) on = str(params.get("enable")).lower() in ("1", "true", "an", "on", "ja") return _fb_call(ip, user, pw, {"service": auth["short"], "action": act["name"], "args": {arg: 1 if on else 0}, "confirmed": params.get("confirmed"), "button_wait": params.get("button_wait")}) def _fb_backup_get(params: dict) -> dict: """Eine vorhandene Sicherung als Datei zurueckgeben (Download im Diagnostic, bei ARIA landet sie unter /shared/uploads/ fuer den Chat).""" path = _fb_backup_path(params.get("file") or params.get("datei") or "") if not path: return {"ok": False, "error": "Sicherung nicht gefunden (Dateiname aus fritzbox.backups)."} data = open(path, "rb").read() if len(data) > 20 * 1024 * 1024: return {"ok": False, "error": "Sicherung zu gross fuer die Uebertragung (> 20 MB)."} return {"ok": True, "result": {"datei": os.path.basename(path), "bytes": len(data), "filename": os.path.basename(path), "file_b64": base64.b64encode(data).decode()}} def _fb_backup_delete(params: dict) -> dict: path = _fb_backup_path(params.get("file") or params.get("datei") or "") if not path: return {"ok": False, "error": "Sicherung nicht gefunden."} if params.get("confirmed") not in (True, "true"): return {"ok": False, "needs_confirmation": True, "error": f"{os.path.basename(path)} endgueltig loeschen? Erst Stefan fragen, dann confirmed=true."} for ext in (".export", ".json"): try: os.remove(path[:-len(".export")] + ext) except OSError: pass return {"ok": True, "result": {"geloescht": os.path.basename(path)}} def _do_fritzbox(action: str, params: dict) -> dict: """fritzbox.info -> Modell/Firmware/Verbindung/externe IP/Datenrate. fritzbox.hosts -> Liste der bekannten Geraete (Name/IP/MAC/aktiv) plus Verbindungsart (LAN/WLAN), LAN-Port und Geschwindigkeit. params.find: nur Geraete, deren Name/IP/MAC das enthaelt.""" ip = _snmp_target(params) if not ip: return {"ok": False, "error": "ip/host erforderlich."} if action == "fritzbox.describe": # braucht keine Anmeldung return _fb_describe(ip, params) if action == "fritzbox.backups": return _fb_backups(ip, params) if action == "fritzbox.backup_view": return _fb_backup_view(params) if action == "fritzbox.backup_delete": return _fb_backup_delete(params) if action == "fritzbox.backup_get": return _fb_backup_get(params) if action == "fritzbox.backup_password": # nur fuers Diagnostic (Brain sperrt es) return _fb_backup_pw_action(ip, params) fb = _creds_for(ip).get("fritzbox", {}) user = str(params.get("user") or fb.get("user") or "") pw = str(params.get("pass") or params.get("password") or fb.get("pass") or "") if not pw: return {"ok": False, "error": "Kein FritzBox-Login hinterlegt. In der Geraeteliste Credentials " "(Typ 'fritzbox') fuer diese IP setzen."} if action == "fritzbox.call": return _fb_call(ip, user, pw, params) if action == "fritzbox.wlan": return _fb_wlan(ip, user, pw, params) if action == "fritzbox.backup": return _fb_backup(ip, user, pw, params) if action == "fritzbox.restore": return _fb_restore(ip, user, pw, params) if action == "fritzbox.second_factor": return _fb_second_factor_info(ip, user, pw) if action == "fritzbox.second_factor_set": return _fb_second_factor_set(ip, user, pw, params) if action == "fritzbox.hosts": p = _tr064(ip, user, pw, "urn:dslforum-org:service:Hosts:1", "/upnp/control/hosts", "X_AVM-DE_GetHostListPath") if not p.get("ok"): return p path = p["fields"].get("NewX_AVM-DE_HostListPath", "") if not path: return {"ok": False, "error": "FritzBox lieferte keinen Host-Listen-Pfad."} import requests from requests.auth import HTTPDigestAuth try: r = requests.get(f"http://{ip}:49000{path}", auth=HTTPDigestAuth(user, pw), timeout=HTTP_TIMEOUT_SEC) except Exception as exc: return {"ok": False, "error": f"Host-Liste nicht abrufbar: {exc}"} hosts = [] find = str(params.get("find") or "").strip().lower() for item in re.finditer(r"(.*?)", r.text, re.DOTALL): blk = item.group(1) def _t(tag): m = re.search(rf"<{tag}>(.*?)", blk, re.DOTALL) return m.group(1).strip() if m else "" h = {"name": _t("HostName"), "ip": _t("IPAddress"), "mac": _t("MACAddress"), "active": _t("Active") in ("1", "true")} # AVM-Zusatzfelder (je nach FRITZ!OS vorhanden): Verbindungsart, # LAN-Port der Box (0 = kein direkter LAN-Port, z.B. WLAN/Mesh), # Geschwindigkeit in Mbit/s, Gast-Netz. iface = _t("InterfaceType") if iface: h["interface"] = iface port = _to_int(_t("X_AVM-DE_Port")) if port: h["lan_port"] = port speed = _to_int(_t("X_AVM-DE_Speed")) if speed: h["speed_mbit"] = speed if _t("X_AVM-DE_Guest") in ("1", "true"): h["guest"] = True if find and not any(find in str(h.get(k, "")).lower() for k in ("name", "ip", "mac")): continue hosts.append(h) return {"ok": True, "result": {"ip": ip, "count": len(hosts), "hosts": hosts, "hinweis": "lan_port = LAN-Buchse DIESER Box. Haengt ein Geraet " "hinter einem Switch/Repeater, steht hier dessen Port."}} # fritzbox.info (Default): mehrere Services, Teil-Fehler tolerieren. info = {"ip": ip} dev = _tr064(ip, user, pw, "urn:dslforum-org:service:DeviceInfo:1", "/upnp/control/deviceinfo", "GetInfo") if dev.get("ok"): f = dev["fields"] info.update({"model": f.get("NewModelName"), "firmware": f.get("NewSoftwareVersion"), "serial": f.get("NewSerialNumber"), "uptime_s": _to_int(f.get("NewUpTime"))}) st = _tr064(ip, user, pw, "urn:dslforum-org:service:WANIPConnection:1", "/upnp/control/wanipconnection1", "GetStatusInfo") if st.get("ok"): info["connection"] = st["fields"].get("NewConnectionStatus") info["connection_uptime_s"] = _to_int(st["fields"].get("NewUptime")) ext = _tr064(ip, user, pw, "urn:dslforum-org:service:WANIPConnection:1", "/upnp/control/wanipconnection1", "GetExternalIPAddress") if ext.get("ok"): info["external_ip"] = ext["fields"].get("NewExternalIPAddress") link = _tr064(ip, user, pw, "urn:dslforum-org:service:WANCommonInterfaceConfig:1", "/upnp/control/wancommonifconfig1", "GetCommonLinkProperties") if link.get("ok"): f = link["fields"] dn = _to_int(f.get("NewLayer1DownstreamMaxBitRate")) upr = _to_int(f.get("NewLayer1UpstreamMaxBitRate")) info["downstream_mbit"] = round(dn / 1_000_000, 1) if dn else None info["upstream_mbit"] = round(upr / 1_000_000, 1) if upr else None info["physical_link"] = f.get("NewPhysicalLinkStatus") if len(info) == 1: return {"ok": False, "error": "FritzBox antwortet nicht auf TR-064 (Login/Rechte pruefen; TR-064 in " "der FritzBox unter Heimnetz > Netzwerkeinstellungen aktivieren)."} return {"ok": True, "result": info} # ─── SSH ─────────────────────────────────────────────────────────── def _ssh_load_key(key_str: str): """Laedt einen privaten Schluessel aus einem String (RSA/Ed25519/ECDSA/DSS).""" import io import paramiko for cls in (paramiko.Ed25519Key, paramiko.RSAKey, paramiko.ECDSAKey, paramiko.DSSKey): try: return cls.from_private_key(io.StringIO(key_str)) except Exception: continue return None def _do_ssh(params: dict) -> dict: """Fuehrt EIN Kommando per SSH auf einem Geraet aus. params: {ip|host, cmd, user?, pass?, key?, port?}. Auth bevorzugt aus dem Credential-Store (Typ 'ssh'). Grosse Ausgaben werden wie bei http.get gefenstert (contains/offset/max_chars).""" import paramiko host = _snmp_target(params) if not host: return {"ok": False, "error": "ip/host erforderlich."} cmd = params.get("cmd") or params.get("command") or "" if not cmd: return {"ok": False, "error": "cmd (Kommando) erforderlich."} creds = _creds_for(host).get("ssh", {}) user = str(params.get("user") or creds.get("user") or "") if not user: return {"ok": False, "error": "SSH-Benutzer fehlt (Credentials fuer diese IP setzen)."} port = _to_int(params.get("port") or creds.get("port")) or 22 password = params.get("pass") or params.get("password") or creds.get("pass") key_str = params.get("key") or creds.get("key") client = paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) try: kwargs = {"hostname": host, "port": port, "username": user, "timeout": SSH_TIMEOUT_SEC, "banner_timeout": SSH_TIMEOUT_SEC, "auth_timeout": SSH_TIMEOUT_SEC, "allow_agent": False, "look_for_keys": False} if key_str: pkey = _ssh_load_key(str(key_str)) if pkey is None: return {"ok": False, "error": "Privater SSH-Key nicht lesbar (Format?)."} kwargs["pkey"] = pkey if password: kwargs["password"] = str(password) client.connect(**kwargs) stdin, stdout, stderr = client.exec_command(cmd, timeout=SSH_TIMEOUT_SEC) exit_code = stdout.channel.recv_exit_status() out = stdout.read().decode("utf-8", "ignore") err = stderr.read().decode("utf-8", "ignore") except paramiko.AuthenticationException: return {"ok": False, "error": "SSH-Auth fehlgeschlagen (User/Passwort/Key pruefen)."} except Exception as exc: return {"ok": False, "error": f"SSH fehlgeschlagen: {exc}"} finally: try: client.close() except Exception: pass win = _window_text(out, params) return {"ok": True, "result": {"host": host, "exit_code": exit_code, "stderr": err[:4000], **win}} # ─── ADB: Android-Geraete im LAN (Fire TV Sticks) ─────────────────── def _adb_run(args: list, timeout: float = 0) -> tuple: """Fuehrt adb aus. Gibt (rc, stdout, stderr). Muster wie _snmp_run.""" import subprocess try: r = subprocess.run(["adb"] + args, capture_output=True, text=True, timeout=timeout or ADB_TIMEOUT_SEC) return r.returncode, (r.stdout or "").strip(), (r.stderr or "").strip() except FileNotFoundError: return 127, "", ("adb fehlt im Container — 'adb' ins satellite/Dockerfile " "aufnehmen und neu bauen.") except subprocess.TimeoutExpired: return 124, "", f"adb-Timeout nach {timeout or ADB_TIMEOUT_SEC:.0f}s." def _adb_target(params: dict) -> str: """Ziel als ip:port (Port default 5555).""" host = _snmp_target(params) if not host: return "" return host if ":" in host else f"{host}:{int(params.get('port') or ADB_PORT)}" def _adb_connect(target: str) -> tuple: """Verbindet zum Geraet. Gibt (ok, meldung).""" rc, out, err = _adb_run(["connect", target], timeout=25) txt = f"{out} {err}".strip() low = txt.lower() if "connected to" in low: # connected / already connected return True, txt if "unauthorized" in low: return False, ("Geraet hat die Verbindung nicht autorisiert. Auf dem " "Bildschirm des Geraets den Dialog 'USB-Debugging zulassen?' " "bestaetigen (Haken bei 'Immer erlauben').") if rc == 127 or "adb fehlt" in low: return False, txt return False, (txt or "Verbindung fehlgeschlagen") + \ " — ADB-Debugging aktiv? (Fire TV: Einstellungen > Mein Fire TV > " \ "Entwickleroptionen > ADB-Debugging)" def _adb_state(target: str) -> str: """'device' | 'unauthorized' | 'offline' | '' (unbekannt).""" rc, out, _ = _adb_run(["devices"], timeout=20) for line in out.splitlines()[1:]: parts = line.split() if len(parts) >= 2 and parts[0] == target: return parts[1] return "" def _adb_wait_authorized(target: str, seconds: float) -> tuple: """Wartet, bis das Geraet autorisiert ist ('device'). Beim ersten Kontakt erscheint auf dem Fernseher/Handy der Dialog "USB-Debugging zulassen?". Frueher brach die Installation hier ab und man musste nach dem Bestaetigen ein zweites Mal starten — jetzt wartet der Satellit einfach, bis der Nutzer bestaetigt hat. Gibt (ok, zustand). """ import time as _time deadline = _time.time() + max(0.0, seconds) last = _adb_state(target) while _time.time() < deadline: if last == "device": return True, last _time.sleep(2.0) # Erneut verbinden: nach dem Bestaetigen wechselt der Zustand oft erst # mit einem frischen connect von 'unauthorized'/'offline' auf 'device'. _adb_run(["connect", target], timeout=15) last = _adb_state(target) return last == "device", last def _do_adb_devices(params: dict) -> dict: """Listet die dem Satelliten bekannten ADB-Geraete.""" rc, out, err = _adb_run(["devices", "-l"], timeout=20) if rc not in (0,): return {"ok": False, "error": err or "adb devices fehlgeschlagen"} devices = [] for line in out.splitlines()[1:]: parts = line.split() if len(parts) >= 2: devices.append({"target": parts[0], "state": parts[1], "info": " ".join(parts[2:])}) return {"ok": True, "result": {"devices": devices, "count": len(devices)}} def _do_adb_connect(params: dict) -> dict: target = _adb_target(params) if not target: return {"ok": False, "error": "ip/host erforderlich."} ok, msg = _adb_connect(target) state = _adb_state(target) wait = float(params.get("auth_wait") or ADB_AUTH_WAIT_SEC) if (not ok or state != "device") and wait > 0: authed, state = _adb_wait_authorized(target, wait) if not authed: return {"ok": False, "error": "Nicht autorisiert — auf dem Bildschirm des Geraets " "'USB-Debugging zulassen?' bestaetigen. " f"(gewartet: {wait:.0f}s, Zustand: '{state or '?'}')"} msg = f"verbunden (nach Bestaetigung am Geraet)" elif not ok: return {"ok": False, "error": msg} return {"ok": True, "result": {"target": target, "state": state, "message": msg}} AGENT_PACKAGE = os.environ.get("AGENT_PACKAGE", "de.hackersoft.ariaagent") def _adb_app_installed(target: str, package: str) -> bool: rc, out, _ = _adb_run(["-s", target, "shell", "pm", "list", "packages", package], timeout=30) return rc == 0 and f"package:{package}" in out def _adb_app_running(target: str, package: str) -> bool: """Laeuft der Vordergrund-Dienst des Agents? (Fire TV / Android).""" rc, out, _ = _adb_run(["-s", target, "shell", "dumpsys", "activity", "services", package], timeout=30) return rc == 0 and "ServiceRecord" in out def _do_adb_status(params: dict) -> dict: """Ist der ARIA-Agent auf dem Geraet installiert und laeuft er?""" target = _adb_target(params) if not target: return {"ok": False, "error": "ip/host erforderlich."} package = (params.get("package") or AGENT_PACKAGE).strip() ok, msg = _adb_connect(target) if not ok: return {"ok": False, "error": msg} installed = _adb_app_installed(target, package) running = _adb_app_running(target, package) if installed else False ver = "" if installed: rc, out, _ = _adb_run(["-s", target, "shell", "dumpsys", "package", package], timeout=30) m = re.search(r"versionName=(\S+)", out or "") ver = m.group(1) if m else "" return {"ok": True, "result": { "target": target, "package": package, "installed": installed, "running": running, "version": ver, "message": ("Agent laeuft" if running else "installiert, laeuft aber nicht" if installed else "nicht installiert")}} def _adb_grant_permissions(target: str, package: str) -> dict: """Erteilt dem Agenten die Sonderrechte per ADB — das spart das Antippen mit der Fernbedienung (auf einem Fire TV besonders muehsam): PROJECT_MEDIA Bildschirmaufnahme ohne Nachfrage-Dialog (Screenshots) SYSTEM_ALERT_WINDOW Einblendung ueber dem laufenden Bild (Overlay) RECORD_AUDIO Mikrofon (Sprachdialog) Best effort: was nicht geht, wird gemeldet, bricht aber nichts ab.""" granted, failed = [], [] ops = [("appops", ["appops", "set", package, "PROJECT_MEDIA", "allow"], "Bildschirmaufnahme"), ("appops", ["appops", "set", package, "SYSTEM_ALERT_WINDOW", "allow"], "Einblendung")] perms = [("android.permission.RECORD_AUDIO", "Mikrofon")] for _, cmd, label in ops: rc, out, err = _adb_run(["-s", target, "shell"] + cmd, timeout=30) (granted if rc == 0 and "error" not in f"{out}{err}".lower() else failed).append(label) for perm, label in perms: rc, out, err = _adb_run(["-s", target, "shell", "pm", "grant", package, perm], timeout=30) (granted if rc == 0 and "error" not in f"{out}{err}".lower() else failed).append(label) return {"granted": granted, "failed": failed} def _do_adb_permissions(params: dict) -> dict: """Nur die Berechtigungen setzen (ohne Neuinstallation/Einrichtung).""" target = _adb_target(params) if not target: return {"ok": False, "error": "ip/host erforderlich."} package = (params.get("package") or AGENT_PACKAGE).strip() ok, msg = _adb_connect(target) if not ok: return {"ok": False, "error": msg} r = _adb_grant_permissions(target, package) msg = "erteilt: " + (", ".join(r["granted"]) or "—") if r["failed"]: msg += " · nicht moeglich: " + ", ".join(r["failed"]) return {"ok": True, "result": {"target": target, "message": msg, **r}} def _do_adb_provision(params: dict) -> dict: """Richtet den ARIA-Agent auf dem Geraet FERN ein (RVS-Zugang uebergeben). Auf einem Fire TV gibt es keine Kamera (kein QR-Scan) und Tippen mit der Fernbedienung ist muehsam. Der Satellit startet die App deshalb per ADB direkt mit den Zugangsdaten als Intent-Extras. params: {ip, rvs_host, rvs_token, rvs_port?, rvs_tls?, name?, control?, reply_mode?, package?} """ target = _adb_target(params) if not target: return {"ok": False, "error": "ip/host erforderlich."} rvs_host = (params.get("rvs_host") or "").strip() rvs_token = (params.get("rvs_token") or "").strip() if not rvs_host or not rvs_token: return {"ok": False, "error": "rvs_host und rvs_token erforderlich."} package = (params.get("package") or AGENT_PACKAGE).strip() ok, msg = _adb_connect(target) state = _adb_state(target) wait = float(params.get("auth_wait") or ADB_AUTH_WAIT_SEC) if (not ok or state != "device") and wait > 0: authed, state = _adb_wait_authorized(target, wait) if not authed: return {"ok": False, "error": "Nicht autorisiert — 'USB-Debugging zulassen?' auf dem " f"Bildschirm bestaetigen. (Zustand: '{state or '?'}')"} elif not ok: return {"ok": False, "error": msg} if not _adb_app_installed(target, package): return {"ok": False, "error": f"'{package}' ist nicht installiert — " "erst adb.install ausfuehren."} # Sonderrechte gleich mit erteilen (Bildschirm/Overlay/Mikro) — sonst muesste # man sie am Fernseher mit der Fernbedienung zusammenklicken. perms = ({"granted": [], "failed": []} if params.get("skip_permissions") else _adb_grant_permissions(target, package)) # WICHTIG: 'adb shell' haengt die Argumente zu EINER Kommandozeile zusammen, # die auf dem Geraet nochmal von der Shell zerlegt wird. Werte mit Leerzeichen # ("Wohnzimmer Fire TV") wuerden dabei abgeschnitten -> jeden Wert quoten. import shlex q = shlex.quote # Der Intent MUSS eindeutig sein: Android vergleicht beim Start Komponente, # Action und Daten — NICHT die Extras. Ohne eigene Daten-URI gilt ein zweites # Provisioning als "gleicher Intent", die Activity wird nur nach vorn geholt # und onCreate/onNewIntent laufen nie -> der alte Name bliebe stehen. stamp = str(int(time.time() * 1000)) args = ["-s", target, "shell", "am", "start", "-n", f"{package}/.MainActivity", "-a", f"{package}.PROVISION", "-d", q(f"aria://provision/{stamp}"), "-e", "host", q(rvs_host), "-e", "token", q(rvs_token), "--ei", "port", str(int(params.get("rvs_port") or 443)), "--ez", "tls", "true" if _as_bool(params.get("rvs_tls"), True) else "false", "--ez", "control", "true" if _as_bool(params.get("control"), True) else "false", # Nur wenn die Bildschirm-Berechtigung wirklich sitzt: dann darf sich # die Aufnahme nach jedem Neustart selbst scharf schalten (dialogfrei). "--ez", "capture", "true" if "Bildschirmaufnahme" in perms.get("granted", []) else "false"] if params.get("name"): args += ["-e", "name", q(str(params["name"]))] if params.get("reply_mode"): args += ["-e", "reply_mode", q(str(params["reply_mode"]))] rc, out, err = _adb_run(args, timeout=60) txt = f"{out} {err}".strip() if rc != 0 or "Error" in txt: return {"ok": False, "error": (txt or f"exit {rc}")[:400]} # Android meldet das, wenn es den Intent als "schon offen" abtut und die # Activity nur nach vorn holt — dann kaeme die neue Konfiguration NICHT an. if "brought to the front" in txt.lower() or "not started" in txt.lower(): return {"ok": False, "error": "Android hat den Einrichtungs-Intent nicht ausgeliefert (Activity " "war schon offen). Auf dem Geraet die Agent-App einmal schliessen " "und erneut 'Nur einrichten' druecken. Details: " + txt[:200]} msg = f"Agent eingerichtet (RVS {rvs_host}) und gestartet" if perms.get("granted"): msg += " · Rechte erteilt: " + ", ".join(perms["granted"]) if perms.get("failed"): msg += " · manuell noetig: " + ", ".join(perms["failed"]) return {"ok": True, "result": { "target": target, "package": package, "permissions": perms, "message": msg}} def _as_bool(v, default: bool) -> bool: if v is None: return default if isinstance(v, bool): return v return str(v).strip().lower() in ("1", "true", "yes", "on", "ja") def _do_adb_shell(params: dict) -> dict: """Shell-Kommando auf dem Android-Geraet (Diagnose, z.B. 'pm list packages').""" target = _adb_target(params) cmd = (params.get("cmd") or params.get("command") or "").strip() if not target: return {"ok": False, "error": "ip/host erforderlich."} if not cmd: return {"ok": False, "error": "cmd erforderlich."} ok, msg = _adb_connect(target) if not ok: return {"ok": False, "error": msg} rc, out, err = _adb_run(["-s", target, "shell", cmd], timeout=60) win = _window_text(out, params) return {"ok": rc == 0, "result": {"target": target, "exit_code": rc, "stderr": err[:2000], **win}, **({} if rc == 0 else {"error": err[:500] or f"exit {rc}"})} def _apk_fetch(params: dict) -> tuple: """Besorgt die APK lokal. Gibt (pfad, fehler). Quelle: 'url' (HTTP-Pull, gecached in APK_CACHE_DIR) ODER 'apk_base64'. HTTP-Pull ist der Normalweg — der RVS ist ein Broadcast-Relay, eine APK als base64 ginge an JEDEN Client im Raum.""" import base64 import hashlib import requests os.makedirs(APK_CACHE_DIR, exist_ok=True) url = (params.get("url") or "").strip() b64 = params.get("apk_base64") want_sha = (params.get("sha256") or "").strip().lower() if b64: try: data = base64.b64decode(b64) except Exception as exc: return "", f"apk_base64 ungueltig: {exc}" elif url: digest = hashlib.sha256(url.encode()).hexdigest()[:16] cached = os.path.join(APK_CACHE_DIR, f"{digest}.apk") if os.path.isfile(cached) and os.path.getsize(cached) > 0: return cached, "" try: headers = {} if params.get("token"): headers["Authorization"] = f"token {params['token']}" with requests.get(url, stream=True, timeout=60, headers=headers, verify=not params.get("insecure")) as r: r.raise_for_status() total = 0 tmp = cached + ".part" with open(tmp, "wb") as f: for chunk in r.iter_content(65536): total += len(chunk) if total > APK_MAX_BYTES: f.close(); os.remove(tmp) return "", f"APK groesser als {APK_MAX_BYTES // 1024 // 1024} MB." f.write(chunk) os.replace(tmp, cached) except Exception as exc: return "", f"APK-Download fehlgeschlagen: {exc}" data = None else: return "", "url ODER apk_base64 erforderlich." if data is not None: if len(data) > APK_MAX_BYTES: return "", f"APK groesser als {APK_MAX_BYTES // 1024 // 1024} MB." path = os.path.join(APK_CACHE_DIR, hashlib.sha256(data).hexdigest()[:16] + ".apk") with open(path, "wb") as f: f.write(data) cached = path if want_sha: h = hashlib.sha256() with open(cached, "rb") as f: for blk in iter(lambda: f.read(65536), b""): h.update(blk) if h.hexdigest().lower() != want_sha: return "", "sha256 stimmt nicht — Download verworfen." return cached, "" def _do_adb_install(params: dict) -> dict: """Rollt eine APK auf ein Android-Geraet aus (Fire TV & Co.). params: {ip|host|device, url|apk_base64, sha256?, port?, downgrade?, timeout?} Selbstheilend: verbindet, installiert, gibt den Status zurueck.""" target = _adb_target(params) if not target: return {"ok": False, "error": "ip/host erforderlich."} apk, err = _apk_fetch(params) if err: return {"ok": False, "error": err} ok, msg = _adb_connect(target) state = _adb_state(target) # Erstkontakt: Das Geraet zeigt jetzt "USB-Debugging zulassen?" auf dem # Bildschirm. Statt abzubrechen warten wir auf die Bestaetigung — ein Klick # im Diagnostic genuegt, kein zweiter Anlauf noetig. wait = float(params.get("auth_wait") or ADB_AUTH_WAIT_SEC) if (not ok or state != "device") and wait > 0: logger.info("[adb] %s: warte bis zu %.0fs auf Autorisierung (Zustand '%s')", target, wait, state or "?") authed, state = _adb_wait_authorized(target, wait) if not authed: hint = ("Das Geraet hat die Verbindung nicht autorisiert. Auf dem " "Bildschirm erscheint 'USB-Debugging zulassen?' — bitte " "bestaetigen (Haken bei 'Immer erlauben'). " f"Ich habe {wait:.0f}s gewartet, Zustand: '{state or 'nicht erreichbar'}'.") return {"ok": False, "error": hint if (state or ok) else msg} elif not ok: return {"ok": False, "error": msg} args = ["-s", target, "install", "-r"] # -r = Update statt Neuinstallation if params.get("downgrade"): args.append("-d") args.append(apk) rc, out, err2 = _adb_run(args, timeout=float(params.get("timeout") or ADB_TIMEOUT_SEC)) txt = f"{out} {err2}".strip() size_mb = round(os.path.getsize(apk) / 1024 / 1024, 1) if rc == 0 and "success" in txt.lower(): res = {"target": target, "apk_mb": size_mb, "message": f"APK installiert ({size_mb} MB)"} # Direkt einrichten, wenn die RVS-Daten mitkamen — auf einem Fire TV # spart das der Fernbedienung das Abtippen von Host und Token. if params.get("rvs_host") and params.get("rvs_token"): prov = _do_adb_provision(params) res["provisioned"] = bool(prov.get("ok")) if prov.get("ok"): res["message"] += " und eingerichtet — der Agent sollte gleich online sein" else: res["message"] += (f". Einrichtung fehlgeschlagen: " f"{prov.get('error', '?')}") return {"ok": True, "result": res} low = txt.lower() if "signatures do not match" in low or "update_incompatible" in low: return {"ok": False, "error": "Signatur passt nicht zur installierten Version — " "die alte App muss zuerst deinstalliert werden (adb.shell " "'pm uninstall ')."} if "insufficient" in low or "no space" in low: return {"ok": False, "error": "Nicht genug Speicher auf dem Geraet."} return {"ok": False, "error": (txt or f"Installation fehlgeschlagen (exit {rc})")[:500]} def _to_int(s: str): try: return int(str(s).strip()) except (TypeError, ValueError): return None def _snmp_system(ip: str, community: str = "", version: str = "") -> Optional[dict]: """Fragt die SNMP-System-Group eines Hosts ab (ein snmpget, alle 6 OIDs). Gibt {descr,name,contact,location,uptime,objectid} oder None (kein SNMP). Nutzt gespeicherte Creds fuer die IP; kurzer Timeout, keine Retries -> Nicht-SNMP-Hosts scheitern schnell.""" params = {} if community: params["community"] = community if version: params["version"] = version base = [*_snmp_base_args(params, ip), "-t", "1", "-r", "0"] keys = list(SNMP_SYS_OIDS.keys()) ok, out = _snmp_run(["snmpget", "-Oqv", *base, ip, *SNMP_SYS_OIDS.values()], SNMP_DISCOVERY_TIMEOUT) if not ok: return None vals = out.splitlines() info = {} for k, v in zip(keys, vals): v = (v or "").strip().strip('"') if v and "No Such" not in v and "No more" not in v: info[k] = v return info or None def _snmp_kind(descr: str) -> str: """Grobe Geraeteklasse aus sysDescr (fuer type im Inventar).""" d = (descr or "").lower() if any(k in d for k in ("switch", "catalyst", "procurve", "aruba", "powerconnect")): return "switch" if any(k in d for k in ("router", "mikrotik", "routeros", "edgeos", "openwrt", "pfsense", "fritz!box")): return "router" if any(k in d for k in ("access point", "accesspoint", "unifi", "wifi", "wlan")): return "access-point" if any(k in d for k in ("printer", "laserjet", "officejet", "brother", "epson", "kyocera")): return "printer" if any(k in d for k in ("nas", "synology", "qnap", "truenas", "diskstation")): return "nas" if any(k in d for k in ("ups", "usv", "smart-ups")): return "ups" return "" # ─── Helpers ──────────────────────────────────────────────────────── def _slug(s: str) -> str: s = (s or "").strip().lower() s = re.sub(r"[^a-z0-9]+", "-", s).strip("-") return s or "dev" def _parse_http_headers(text: str) -> dict: headers = {} for line in text.split("\r\n")[1:]: if ":" in line: k, _, v = line.partition(":") headers[k.strip().lower()] = v.strip() return headers def _looks_amazon(dev: dict) -> bool: """Fire TV? Amazon-Hersteller oder Modell-Praefix AFT (AFTMM, AFTKA, ...).""" blob = f"{dev.get('manufacturer', '')} {dev.get('name', '')}".lower() model = str(dev.get("model", "")).upper() return "amazon" in blob or "fire tv" in blob or model.startswith("AFT") def _xml_tag(xml: str, tag: str) -> str: m = re.search(rf"<{tag}>(.*?)", xml, re.IGNORECASE | re.DOTALL) return m.group(1).strip() if m else "" # ─── Satellit (RVS-Client) ────────────────────────────────────────── class Satellite: def __init__(self) -> None: self.ws: Optional[websockets.WebSocketClientProtocol] = None self._devices: list[dict] = [] self._devices_ts: float = 0.0 self._scanning = False self._scan_stage = "" # Offene Port-Tunnel: tunnelId -> {"writer", "task", "last", "ziel"} self._tunnels: dict[str, dict] = {} async def _scan(self, force: bool = False) -> list[dict]: fresh = (time.time() - self._devices_ts) < DEVICE_CACHE_TTL_SEC if self._devices and fresh and not force: return self._devices if self._scanning: # Laufenden Scan abwarten — bis er fertig ist (hoechstens SCAN_MAX_SEC). # Frueher nur 6 s: direkt nach dem Start (Scan ~25 s, Liste noch # leer) kam so "keine Geraete gefunden" zurueck. deadline = time.time() + SCAN_MAX_SEC + 5 while self._scanning and time.time() < deadline: await asyncio.sleep(0.2) return self._devices self._scanning = True self._scan_stage = "start" try: await asyncio.wait_for(self._scan_inner(), SCAN_MAX_SEC) except asyncio.TimeoutError: logger.warning("[scan] nach %.0fs abgebrochen — haengt bei: %s. Liefere die letzte Liste (%d Geraete).", SCAN_MAX_SEC, self._scan_stage, len(self._devices)) finally: self._scanning = False return self._devices async def _scan_inner(self) -> None: """Ein kompletter Scan; _scan begrenzt ihn auf SCAN_MAX_SEC.""" loop = asyncio.get_event_loop() self._scan_stage = "mdns" mdns = await loop.run_in_executor(None, _discover_mdns, DISCOVER_TIMEOUT_SEC) self._scan_stage = "ssdp" ssdp = await loop.run_in_executor(None, _discover_ssdp, DISCOVER_TIMEOUT_SEC) self._scan_stage = "arp" arp = await loop.run_in_executor(None, _discover_arp) _self = [_self_device()] if LIST_SELF else [] _self = [d for d in _self if d] devices = _apply_names(_merge_devices(mdns, ssdp, arp, _self)) if SNMP_DISCOVERY: self._scan_stage = "snmp" await self._enrich_snmp(devices) self._scan_stage = "adb" await self._enrich_adb(devices) self._devices = devices self._devices_ts = time.time() n_snmp = sum(1 for d in self._devices if d.get("snmpCapable")) n_adb = sum(1 for d in self._devices if d.get("adbCapable")) logger.info("[scan] %d Geraete (mdns=%d ssdp=%d arp=%d, snmp=%d, adb=%d)", len(self._devices), len(mdns), len(ssdp), len(arp), n_snmp, n_adb) async def _enrich_snmp(self, devices: list) -> None: """Fragt jedes Geraet mit IP parallel per SNMP-System-Group ab und haengt die Infos an. Verbessert Name/Typ, wenn bisher nur eine IP bekannt war.""" loop = asyncio.get_event_loop() sem = asyncio.Semaphore(SNMP_DISCOVERY_CONCURRENCY) async def _one(dev: dict) -> None: ip = dev.get("ip") or "" if not ip: return async with sem: info = await loop.run_in_executor(None, _snmp_system, ip) if not info: return dev["snmp"] = info dev["snmpCapable"] = True # Name aufwerten, wenn er bisher nur die IP/leer war. if info.get("name") and dev.get("name", "") in ("", ip): dev["name"] = info["name"] # Typ aufwerten, wenn bisher generisch (host/leer). kind = _snmp_kind(info.get("descr", "")) if kind and dev.get("type", "") in ("", "host"): dev["type"] = kind await asyncio.gather(*(_one(d) for d in devices)) async def _enrich_adb(self, devices: list) -> None: """Prueft parallel, welche Geraete ADB-Netzwerk-Debugging offen haben (TCP 5555). Damit sieht ARIA/die Diagnostic sofort, auf welche Geraete (v.a. Fire TV Sticks) sich eine APK ausrollen laesst.""" sem = asyncio.Semaphore(16) async def _one(dev: dict) -> None: ip = dev.get("ip") or "" if not ip: return async with sem: try: fut = asyncio.open_connection(ip, ADB_PORT) reader, writer = await asyncio.wait_for(fut, timeout=1.5) writer.close() try: await writer.wait_closed() except Exception: pass except Exception: return dev["adbCapable"] = True dev["adbPort"] = ADB_PORT if dev.get("type", "") in ("", "host", "unknown", "dial"): dev["type"] = "firetv" if _looks_amazon(dev) else "android" await asyncio.gather(*(_one(d) for d in devices)) async def _send(self, message: dict) -> None: if self.ws is None: return try: await self.ws.send(json.dumps(message)) except Exception as exc: logger.warning("send fehlgeschlagen: %s", exc) async def _hello(self, log: bool = False) -> None: await self._send({ "type": "sat_hello", "payload": { "id": SATELLITE_ID, "location": SATELLITE_LOCATION, "caps": CAPABILITIES, "control": CONTROL_ENABLED, "net": NET, }, "timestamp": int(time.time() * 1000), }) if log: logger.info("sat_hello gesendet: id=%s location=%s caps=%s", SATELLITE_ID, SATELLITE_LOCATION, CAPABILITIES) # ── Port-Tunnel ────────────────────────────────────────────────── async def _tunnel_open(self, payload: dict) -> None: """Oeffnet eine TCP-Verbindung ins LAN und streamt sie zur Bridge. Gegenstueck: die Bridge haelt einen lokalen Lauscher, damit ARIA ganz normal gegen eine Adresse sprechen kann (z.B. http://aria-bridge:19001). """ tid = (payload.get("tunnelId") or "").strip() if not tid: return target = (payload.get("target") or payload.get("host") or "").strip() try: port = int(payload.get("port") or 0) except (TypeError, ValueError): port = 0 async def fail(msg: str) -> None: await self._send({ "type": "sat_tunnel_opened", "payload": {"tunnelId": tid, "satellite": SATELLITE_ID, "ok": False, "error": msg}, "timestamp": int(time.time() * 1000), }) if len(self._tunnels) >= TUNNEL_MAX: await fail(f"Zu viele offene Tunnel ({TUNNEL_MAX}).") return devices = await self._scan() ip = _resolve_target(target, devices) ok, why = _tunnel_allowed(ip, port, devices) if not ok: await fail(why) return try: reader, writer = await asyncio.wait_for( asyncio.open_connection(ip, port), timeout=10) except Exception as exc: await fail(f"Verbindung zu {ip}:{port} fehlgeschlagen: {exc}") return task = asyncio.create_task(self._tunnel_reader(tid, reader)) self._tunnels[tid] = {"writer": writer, "task": task, "last": time.time(), "ziel": f"{ip}:{port}"} logger.info("Tunnel auf: %s -> %s:%d (%s)", tid[:8], ip, port, target or ip) await self._send({ "type": "sat_tunnel_opened", "payload": {"tunnelId": tid, "satellite": SATELLITE_ID, "ok": True, "ip": ip, "port": port, "target": target}, "timestamp": int(time.time() * 1000), }) async def _tunnel_reader(self, tid: str, reader: asyncio.StreamReader) -> None: """LAN -> RVS. Endet, wenn die Gegenstelle zumacht.""" try: while True: data = await reader.read(32768) if not data: break t = self._tunnels.get(tid) if t: t["last"] = time.time() await self._send({ "type": "sat_tunnel_data", "payload": {"tunnelId": tid, "b64": base64.b64encode(data).decode("ascii")}, "timestamp": int(time.time() * 1000), }) except asyncio.CancelledError: raise except Exception as exc: logger.info("Tunnel %s Lesefehler: %s", tid[:8], exc) finally: await self._tunnel_close(tid, reason="Gegenstelle hat geschlossen") async def _tunnel_close(self, tid: str, reason: str = "") -> None: t = self._tunnels.pop(tid, None) if not t: return try: t["task"].cancel() except Exception: pass try: t["writer"].close() except Exception: pass logger.info("Tunnel zu: %s (%s) %s", tid[:8], t.get("ziel", "?"), reason) await self._send({ "type": "sat_tunnel_closed", "payload": {"tunnelId": tid, "satellite": SATELLITE_ID, "reason": reason}, "timestamp": int(time.time() * 1000), }) async def _tunnel_reaper(self) -> None: """Raeumt Tunnel weg, durch die lange nichts mehr geht. Ohne das bleibt eine vergessene Verbindung offen, bis der Satellit neu startet.""" while True: await asyncio.sleep(60) now = time.time() for tid in [k for k, v in self._tunnels.items() if now - v.get("last", now) > TUNNEL_IDLE_SEC]: await self._tunnel_close(tid, reason=f"seit {TUNNEL_IDLE_SEC}s ungenutzt") def _for_me(self, payload: dict) -> bool: target = (payload.get("satellite") or "").strip().lower() if not target or target in ("all", "*"): return True return target in (SATELLITE_ID.lower(), SATELLITE_LOCATION.lower()) async def _discover_reply(self, payload: dict) -> None: devices = await self._scan(force=bool(payload.get("force"))) await self._send({ "type": "sat_devices", "payload": {"requestId": payload.get("requestId", ""), "satellite": SATELLITE_ID, "location": SATELLITE_LOCATION, "devices": devices, "net": NET}, "timestamp": int(time.time() * 1000), }) async def _command_reply(self, payload: dict) -> None: req_id = payload.get("requestId", "") action = (payload.get("action") or "").strip() params = payload.get("params") or {} if payload.get("device") and "device" not in params: params["device"] = payload.get("device") # Geraeteliste nur scannen, wenn die Aktion sie wirklich braucht # (dial.launch loest ein Geraet auf, oder es wurde ein device-Ref # mitgegeben). http.get/http.post/wol arbeiten direkt mit url/mac — # ein voller LAN-Scan davor kostete nur unnoetig viele Sekunden. needs_devices = (action in ("dial.launch", "fritzbox.backup_all") or bool(params.get("device")) or (action.startswith("fritzbox.") and not re.match(r"^\d{1,3}(\.\d{1,3}){3}$", str(params.get("ip") or "")))) devices = await self._scan() if needs_devices else self._devices result = await _control(action, params, devices) await self._send({ "type": "sat_result", "payload": {"requestId": req_id, "satellite": SATELLITE_ID, **result}, "timestamp": int(time.time() * 1000), }) async def _handle(self, raw: str) -> None: try: msg = json.loads(raw) except Exception: return mtype = msg.get("type", "") payload = msg.get("payload", {}) or {} if mtype == "sat_discover": if not self._for_me(payload): return # Eigener Task: ein langer Scan darf den Nachrichten-Loop (Tunnel, # Befehle, weitere Scans) nicht anhalten. asyncio.create_task(self._discover_reply(payload)) elif mtype == "sat_tunnel_open": if not self._for_me(payload): return await self._tunnel_open(payload) elif mtype == "sat_tunnel_data": # Kein _for_me: die tunnelId ist eindeutig, und ein fremder # Satellit kennt sie nicht. tid = (payload.get("tunnelId") or "").strip() t = self._tunnels.get(tid) if t: try: t["writer"].write(base64.b64decode(payload.get("b64") or "")) t["last"] = time.time() except Exception as exc: logger.warning("Tunnel %s schreiben fehlgeschlagen: %s", tid[:8], exc) await self._tunnel_close(tid, reason=str(exc)[:80]) elif mtype == "sat_tunnel_close": await self._tunnel_close((payload.get("tunnelId") or "").strip(), reason="von ARIA geschlossen") elif mtype == "sat_name_set": if not self._for_me(payload): return key = (payload.get("mac") or payload.get("ip") or "").strip().lower() name = (payload.get("name") or "").strip()[:60] req_id = payload.get("requestId", "") if key: if name: _names[key] = name else: _names.pop(key, None) _names_save() _apply_names(self._devices) logger.info("Geraetename: %s -> %r", key, name or "(entfernt)") await self._send({ "type": "sat_name_result", "payload": {"requestId": req_id, "satellite": SATELLITE_ID, "ok": bool(key), "key": key, "name": name, "error": "" if key else "mac oder ip noetig"}, "timestamp": int(time.time() * 1000), }) elif mtype == "sat_command": if not self._for_me(payload): return # Eigener Task: eine Aktion kann lange dauern (FritzBox wartet bis # 2 min auf die Taste, adb.install Minuten) — der Loop bleibt frei. asyncio.create_task(self._command_reply(payload)) elif mtype == "sat_creds_set": if not self._for_me(payload): return ip = (payload.get("ip") or "").strip() creds = payload.get("creds") or {} ok = False weak = _strong_password_problem((creds.get("fritzbox") or {}).get("backupPass")) \ if isinstance(creds, dict) and (creds.get("fritzbox") or {}).get("backupPass") else "" if weak: await self._send({"type": "sat_creds_result", "payload": {"requestId": payload.get("requestId", ""), "satellite": SATELLITE_ID, "ok": False, "ip": ip, "error": f"Sicherungs-Kennwort zu schwach — es fehlt: {weak}."}, "timestamp": int(time.time() * 1000)}) return if ip and isinstance(creds, dict): entry = _CREDS.setdefault(ip, {}) for t in ("snmp", "http", "fritzbox", "ssh"): if t in creds: if creds[t]: # leeres Objekt = Typ loeschen entry[t] = _creds_keep_secrets(creds[t], entry.get(t) or {}) else: entry.pop(t, None) if not entry: _CREDS.pop(ip, None) try: _creds_save() ok = True except Exception as exc: logger.warning("[creds] speichern fehlgeschlagen: %s", exc) await self._send({"type": "sat_creds_result", "payload": {"requestId": payload.get("requestId", ""), "satellite": SATELLITE_ID, "ok": ok, "ip": ip}, "timestamp": int(time.time() * 1000)}) elif mtype == "sat_creds_delete": if not self._for_me(payload): return ip = (payload.get("ip") or "").strip() ctype = (payload.get("type") or "").strip() if ip in _CREDS: if ctype: _CREDS[ip].pop(ctype, None) if not _CREDS[ip]: _CREDS.pop(ip, None) else: _CREDS.pop(ip, None) try: _creds_save() except Exception as exc: logger.warning("[creds] speichern fehlgeschlagen: %s", exc) await self._send({"type": "sat_creds_result", "payload": {"requestId": payload.get("requestId", ""), "satellite": SATELLITE_ID, "ok": True, "ip": ip}, "timestamp": int(time.time() * 1000)}) elif mtype == "sat_creds_list": if not self._for_me(payload): return ip = (payload.get("ip") or "").strip() reveal = bool(payload.get("reveal")) and bool(ip) await self._send({"type": "sat_creds_list_result", "payload": {"requestId": payload.get("requestId", ""), "satellite": SATELLITE_ID, "location": SATELLITE_LOCATION, "ip": ip, "revealed": reveal, "items": _creds_public_summary(ip, reveal)}, "timestamp": int(time.time() * 1000)}) async def _periodic_scan(self) -> None: while True: try: await self._scan(force=True) except Exception as exc: logger.warning("periodischer Scan-Fehler: %s", exc) await asyncio.sleep(SCAN_INTERVAL_SEC) async def _heartbeat(self) -> None: # Re-announce bei jedem Heartbeat: falls die Bridge NACH uns (neu) # verbindet, lernt sie uns so innerhalb von HEARTBEAT_SEC — RVS replayt # nichts. Haelt zugleich last_seen in der Bridge-Registry frisch. while True: await asyncio.sleep(HEARTBEAT_SEC) await self._send({"type": "heartbeat", "timestamp": int(time.time() * 1000)}) await self._hello() async def run(self) -> None: if not RVS_HOST or not RVS_TOKEN: logger.error("RVS_HOST und RVS_TOKEN sind Pflicht (siehe .env.example).") return _creds_load() _names_load() asyncio.create_task(self._periodic_scan()) if TUNNEL_ENABLED: asyncio.create_task(self._tunnel_reaper()) backoff = 1 # use_tls kann bei Fehlschlag einmal auf ws:// fallen (RVS_TLS_FALLBACK), # danach wieder zurueck auf RVS_TLS (kein Sticky-Fallback). Bei TLS+RVS_SNI # nutzt die URI den HOSTNAMEN (Host-Header/SNI/Cert), getaddrinfo mappt ihn # auf die IP in RVS_HOST. use_tls = RVS_TLS tls_fallback_tried = False connect_kwargs = {"max_size": 8 * 1024 * 1024, "ping_interval": 20, "ping_timeout": 20} while True: proto = "wss" if use_tls else "ws" uri_host = RVS_SNI if (use_tls and RVS_SNI) else RVS_HOST url = f"{proto}://{uri_host}:{RVS_PORT}?token={RVS_TOKEN}" fallback = False try: logger.info("Verbinde mit RVS %s://%s:%s%s …", proto, uri_host, RVS_PORT, f" (TCP {RVS_HOST})" if (use_tls and RVS_SNI) else "") async with websockets.connect(url, **connect_kwargs) as ws: self.ws = ws backoff = 1 tls_fallback_tried = False await self._hello(log=True) hb = asyncio.create_task(self._heartbeat()) try: async for raw in ws: await self._handle(raw) finally: hb.cancel() except Exception as exc: logger.warning("RVS-Verbindung verloren: %s", exc) if use_tls and RVS_TLS_FALLBACK and not tls_fallback_tried: logger.info("TLS fehlgeschlagen — Fallback auf ws://") use_tls = False tls_fallback_tried = True fallback = True finally: self.ws = None if fallback: continue await asyncio.sleep(backoff) backoff = min(backoff * 2, 30) use_tls = RVS_TLS tls_fallback_tried = False def main() -> None: logger.info("ARIA Satellit startet — id=%s location=%s control=%s", SATELLITE_ID, SATELLITE_LOCATION, CONTROL_ENABLED) logger.info("Netz: primary_ip=%s mac=%s alle=%s", NET.get("primary_ip"), NET.get("mac") or "?", NET.get("ips")) if NET.get("warning"): logger.warning("⚠ %s", NET["warning"]) try: asyncio.run(Satellite().run()) except KeyboardInterrupt: pass if __name__ == "__main__": main()