feat(satellite): FritzBox komplett steuern — describe/call/wlan ueber TR-064
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 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -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.
|
||||
|
||||
+184
-5
@@ -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"<service>(.*?)</service>", xml, re.DOTALL):
|
||||
blk = m.group(1)
|
||||
|
||||
def _t(tag):
|
||||
mm = re.search(rf"<{tag}>(.*?)</{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"<action>(.*?)</action>", xml, re.DOTALL):
|
||||
blk = m.group(1)
|
||||
name = re.search(r"<name>(.*?)</name>", blk)
|
||||
if not name:
|
||||
continue
|
||||
args_in, args_out = [], []
|
||||
for a in re.finditer(r"<argument>(.*?)</argument>", blk, re.DOTALL):
|
||||
an = re.search(r"<name>(.*?)</name>", a.group(1))
|
||||
ad = re.search(r"<direction>(.*?)</direction>", 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({
|
||||
|
||||
Reference in New Issue
Block a user