feat(satellite): FritzBox-Bestaetigung per Taste (2 min) + Sicherung in den Chat
Der Export der Einstellungen scheiterte mit "HTTP 500" — dahinter steckt UPnP-Fehler 866: die Box will die Aktion zusaetzlich am Geraet bestaetigt. - _tr064 meldet jetzt den UPnP-Fehlercode statt nur "HTTP 500". - 866 → ohne button_wait sofort zurueck mit needs_button (ARIA fragt Stefan, ob er an der Box steht). Mit button_wait=true: X_AVM-DE_Auth SetConfig (start), GetState alle 2 s bis FB_BUTTON_WAIT_SEC (120 s); "authenticated" → Aktion erneut, sonst Timeout "Taste nicht gedrueckt, nichts ausgefuehrt" (Bestaetigung wird gestoppt). - fritzbox.backup: X_AVM-DE_GetConfigFile mit Kennwort (Standard: Login- Kennwort), Download mit Digest-Auth, Kopie auf dem Satelliten (die letzten 10 je Box) und Datei zurueck an ARIA. - Brain: Datei aus dem Ergebnis landet in /shared/uploads/, ARIA legt sie mit [FILE: …] in den Chat; Timeout 200 s bei button_wait; Beschreibung erklaert den Tasten-Ablauf. Getestet gegen nachgebaute Box: ohne Bestaetigung → needs_button, Taste nach 3 s → Sicherung geladen, keine Taste → Timeout. Gegen eine echte Box (X_AVM-DE_Auth-Ablauf) noch nicht. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
+112
-1
@@ -860,7 +860,10 @@ async def _control(action: str, params: dict, devices: list[dict]) -> dict:
|
||||
# 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})
|
||||
extra = {"ip": ip}
|
||||
if t and t != ip and not params.get("name"):
|
||||
extra["name"] = t # z.B. Dateiname der Sicherung: fritzbox-keller_…
|
||||
return await loop.run_in_executor(None, _do_fritzbox, action, {**params, **extra})
|
||||
if action == "ssh.exec":
|
||||
return await loop.run_in_executor(None, _do_ssh, params)
|
||||
if action == "adb.devices":
|
||||
@@ -1212,6 +1215,11 @@ def _tr064(ip: str, user: str, pw: str, service: str, control: str,
|
||||
if r.status_code == 401:
|
||||
return {"ok": False, "error": "TR-064 Auth fehlgeschlagen (FritzBox-Login pruefen)."}
|
||||
if r.status_code != 200:
|
||||
code = re.search(r"<errorCode>(\d+)</errorCode>", r.text or "")
|
||||
desc = re.search(r"<errorDescription>(.*?)</errorDescription>", r.text or "", re.DOTALL)
|
||||
if code:
|
||||
return {"ok": False, "upnp_error": code.group(1),
|
||||
"error": f"FritzBox lehnt ab: {code.group(1)} {(desc.group(1).strip() if desc else '')}".strip()}
|
||||
return {"ok": False, "error": f"TR-064 HTTP {r.status_code}"}
|
||||
fields = {m.group(1): m.group(2) for m in
|
||||
re.finditer(r"<(New[^>/]+)>(.*?)</\1>", r.text, re.DOTALL)}
|
||||
@@ -1327,6 +1335,42 @@ def _fb_describe(ip: str, params: dict) -> dict:
|
||||
+ (f" Gekuerzt auf {limit} — mit 'filter' eingrenzen (z.B. 'WLAN', 'Dect', 'Hosts')." if len(lines) > limit else ""))}}
|
||||
|
||||
|
||||
# Manche Aktionen (z.B. Einstellungen sichern) verlangt die Box zusaetzlich am
|
||||
# Geraet bestaetigt: Fehler 866 = "zweiter Faktor noetig". Ablauf laut AVM:
|
||||
# X_AVM-DE_Auth SetConfig(start) → jemand drueckt eine Taste an der Box (oder
|
||||
# waehlt den DTMF-Code am Telefon) → GetState meldet "authenticated" → Aktion
|
||||
# erneut ausfuehren.
|
||||
FB_BUTTON_WAIT_SEC = float(os.environ.get("FB_BUTTON_WAIT_SEC", "120") or "120")
|
||||
_FB_2FA_CODES = ("866",)
|
||||
|
||||
|
||||
def _fb_second_factor(ip: str, user: str, pw: str) -> dict:
|
||||
"""Startet die Bestaetigung am Geraet und wartet bis FB_BUTTON_WAIT_SEC."""
|
||||
auth = _fb_find_service(ip, "X_AVM-DE_Auth")
|
||||
if not auth:
|
||||
return {"ok": False, "error": "Die Box bietet keine Bestaetigung per TR-064 an (X_AVM-DE_Auth fehlt)."}
|
||||
st = _tr064(ip, user, pw, auth["type"], auth["control"], "SetConfig", {"NewAction": "start"})
|
||||
if not st.get("ok"):
|
||||
return {"ok": False, "error": f"Bestaetigung nicht startbar: {st.get('error')}"}
|
||||
methods = st["fields"].get("NewMethods", "")
|
||||
logger.info("[fritzbox] %s: warte bis %.0fs auf Bestaetigung am Geraet (%s)", ip, FB_BUTTON_WAIT_SEC, methods)
|
||||
deadline = time.time() + FB_BUTTON_WAIT_SEC
|
||||
state = st["fields"].get("NewState", "")
|
||||
while time.time() < deadline:
|
||||
if state == "authenticated":
|
||||
return {"ok": True}
|
||||
if state in ("blocked", "failure", "stopped"):
|
||||
return {"ok": False, "error": f"Bestaetigung abgebrochen (Box meldet '{state}')."}
|
||||
time.sleep(2)
|
||||
g = _tr064(ip, user, pw, auth["type"], auth["control"], "GetState")
|
||||
state = g.get("fields", {}).get("NewState", state) if g.get("ok") else state
|
||||
_tr064(ip, user, pw, auth["type"], auth["control"], "SetConfig", {"NewAction": "stop"})
|
||||
return {"ok": False, "timeout": True,
|
||||
"error": f"Timeout: Die Taste an der Box wurde nicht innerhalb von {FB_BUTTON_WAIT_SEC:.0f} s "
|
||||
f"gedrueckt — abgebrochen, nichts ausgefuehrt."
|
||||
+ (f" (Moeglich waeren: {methods})" if methods else "")}
|
||||
|
||||
|
||||
def _fb_call(ip: str, user: str, pw: str, params: dict) -> dict:
|
||||
action = str(params.get("action") or "").strip()
|
||||
svc = _fb_find_service(ip, str(params.get("service") or ""))
|
||||
@@ -1344,11 +1388,76 @@ def _fb_call(ip: str, user: str, pw: str, params: dict) -> dict:
|
||||
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") and r.get("upnp_error") in _FB_2FA_CODES:
|
||||
if params.get("button_wait") not in (True, "true"):
|
||||
return {"ok": False, "needs_button": True, "error": (
|
||||
f"'{svc['short']} {action}' muss an der FritzBox bestaetigt werden (Taste druecken). "
|
||||
"Frag Stefan, ob er jetzt an der Box steht. Erst nach seinem Ja erneut mit "
|
||||
f"button_wait=true aufrufen — ab dann hat er {FB_BUTTON_WAIT_SEC:.0f} s Zeit, eine Taste "
|
||||
"an der Box zu druecken.")}
|
||||
sf = _fb_second_factor(ip, user, pw)
|
||||
if not sf.get("ok"):
|
||||
return sf
|
||||
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"]}}
|
||||
|
||||
|
||||
FB_BACKUP_DIR = os.environ.get("FB_BACKUP_DIR", "/data/fritzbox-backups")
|
||||
FB_BACKUP_KEEP = int(os.environ.get("FB_BACKUP_KEEP", "10") or "10")
|
||||
|
||||
|
||||
def _fb_backup(ip: str, user: str, pw: str, params: dict) -> dict:
|
||||
"""Einstellungen der Box sichern (.export) — landet auf dem Satelliten unter
|
||||
FB_BACKUP_DIR, die letzten FB_BACKUP_KEEP je Box bleiben. Kennwort der
|
||||
Sicherung = params.password, sonst das FritzBox-Login (zum Wiederherstellen
|
||||
noetig). Braucht meist die Taste an der Box (button_wait)."""
|
||||
import requests
|
||||
from requests.auth import HTTPDigestAuth
|
||||
export_pw = str(params.get("password") or pw)
|
||||
r = _fb_call(ip, user, pw, {"service": "DeviceConfig", "action": "X_AVM-DE_GetConfigFile",
|
||||
"args": {"X_AVM-DE_Password": export_pw},
|
||||
"button_wait": params.get("button_wait")})
|
||||
if not r.get("ok"):
|
||||
return r
|
||||
url = r["result"].get("NewX_AVM-DE_ConfigFileUrl", "")
|
||||
if not url:
|
||||
return {"ok": False, "error": "Die Box lieferte keine Download-Adresse fuer die Sicherung."}
|
||||
try:
|
||||
d = requests.get(url, auth=HTTPDigestAuth(user, pw), timeout=60, verify=False)
|
||||
d.raise_for_status()
|
||||
except Exception as exc:
|
||||
return {"ok": False, "error": f"Sicherung nicht herunterladbar: {exc}"}
|
||||
os.makedirs(FB_BACKUP_DIR, exist_ok=True)
|
||||
label = _slug(str(params.get("name") or ip))
|
||||
fname = f"{label}_{time.strftime('%Y-%m-%d_%H%M')}.export"
|
||||
path = os.path.join(FB_BACKUP_DIR, fname)
|
||||
with open(path, "wb") as fh:
|
||||
fh.write(d.content)
|
||||
try:
|
||||
os.chmod(path, 0o600)
|
||||
except OSError:
|
||||
pass
|
||||
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
|
||||
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 ""
|
||||
return {"ok": True, "result": {
|
||||
"ip": ip, "datei": path, "bytes": len(d.content), "auf": "Satellit",
|
||||
**({"file_b64": file_b64, "filename": fname} if file_b64 else {}),
|
||||
"kennwort": "das angegebene Kennwort" if params.get("password") else "das FritzBox-Login-Kennwort",
|
||||
"hinweis": f"Liegt auf dem Satelliten, die letzten {FB_BACKUP_KEEP} je Box bleiben. "
|
||||
"Zum Wiederherstellen in der FritzBox: System → Sicherung → Wiederherstellen, mit dem Kennwort."}}
|
||||
|
||||
|
||||
def _fb_wlan(ip: str, user: str, pw: str, params: dict) -> dict:
|
||||
"""Alle WLANs der Box (2,4/5/6 GHz, Gast) mit SSID und an/aus; mit
|
||||
index + enable schalten (braucht confirmed=true)."""
|
||||
@@ -1399,6 +1508,8 @@ def _do_fritzbox(action: str, params: dict) -> dict:
|
||||
return _fb_call(ip, user, pw, params)
|
||||
if action == "fritzbox.wlan":
|
||||
return _fb_wlan(ip, user, pw, params)
|
||||
if action == "fritzbox.backup":
|
||||
return _fb_backup(ip, user, pw, params)
|
||||
|
||||
if action == "fritzbox.hosts":
|
||||
p = _tr064(ip, user, pw, "urn:dslforum-org:service:Hosts:1",
|
||||
|
||||
Reference in New Issue
Block a user