feat: FritzBox-Sicherungen verwalten — Liste, Ansicht, Wiederherstellen, Loeschen
Satellit: - sat_command laeuft als eigener Task (eine Aktion, die 2 min auf die Taste wartet, blockiert den Satelliten nicht mehr). - Sicherung bekommt eine Begleitdatei (Box, Modell, Firmware, Seriennr., Datum, QUELLE des Kennworts — nie der Wert). - fritzbox.backups: Sicherungen je Box, neueste zuerst. - fritzbox.backup_view: lesbar — Kopfdaten, wichtigste Einstellungen (Name, LAN/DHCP, WLAN, Kanal, DNS, SIP …), mit full alle Bereiche; verschluesselte Werte als «verschluesselt»; nur Dateien aus dem Sicherungsordner (kein Pfad von aussen). - fritzbox.restore: X_AVM-DE_SetConfigFile — der Satellit stellt die Datei kurz per HTTP bereit (Zufallspfad, endet nach Abruf/10 min), die Box holt sie ab und startet neu. Nur mit confirmed, Taste per button_wait, mit dem Kennwort aus der Quelle der Sicherung; fremde Box nur mit other_box. - fritzbox.backup_delete (mit confirmed). no_file fuer das Diagnostic. Diagnostic: 💾 bei FritzBox-Geraeten → Sicherungen: Jetzt sichern, ℹ (lesbar + Suche), Wiederherstellen, Loeschen; Sichern/Wiederherstellen mit 2-Minuten-Countdown fuer die Taste, Ergebnis im Fenster (✓ / Taste nicht gedrueckt / Grund), Erfolg blendet sich aus. Brain: Beschreibung fuer Liste/Ansicht/Wiederherstellen/Loeschen. Getestet: Satellit gegen nachgebaute Box (Sichern, Liste, Ansicht ohne Geheimnisse, Pfadschutz, Wiederherstellen inkl. Abholen der Datei, Loeschen), Diagnostic-Fenster in headless Chromium mit simuliertem Satelliten. Gegen eine echte Box noch nicht. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
+235
-24
@@ -35,6 +35,7 @@ import os
|
||||
import re
|
||||
import socket
|
||||
import struct
|
||||
import threading
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
@@ -1484,17 +1485,29 @@ def _fb_backup(ip: str, user: str, pw: str, params: dict) -> dict:
|
||||
os.chmod(path, 0o600)
|
||||
except OSError:
|
||||
pass
|
||||
info = _tr064(ip, user, pw, "urn:dslforum-org:service:DeviceInfo:1", "/upnp/control/deviceinfo", "GetInfo")
|
||||
fi = info.get("fields", {}) if info.get("ok") else {}
|
||||
meta = {"ip": ip, "name": str(params.get("name") or ip), "created": time.strftime("%Y-%m-%dT%H:%M:%S"),
|
||||
"bytes": len(d.content), "model": fi.get("NewModelName", ""), "firmware": fi.get("NewSoftwareVersion", ""),
|
||||
"serial": fi.get("NewSerialNumber", ""),
|
||||
# nur die QUELLE des Kennworts — zum Wiederherstellen noetig, nie der Wert
|
||||
"kennwort": "angegeben" if params.get("password") else ("backupPass" if stored else "login")}
|
||||
with open(path[:-len(".export")] + ".json", "w", encoding="utf-8") as fh:
|
||||
json.dump(meta, fh, ensure_ascii=False, indent=1)
|
||||
old = sorted(f for f in os.listdir(FB_BACKUP_DIR) if f.startswith(label + "_") and f.endswith(".export"))
|
||||
for f in old[:-FB_BACKUP_KEEP]:
|
||||
try:
|
||||
os.remove(os.path.join(FB_BACKUP_DIR, f))
|
||||
except OSError:
|
||||
pass
|
||||
for ext in (".export", ".json"):
|
||||
try:
|
||||
os.remove(os.path.join(FB_BACKUP_DIR, f[:-len(".export")] + ext))
|
||||
except OSError:
|
||||
pass
|
||||
logger.info("[fritzbox] %s: Sicherung %s (%d Bytes)", ip, fname, len(d.content))
|
||||
import base64
|
||||
# Die Datei geht mit zurueck — ARIAs Brain legt sie unter /shared/uploads/
|
||||
# ab, damit sie in den Chat kann. Sie ist mit dem Kennwort verschluesselt.
|
||||
file_b64 = base64.b64encode(d.content).decode() if len(d.content) <= 20 * 1024 * 1024 else ""
|
||||
# no_file: das Diagnostic braucht nur die Meldung, die Datei bleibt auf dem Satelliten
|
||||
file_b64 = (base64.b64encode(d.content).decode()
|
||||
if len(d.content) <= 20 * 1024 * 1024 and not params.get("no_file") else "")
|
||||
return {"ok": True, "result": {
|
||||
"ip": ip, "datei": path, "bytes": len(d.content), "auf": "Satellit",
|
||||
**({"file_b64": file_b64, "filename": fname} if file_b64 else {}),
|
||||
@@ -1533,6 +1546,191 @@ def _fb_wlan(ip: str, user: str, pw: str, params: dict) -> dict:
|
||||
"hinweis": "Das letzte WLAN ist meist das Gast-WLAN. Schalten: index + enable (+ confirmed=true nach Stefans Ja)."}}
|
||||
|
||||
|
||||
# ─── Sicherungen verwalten: Liste, Ansicht, Wiederherstellen, Loeschen ───
|
||||
|
||||
def _fb_backup_path(name: str) -> str:
|
||||
"""Nur Dateien aus FB_BACKUP_DIR — kein Pfad von aussen."""
|
||||
base = os.path.basename(str(name or ""))
|
||||
if not base.endswith(".export"):
|
||||
base += ".export"
|
||||
path = os.path.join(FB_BACKUP_DIR, base)
|
||||
return path if os.path.isfile(path) else ""
|
||||
|
||||
|
||||
def _fb_backup_meta(path: str) -> dict:
|
||||
try:
|
||||
with open(path[:-len(".export")] + ".json", encoding="utf-8") as fh:
|
||||
return json.load(fh)
|
||||
except Exception:
|
||||
return {}
|
||||
|
||||
|
||||
def _fb_backups(ip: str, params: dict) -> dict:
|
||||
label = _slug(str(params.get("name") or ""))
|
||||
out = []
|
||||
for f in sorted(os.listdir(FB_BACKUP_DIR) if os.path.isdir(FB_BACKUP_DIR) else [], reverse=True):
|
||||
if not f.endswith(".export"):
|
||||
continue
|
||||
m = _fb_backup_meta(os.path.join(FB_BACKUP_DIR, f))
|
||||
if m.get("ip") != ip and not (label and f.startswith(label + "_")):
|
||||
continue
|
||||
out.append({"datei": f, "erstellt": m.get("created", ""), "bytes": m.get("bytes") or
|
||||
os.path.getsize(os.path.join(FB_BACKUP_DIR, f)), "modell": m.get("model", ""),
|
||||
"firmware": m.get("firmware", ""), "kennwort": m.get("kennwort", "")})
|
||||
return {"ok": True, "result": {"ip": ip, "count": len(out), "sicherungen": out}}
|
||||
|
||||
|
||||
# Lesbare Hinweise aus der Sicherung: AVM-Konfig "schluessel = wert;"
|
||||
_FB_VIEW_KEYS = {
|
||||
"hostname": "Name der Box", "ssid": "WLAN-Name (2,4 GHz)", "ssid_scnd": "WLAN-Name (5 GHz)",
|
||||
"guest_ssid": "Gast-WLAN-Name", "channel": "WLAN-Kanal", "ipaddr": "IP-Adresse",
|
||||
"netmask": "Netzmaske", "dhcpstart": "DHCP von", "dhcpend": "DHCP bis",
|
||||
"username": "Benutzername", "registrar": "SIP-Registrar", "msn": "Rufnummer",
|
||||
"dnsserver1": "DNS-Server 1", "dnsserver2": "DNS-Server 2", "ntp_server": "Zeitserver",
|
||||
}
|
||||
|
||||
|
||||
def _fb_backup_view(params: dict) -> dict:
|
||||
path = _fb_backup_path(params.get("file") or params.get("datei") or "")
|
||||
if not path:
|
||||
return {"ok": False, "error": "Sicherung nicht gefunden (Dateiname aus fritzbox.backups)."}
|
||||
text = open(path, encoding="utf-8", errors="replace").read()
|
||||
header, sections, binfiles = {}, {}, []
|
||||
cur, buf = None, []
|
||||
for line in text.splitlines():
|
||||
m = re.match(r"\*{4} (CFGFILE|BINFILE|B64FILE):(\S+)", line)
|
||||
if m:
|
||||
if cur:
|
||||
sections[cur] = "\n".join(buf)
|
||||
cur, buf = (m.group(2) if m.group(1) == "CFGFILE" else None), []
|
||||
if m.group(1) != "CFGFILE":
|
||||
binfiles.append(m.group(2))
|
||||
continue
|
||||
if line.startswith("**** END OF FILE"):
|
||||
if cur:
|
||||
sections[cur] = "\n".join(buf)
|
||||
cur, buf = None, []
|
||||
continue
|
||||
if cur is not None:
|
||||
buf.append(line)
|
||||
elif not sections and not binfiles and "=" in line and not line.startswith("****"):
|
||||
k, v = line.split("=", 1)
|
||||
header[k.strip()] = "«verschluesselt»" if v.strip().startswith("$$$$") else v.strip()
|
||||
secret = re.compile(r"\$\$\$\$\S+")
|
||||
highlights = []
|
||||
for name, body in sections.items():
|
||||
for m in re.finditer(r"^\s*(\w+)\s*=\s*\"?([^;\"]*)\"?\s*;", body, re.MULTILINE):
|
||||
key, val = m.group(1), m.group(2).strip()
|
||||
if key in _FB_VIEW_KEYS and val and len(highlights) < 80:
|
||||
highlights.append({"was": _FB_VIEW_KEYS[key], "wert": "«verschluesselt»" if val.startswith("$$$$") else val,
|
||||
"datei": name, "schluessel": key})
|
||||
res = {"datei": os.path.basename(path), "meta": _fb_backup_meta(path), "kopf": header,
|
||||
"abschnitte": sorted(sections), "anhaenge": binfiles, "wichtig": highlights}
|
||||
if params.get("full"):
|
||||
budget, full = 900_000, {}
|
||||
for name, body in sections.items():
|
||||
body = secret.sub("«verschluesselt»", body)
|
||||
full[name] = body[:max(0, budget)]
|
||||
budget -= len(body)
|
||||
res["inhalt"] = full
|
||||
return {"ok": True, "result": res}
|
||||
|
||||
|
||||
def _fb_serve_once(path: str, seconds: float = 600):
|
||||
"""Stellt EINE Datei kurz per HTTP bereit (die Box holt sich die Sicherung
|
||||
beim Wiederherstellen selbst ab). Zufaelliger Pfad, endet nach dem ersten
|
||||
Abruf oder nach `seconds`. Gibt (url, served_event)."""
|
||||
import http.server
|
||||
import secrets as _secrets
|
||||
token = _secrets.token_urlsafe(16)
|
||||
fname = os.path.basename(path)
|
||||
served = threading.Event()
|
||||
|
||||
class H(http.server.BaseHTTPRequestHandler):
|
||||
def log_message(self, *a):
|
||||
pass
|
||||
|
||||
def do_GET(self):
|
||||
if self.path != f"/{token}/{fname}":
|
||||
self.send_response(404)
|
||||
self.end_headers()
|
||||
return
|
||||
data = open(path, "rb").read()
|
||||
self.send_response(200)
|
||||
self.send_header("Content-Length", str(len(data)))
|
||||
self.end_headers()
|
||||
self.wfile.write(data)
|
||||
served.set()
|
||||
|
||||
srv = http.server.ThreadingHTTPServer(("0.0.0.0", 0), H)
|
||||
threading.Thread(target=srv.serve_forever, daemon=True).start()
|
||||
|
||||
def _stop():
|
||||
served.wait(seconds)
|
||||
time.sleep(5)
|
||||
srv.shutdown()
|
||||
threading.Thread(target=_stop, daemon=True).start()
|
||||
host = (NET or {}).get("primary_ip") or "127.0.0.1"
|
||||
return f"http://{host}:{srv.server_address[1]}/{token}/{fname}", served
|
||||
|
||||
|
||||
def _fb_restore(ip: str, user: str, pw: str, params: dict) -> dict:
|
||||
path = _fb_backup_path(params.get("file") or params.get("datei") or "")
|
||||
if not path:
|
||||
return {"ok": False, "error": "Sicherung nicht gefunden (Dateiname aus fritzbox.backups)."}
|
||||
meta = _fb_backup_meta(path)
|
||||
if meta.get("ip") and meta["ip"] != ip and not params.get("other_box"):
|
||||
return {"ok": False, "error": (f"Diese Sicherung stammt von {meta.get('name') or meta['ip']} ({meta.get('model', '?')}), "
|
||||
f"nicht von {ip}. Auf eine andere Box nur mit other_box=true einspielen.")}
|
||||
if params.get("confirmed") not in (True, "true"):
|
||||
return {"ok": False, "needs_confirmation": True, "error": (
|
||||
f"Wiederherstellen von {os.path.basename(path)} ist GEFAEHRLICH: die Box uebernimmt ALLE "
|
||||
"Einstellungen dieser Sicherung und startet neu (1–3 Minuten kein Netz/Internet). Erst "
|
||||
"Stefan fragen; nur nach seinem ausdruecklichen Ja erneut mit confirmed=true.")}
|
||||
src = meta.get("kennwort", "")
|
||||
stored = _creds_for(ip).get("fritzbox", {}).get("backupPass")
|
||||
if params.get("password"):
|
||||
export_pw = str(params["password"])
|
||||
elif src == "angegeben":
|
||||
return {"ok": False, "error": "Diese Sicherung wurde mit einem eigens genannten Kennwort erstellt — "
|
||||
"Stefan muss es fuer das Wiederherstellen angeben (password)."}
|
||||
elif src == "backupPass":
|
||||
if not stored:
|
||||
return {"ok": False, "error": "Fuer diese Sicherung ist kein Sicherungs-Kennwort mehr hinterlegt."}
|
||||
export_pw = stored
|
||||
else:
|
||||
export_pw = pw
|
||||
url, served = _fb_serve_once(path)
|
||||
r = _fb_call(ip, user, pw, {"service": "DeviceConfig", "action": "X_AVM-DE_SetConfigFile",
|
||||
"args": {"X_AVM-DE_Password": export_pw, "X_AVM-DE_ConfigFileUrl": url},
|
||||
"confirmed": True, "button_wait": params.get("button_wait")})
|
||||
if not r.get("ok"):
|
||||
return r
|
||||
got = served.wait(30)
|
||||
logger.info("[fritzbox] %s: Wiederherstellen %s (abgeholt: %s)", ip, os.path.basename(path), got)
|
||||
return {"ok": True, "result": {
|
||||
"ip": ip, "datei": os.path.basename(path), "abgeholt": got,
|
||||
"hinweis": ("Die Box hat die Sicherung abgeholt, uebernimmt die Einstellungen und startet neu — "
|
||||
"1–3 Minuten nicht erreichbar." if got else
|
||||
"Die Box hat den Auftrag angenommen, die Datei aber (noch) nicht abgeholt. Erreicht sie "
|
||||
"den Satelliten? In ein paar Minuten pruefen, ob die Einstellungen uebernommen sind.")}}
|
||||
|
||||
|
||||
def _fb_backup_delete(params: dict) -> dict:
|
||||
path = _fb_backup_path(params.get("file") or params.get("datei") or "")
|
||||
if not path:
|
||||
return {"ok": False, "error": "Sicherung nicht gefunden."}
|
||||
if params.get("confirmed") not in (True, "true"):
|
||||
return {"ok": False, "needs_confirmation": True,
|
||||
"error": f"{os.path.basename(path)} endgueltig loeschen? Erst Stefan fragen, dann confirmed=true."}
|
||||
for ext in (".export", ".json"):
|
||||
try:
|
||||
os.remove(path[:-len(".export")] + ext)
|
||||
except OSError:
|
||||
pass
|
||||
return {"ok": True, "result": {"geloescht": os.path.basename(path)}}
|
||||
|
||||
|
||||
def _do_fritzbox(action: str, params: dict) -> dict:
|
||||
"""fritzbox.info -> Modell/Firmware/Verbindung/externe IP/Datenrate.
|
||||
fritzbox.hosts -> Liste der bekannten Geraete (Name/IP/MAC/aktiv) plus
|
||||
@@ -1543,6 +1741,12 @@ def _do_fritzbox(action: str, params: dict) -> dict:
|
||||
return {"ok": False, "error": "ip/host erforderlich."}
|
||||
if action == "fritzbox.describe": # braucht keine Anmeldung
|
||||
return _fb_describe(ip, params)
|
||||
if action == "fritzbox.backups":
|
||||
return _fb_backups(ip, params)
|
||||
if action == "fritzbox.backup_view":
|
||||
return _fb_backup_view(params)
|
||||
if action == "fritzbox.backup_delete":
|
||||
return _fb_backup_delete(params)
|
||||
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 "")
|
||||
@@ -1557,6 +1761,8 @@ def _do_fritzbox(action: str, params: dict) -> dict:
|
||||
return _fb_wlan(ip, user, pw, params)
|
||||
if action == "fritzbox.backup":
|
||||
return _fb_backup(ip, user, pw, params)
|
||||
if action == "fritzbox.restore":
|
||||
return _fb_restore(ip, user, pw, params)
|
||||
|
||||
if action == "fritzbox.hosts":
|
||||
p = _tr064(ip, user, pw, "urn:dslforum-org:service:Hosts:1",
|
||||
@@ -2463,6 +2669,27 @@ class Satellite:
|
||||
"timestamp": int(time.time() * 1000),
|
||||
})
|
||||
|
||||
async def _command_reply(self, payload: dict) -> None:
|
||||
req_id = payload.get("requestId", "")
|
||||
action = (payload.get("action") or "").strip()
|
||||
params = payload.get("params") or {}
|
||||
if payload.get("device") and "device" not in params:
|
||||
params["device"] = payload.get("device")
|
||||
# Geraeteliste nur scannen, wenn die Aktion sie wirklich braucht
|
||||
# (dial.launch loest ein Geraet auf, oder es wurde ein device-Ref
|
||||
# mitgegeben). http.get/http.post/wol arbeiten direkt mit url/mac —
|
||||
# ein voller LAN-Scan davor kostete nur unnoetig viele Sekunden.
|
||||
needs_devices = (action == "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({
|
||||
"type": "sat_result",
|
||||
"payload": {"requestId": req_id, "satellite": SATELLITE_ID, **result},
|
||||
"timestamp": int(time.time() * 1000),
|
||||
})
|
||||
|
||||
async def _handle(self, raw: str) -> None:
|
||||
try:
|
||||
msg = json.loads(raw)
|
||||
@@ -2525,25 +2752,9 @@ class Satellite:
|
||||
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"))
|
||||
or (action.startswith("fritzbox.")
|
||||
and not re.match(r"^\d{1,3}(\.\d{1,3}){3}$", str(params.get("ip") or ""))))
|
||||
devices = await self._scan() if needs_devices else self._devices
|
||||
result = await _control(action, params, devices)
|
||||
await self._send({
|
||||
"type": "sat_result",
|
||||
"payload": {"requestId": req_id, "satellite": SATELLITE_ID, **result},
|
||||
"timestamp": int(time.time() * 1000),
|
||||
})
|
||||
# Eigener Task: eine Aktion kann lange dauern (FritzBox wartet bis
|
||||
# 2 min auf die Taste, adb.install Minuten) — der Loop bleibt frei.
|
||||
asyncio.create_task(self._command_reply(payload))
|
||||
|
||||
elif mtype == "sat_creds_set":
|
||||
if not self._for_me(payload):
|
||||
|
||||
Reference in New Issue
Block a user