""" 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 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.info,fritzbox.hosts,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 {} def _creds_public_summary() -> list: """Fuer sat_creds_list: welche Geraete welche Cred-Typen haben — OHNE Secrets.""" out = [] for ip, entry in sorted(_CREDS.items()): types = [t for t in ("snmp", "http", "fritzbox", "ssh") if entry.get(t)] out.append({"ip": ip, "types": 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"] 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 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 ──────────────────────────────────────────────────────── 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 action not in CONTROL_ALLOWLIST: 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 in ("fritzbox.info", "fritzbox.hosts"): return await loop.run_in_executor(None, _do_fritzbox, action, params) 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) -> 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 = ( '' '' 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: 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} 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).""" ip = _snmp_target(params) if not ip: return {"ok": False, "error": "ip/host erforderlich."} 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.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 = [] 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) if m else "" hosts.append({"name": _t("HostName"), "ip": _t("IPAddress"), "mac": _t("MACAddress"), "active": _t("Active") in ("1", "true")}) return {"ok": True, "result": {"ip": ip, "count": len(hosts), "hosts": hosts}} # 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 (grob) for _ in range(30): await asyncio.sleep(0.2) if not self._scanning: break 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 _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 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 == "dial.launch" or bool(params.get("device")) 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), }) 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 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[t] 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 await self._send({"type": "sat_creds_list_result", "payload": {"requestId": payload.get("requestId", ""), "satellite": SATELLITE_ID, "location": SATELLITE_LOCATION, "items": _creds_public_summary()}, "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()