From 116a947a4a6df82342c0370eaa476c760219b55b Mon Sep 17 00:00:00 2001 From: duffyduck Date: Sat, 31 Oct 2026 17:11:58 +0100 Subject: [PATCH] =?UTF-8?q?feat(satellite):=20FritzBox=20komplett=20steuer?= =?UTF-8?q?n=20=E2=80=94=20describe/call/wlan=20ueber=20TR-064?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Statt einzelner fest programmierter Befehle nutzt der Satellit die Selbstbeschreibung der Box (tr64desc.xml + SCPD je Dienst, ohne Anmeldung lesbar): was DIESE Firmware kann, steht dort — keine Tabellen pro Modell. - fritzbox.describe: alle TR-064-Aktionen der Box, je mit [liest]/ [aendert]/[GEFAEHRLICH]; filter grenzt ein (WLAN, Dect, Hosts …). - fritzbox.call: beliebige Aktion (service kurz "WLANConfiguration:3", args ohne "New"-Praefix moeglich). Alles ausser Get… braucht confirmed=true, sonst needs_confirmation — ARIA muss vorher fragen. - fritzbox.wlan: alle WLANs inkl. Gast (index/ssid/an), schalten mit index + enable (+ confirmed). - FritzBox per Geraetename ("Fritzbox Keller") statt IP ansprechbar. - Allowlist kennt Platzhalter "fritzbox.*" (Default + .env.example). - Brain: Tool-Beschreibung mit Ablauf (describe → call) und Freigabe-Regel, fritzbox.* mit 60 s Timeout. Getestet gegen eine nachgebaute TR-064-Box: describe/filter, lesen, schreiben ohne/mit Freigabe (ohne kommt nichts an der Box an), WLAN schalten, Namensaufloesung, Allowlist. Co-Authored-By: Claude Opus 5.5 --- aria-brain/agent.py | 17 +++- satellite/.env.example | 6 +- satellite/satellite.py | 189 +++++++++++++++++++++++++++++++++++++++-- 3 files changed, 204 insertions(+), 8 deletions(-) diff --git a/aria-brain/agent.py b/aria-brain/agent.py index 79f3b0f..e3ce43c 100644 --- a/aria-brain/agent.py +++ b/aria-brain/agent.py @@ -1295,7 +1295,19 @@ META_TOOLS = [ "— Internetverbindung/Datenrate/externe IP bzw. verbundene Geraete MIT " "Verbindungsart (LAN/WLAN), lan_port (an welcher LAN-Buchse der Box) und " "speed_mbit; 'find' filtert nach Name/IP/MAC (braucht " - "hinterlegten FritzBox-Login). 'ssh.exec' (params={'ip':'...','cmd':'...'}) " + "hinterlegten FritzBox-Login). FRITZ!BOX VOLLSTAENDIG STEUERN: " + "'fritzbox.describe' (params={'ip':'','filter':'WLAN'}) listet, " + "was DIESE Box per TR-064 kann (je Aktion [liest]/[aendert]/[GEFAEHRLICH]); " + "'fritzbox.call' (params={'ip':..,'service':'WLANConfiguration:3'," + "'action':'GetInfo','args':{...}}) ruft jede davon auf; 'fritzbox.wlan' " + "zeigt alle WLANs inkl. Gast (index, ssid, an) und schaltet sie " + "(index + enable). Unbekannte Funktion (WPS, DECT, Anrufliste, Internet " + "sperren …)? Erst describe mit passendem filter, dann call. REGEL: Alles " + "was AENDERT (Set…, WPS, Neustart …) braucht confirmed=true — und das " + "setzt Du NUR, nachdem Du Stefan gesagt hast, was genau passiert, und er " + "ausdruecklich ja gesagt hat. Ohne confirmed kommt needs_confirmation " + "zurueck: dann fragen, nicht raten. [GEFAEHRLICH] (Neustart, " + "Werkseinstellung, Portfreigabe, Passwort) besonders deutlich ankuendigen. 'ssh.exec' (params={'ip':'...','cmd':'...'}) " "— fuehrt EIN Shell-Kommando per SSH auf einem Geraet aus (Server, Pi, NAS) " "und gibt exit_code + stdout (gefenstert: contains/offset/max_chars) + " "stderr zurueck; braucht hinterlegte SSH-Credentials (Benutzer + Passwort " @@ -3233,7 +3245,8 @@ class Agent: # adb.*: Warten auf die Geraete-Autorisierung (bis 90s) + Download + # Installation koennen zusammen mehrere Minuten dauern. sat_timeout = (320.0 if action.startswith("adb.") - else 200.0 if action == "ssh.exec" else 20.0) + else 200.0 if action == "ssh.exec" + else 60.0 if action.startswith("fritzbox.") else 20.0) result = _post("/internal/satellite", { "op": "command", "satellite": sat, "device": arguments.get("device") or "", "action": action, "params": params, diff --git a/satellite/.env.example b/satellite/.env.example index 958112f..0f818ac 100644 --- a/satellite/.env.example +++ b/satellite/.env.example @@ -40,6 +40,10 @@ CONTROL_ENABLED=true # snmp.info Modell/Seriennummer/Firmware-Version (params: ip) # fritzbox.info FritzBox: Verbindung/Datenrate/externe IP (TR-064, braucht Login) # fritzbox.hosts FritzBox: verbundene Geraete (TR-064, braucht Login) +# fritzbox.describe FritzBox: was DIESE Box kann (alle TR-064-Aktionen, ohne Login) +# fritzbox.call FritzBox: beliebige TR-064-Aktion; aendernde nur mit confirmed=true +# fritzbox.wlan FritzBox: alle WLANs (inkl. Gast) anzeigen / schalten +# (fritzbox.* in der Liste erlaubt alle fritzbox-Aktionen) # ssh.exec Kommando per SSH ausfuehren (params: ip, cmd; Auth aus Creds 'ssh') # adb.install APK auf ein Android-Geraet/Fire TV ausrollen (params: ip, url) # adb.connect ADB-Verbindung aufbauen (params: ip) @@ -52,7 +56,7 @@ CONTROL_ENABLED=true # Mikro) — spart das Zusammenklicken mit der Fernbedienung # ACHTUNG: Diese Liste ueberschreibt den Code-Default. Wird eine neue Aktion # ergaenzt, MUSS sie hier nachgetragen werden — sonst lehnt der Satellit sie ab. -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,adb.permissions +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,adb.permissions # ─── ADB / Android-Rollout (optional) ────────────────────────────── # Fire TV Sticks & Android-Geraete im LAN: APK ausrollen ohne Datei-Kopieren. diff --git a/satellite/satellite.py b/satellite/satellite.py index 6459c2e..93115b3 100644 --- a/satellite/satellite.py +++ b/satellite/satellite.py @@ -135,7 +135,7 @@ 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," + "snmp.ports,snmp.info,fritzbox.*,ssh.exec," "adb.devices,adb.connect,adb.install,adb.shell,adb.provision,adb.status").split(",") if a.strip() ] @@ -815,11 +815,19 @@ def _tunnel_allowed(ip: str, port: int, devices: list[dict]) -> tuple[bool, str] # ─── 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 action not in CONTROL_ALLOWLIST: + 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() @@ -838,8 +846,11 @@ async def _control(action: str, params: dict, devices: list[dict]) -> dict: 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.startswith("fritzbox."): + # Geraetename ("Fritzbox Keller") → IP, wie bei den anderen Aktionen + t = _snmp_target(params) + ip = _resolve_target(t, devices) or t + return await loop.run_in_executor(None, _do_fritzbox, action, {**params, "ip": ip}) if action == "ssh.exec": return await loop.run_in_executor(None, _do_ssh, params) if action == "adb.devices": @@ -1197,6 +1208,165 @@ def _tr064(ip: str, user: str, pw: str, service: str, control: str, 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 ""))}} + + +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()} + 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"): + return r + return {"ok": True, "result": {"ip": ip, "service": svc["short"], "action": action, **r["fields"]}} + + +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)."}} + + 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 @@ -1205,6 +1375,8 @@ def _do_fritzbox(action: str, params: dict) -> dict: 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) 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 "") @@ -1213,6 +1385,11 @@ def _do_fritzbox(action: str, params: dict) -> dict: "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.hosts": p = _tr064(ip, user, pw, "urn:dslforum-org:service:Hosts:1", "/upnp/control/hosts", "X_AVM-DE_GetHostListPath") @@ -2189,7 +2366,9 @@ class Satellite: # (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")) + needs_devices = (action == "dial.launch" 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({