Files
ARIA-AGENT/satellite/satellite.py
T
duffyduckandClaude Opus 5.5 2cd8f1a0de feat: Kennwort je FritzBox-Sicherung — gespeichert, per 👁 anzeigen, kopieren
Bisher merkte sich der Satellit nur die QUELLE des Kennworts. Wurde das
Sicherungs-Kennwort spaeter geaendert, passte es nicht mehr zu aelteren
Sicherungen — auch "Wiederherstellen" haette das falsche genommen.

- Satellit: das benutzte Kennwort liegt verschluesselt (Fernet, wie der
  Zugangsdaten-Speicher) in der Begleitdatei; Liste/Ansicht geben es nie
  heraus. Wiederherstellen nimmt genau dieses. fritzbox.backup_password
  liefert es (aeltere Sicherungen: aktuelles der Quelle + Hinweis).
  Feld "kennwort" im Sicherungs-Ergebnis heisst jetzt "verschluesselt_mit".
- Diagnostic-Server: Ergebnisse mit Kennwort gehen nur an den Browser,
  der gefragt hat; ohne Fragenden werden sie verworfen, nie an alle.
- Brain: fritzbox.backup_password ist fuer ARIA gesperrt (Kennwoerter
  gehen nicht ans Sprachmodell).
- Diagnostic: 👁 in der Kennwort-Spalte → Zeile mit dem Kennwort und
  "Kopieren" — klappt auch ohne https (Fallback ueber execCommand).

Getestet: Satellit (Kennwort nicht im Klartext gespeichert, nach
Kennwort-Aenderung zeigt 👁 und nimmt Wiederherstellen das alte), Fenster
in headless Chromium ohne Clipboard-API.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-31 18:26:37 +01:00

3090 lines
139 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
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 threading
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.*,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 {}
_CRED_SECRETS = ("pass", "key", "community", "authKey", "privKey", "backupPass")
def _strong_password_problem(pw: str) -> str:
""""" wenn stark genug, sonst was fehlt. Regel fuer Sicherungs-Kennwoerter:
mind. 12 Zeichen, Gross- und Kleinbuchstabe, Ziffer, Sonderzeichen."""
pw = str(pw or "")
missing = []
if len(pw) < 12:
missing.append("mindestens 12 Zeichen")
if not re.search(r"[A-Z]", pw):
missing.append("ein Grossbuchstabe")
if not re.search(r"[a-z]", pw):
missing.append("ein Kleinbuchstabe")
if not re.search(r"\d", pw):
missing.append("eine Ziffer")
if not re.search(r"[^A-Za-z0-9]", pw):
missing.append("ein Sonderzeichen")
return ", ".join(missing)
def _creds_detail(d: dict, reveal: bool) -> dict:
"""Ein Cred-Eintrag fuer die Diagnostic: Benutzer, Port, Version … immer;
Geheimes nur als has_<feld>=True — ausser reveal (Auge im Formular)."""
out = {}
for k, v in (d or {}).items():
if isinstance(v, dict):
out[k] = _creds_detail(v, reveal)
elif k in _CRED_SECRETS:
out["has_" + k] = bool(v)
if reveal:
out[k] = v
else:
out[k] = v
return out
def _creds_keep_secrets(new: dict, old: dict) -> dict:
"""Beim Speichern: fehlt ein Geheimnis im neuen Eintrag (Feld unangetastet),
bleibt das alte. So muss man ein Passwort nicht neu tippen, um nur den
Benutzer zu aendern."""
out = dict(new)
for k, v in (old or {}).items():
if isinstance(v, dict) and isinstance(out.get(k), dict):
out[k] = _creds_keep_secrets(out[k], v)
elif k in _CRED_SECRETS and k not in out:
out[k] = v
return out
def _creds_public_summary(ip: str = "", reveal: bool = False) -> list:
"""Fuer sat_creds_list: welche Geraete welche Cred-Typen haben, dazu die
Felder OHNE Secrets (Benutzer, Port, SNMP-Version). Mit ip + reveal auch
die Secrets dieses einen Geraets."""
out = []
for dev_ip, entry in sorted(_CREDS.items()):
if ip and dev_ip != ip:
continue
types = [t for t in ("snmp", "http", "fritzbox", "ssh") if entry.get(t)]
out.append({"ip": dev_ip, "types": types,
"detail": {t: _creds_detail(entry[t], reveal and bool(ip)) for t in 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"]
# Router (FritzBox & Co.) melden sich per SSDP als
# InternetGatewayDevice — kein Medienabspieler.
if re.search(r"InternetGatewayDevice|FRITZ!Box", f"{entry['name']} {dev.get('model', '')}"):
entry["type"] = "router"
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
# Docker-interne Netze (Container auf DIESEM Rechner) sind keine
# Geraete — laeuft der Satellit auf einem Docker-Host (z.B. ARIAs
# VM), stuenden sonst alle Container in der Liste.
iface = parts[5] if len(parts) > 5 else ""
if iface.startswith(("docker", "br-", "veth")):
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 ────────────────────────────────────────────────────────
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 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()
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 == "fritzbox.backup_all":
return await loop.run_in_executor(None, _fb_backup_all, params, devices)
if action.startswith("fritzbox."):
# Geraetename ("Fritzbox Keller") → IP, wie bei den anderen Aktionen
t = _snmp_target(params)
ip = _resolve_target(t, devices) or t
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":
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=<videoId> (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, token: str = "") -> dict:
"""Ein TR-064-SOAP-Call. Gibt {ok, fields|error}. fields = alle <NewX>-Tags."""
import requests
from requests.auth import HTTPDigestAuth
body = "".join(f"<{k}>{v}</{k}>" for k, v in (args or {}).items())
envelope = (
'<?xml version="1.0"?>'
'<s:Envelope xmlns:s="http://schemas.xmlsoap.org/soap/envelope/" '
's:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/">'
# Zusatzbestaetigung: der Token aus X_AVM-DE_Auth SetConfig(start) MUSS
# bei GetState und beim Wiederholen der Aktion im SOAP-Header stehen
# (AVM "TR-064 Support – Authentication", Kap. 6.4) — sonst ordnet die
# Box den Tastendruck nicht zu und meldet "stopped".
+ (f'<s:Header><avm:token xmlns:avm="avm.de" s:mustUnderstand="1">{token}</avm:token></s:Header>'
if token else '')
+ '<s:Body>'
f'<u:{action} xmlns:u="{service}">{body}</u:{action}>'
'</s:Body></s:Envelope>'
)
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:
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)}
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 ""))}}
# 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", "")
token = st["fields"].get("NewToken", "")
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", "")
reasons = {"stopped": "die Box hat den Vorgang beendet, ohne Bestaetigung (Zeit abgelaufen?)",
"blocked": "zu viele Versuche — die Box sperrt bis zu 60 Minuten",
"anotherauthprocess": "es laeuft schon eine Bestaetigung fuer einen anderen Benutzer",
"failure": "interner Fehler der Box", "disabled": "die Zusatzbestaetigung ist aus"}
while time.time() < deadline:
if state == "authenticated":
logger.info("[fritzbox] %s: bestaetigt", ip)
return {"ok": True, "token": token}
if state in reasons and state != "disabled":
return {"ok": False, "error": f"Bestaetigung nicht erfolgt — Box meldet '{state}': {reasons[state]}."}
time.sleep(2)
g = _tr064(ip, user, pw, auth["type"], auth["control"], "GetState", token=token)
new = g.get("fields", {}).get("NewState", state) if g.get("ok") else state
if new != state:
logger.info("[fritzbox] %s: Bestaetigung %s → %s", ip, state, new)
state = new
_tr064(ip, user, pw, auth["type"], auth["control"], "SetConfig", {"NewAction": "stop"}, token=token)
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 ""))
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()}
hidden = [k for k, v in args.items() if "(verborgen)" in str(v)]
if hidden:
# Sonst setzte die Box den Platzhalter als neuen Schluessel/Kennwort
return {"ok": False, "error": (
f"{', '.join(hidden)}: das ist nur der Platzhalter fuer ein verborgenes Geheimnis, "
"nicht der echte Wert — so nicht senden. Geheimnisse aendert Stefan selbst "
"(FritzBox-Oberflaeche) oder er nennt den neuen Wert ausdruecklich.")}
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
# Gleiche Aktion, gleiche Werte — jetzt mit dem Token im SOAP-Header
r = _tr064(ip, user, pw, svc["type"], svc["control"], action, args, token=sf.get("token", ""))
if not r.get("ok") and r.get("upnp_error") == "867":
return {"ok": False, "error": "Die Box sperrt die Bestaetigung (zu viele Versuche) — bis zu 60 Minuten warten."}
if not r.get("ok") and r.get("upnp_error") == "868":
return {"ok": False, "error": "Die Box ist mit einer anderen Bestaetigung beschaeftigt — in etwa 2 Minuten nochmal."}
if not r.get("ok"):
return r
# Geheimnisse, die die Box auf Lese-Anfragen herausgibt (WLAN-Schluessel,
# Kennwoerter, PINs), gehen NICHT an ARIA — ihre Antworten landen beim
# Sprachmodell. Stefan sieht sie in der FritzBox-Oberflaeche.
fields = {k: ("•••• (verborgen)" if _FB_SECRET_FIELD.search(k) and v else v) for k, v in r["fields"].items()}
return {"ok": True, "result": {"ip": ip, "service": svc["short"], "action": action, **fields}}
_FB_SECRET_FIELD = re.compile(r"Passphrase|Password|PreSharedKey|WEPKey|KeyPassphrase|Secret|PIN|Passwd|SharedKey",
re.IGNORECASE)
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_keep(ip: str) -> int:
"""Wie viele Sicherungen je Box bleiben: Zugangsdaten (Feld "Maximale
Sicherungen") > FB_BACKUP_KEEP (Standard 10). Aelteste fliegen zuerst."""
try:
n = int(_creds_for(ip).get("fritzbox", {}).get("backupKeep") or 0)
except (TypeError, ValueError):
n = 0
return max(1, min(100, n)) if n else FB_BACKUP_KEEP
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
# Kennwort der Sicherung: ausdruecklich angegeben > im Satelliten hinterlegt
# (Feld "Sicherungs-Kennwort" — ARIA kennt es nicht) > FritzBox-Login.
stored = _creds_for(ip).get("fritzbox", {}).get("backupPass")
export_pw = str(params.get("password") or stored or pw)
weak = _strong_password_problem(export_pw)
if weak:
src = ("Das angegebene Kennwort" if params.get("password")
else "Das hinterlegte Sicherungs-Kennwort" if stored
else "Es ist kein Sicherungs-Kennwort hinterlegt, und das FritzBox-Login")
return {"ok": False, "error": (
f"{src} ist fuer eine Sicherung zu schwach (es fehlt: {weak}). Stefan soll im "
"Diagnostic beim Geraet (🔑) ein Sicherungs-Kennwort hinterlegen: mind. 12 Zeichen, "
"Gross-/Kleinbuchstabe, Ziffer, Sonderzeichen.")}
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
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"),
# Das benutzte Kennwort selbst, verschluesselt wie der Zugangsdaten-
# Speicher: passt auch noch, wenn das Sicherungs-Kennwort spaeter
# geaendert wird (Wiederherstellen + 👁 im Diagnostic).
"kennwort_enc": _creds_cipher().encrypt(export_pw.encode()).decode()}
with open(path[:-len(".export")] + ".json", "w", encoding="utf-8") as fh:
json.dump(meta, fh, ensure_ascii=False, indent=1)
keep = _fb_backup_keep(ip)
old = sorted(f for f in os.listdir(FB_BACKUP_DIR) if f.startswith(label + "_") and f.endswith(".export"))
for f in old[:-keep]:
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.
# 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 {}),
"verschluesselt_mit": ("das angegebene Kennwort" if params.get("password")
else "das im Satelliten hinterlegte Sicherungs-Kennwort" if stored
else "das FritzBox-Login-Kennwort"),
"hinweis": f"Liegt auf dem Satelliten, die letzten {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)."""
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)."}}
# ─── 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, with_secret: bool = False) -> dict:
try:
with open(path[:-len(".export")] + ".json", encoding="utf-8") as fh:
m = json.load(fh)
except Exception:
return {}
if not with_secret:
m.pop("kennwort_enc", None)
return m
def _fb_backup_password(ip: str, path: str) -> tuple:
"""(kennwort, hinweis) fuer genau diese Sicherung. Aeltere Sicherungen ohne
gespeichertes Kennwort: das AKTUELL hinterlegte der Quelle (mit Hinweis)."""
meta = _fb_backup_meta(path, with_secret=True)
if meta.get("kennwort_enc"):
try:
return _creds_cipher().decrypt(meta["kennwort_enc"].encode()).decode(), ""
except Exception:
pass
fb = _creds_for(ip).get("fritzbox", {})
src = meta.get("kennwort", "")
if src == "backupPass" and fb.get("backupPass"):
return fb["backupPass"], ("Aeltere Sicherung: das ist das AKTUELL hinterlegte Sicherungs-Kennwort — "
"wurde es seitdem geaendert, passt es nicht.")
if src == "login" and fb.get("pass"):
return fb["pass"], "Aeltere Sicherung: das ist das aktuelle FritzBox-Kennwort."
return "", "Fuer diese Sicherung ist kein Kennwort bekannt (eigens genanntes Kennwort)."
def _fb_backup_pw_action(ip: 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."}
pw, hint = _fb_backup_password(ip, path)
if not pw:
return {"ok": False, "error": hint}
return {"ok": True, "result": {"datei": os.path.basename(path), "kennwort": pw, "hinweis": hint}}
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,
"behalten": _fb_backup_keep(ip)}}
# 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")
saved_pw, _hint = _fb_backup_password(ip, path)
if params.get("password"):
export_pw = str(params["password"])
elif saved_pw and _fb_backup_meta(path, with_secret=True).get("kennwort_enc"):
export_pw = saved_pw # genau das Kennwort dieser Sicherung
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_reachable(ip: str, timeout: float = 3.0) -> bool:
try:
with socket.create_connection((ip, 49000), timeout=timeout):
return True
except OSError:
return False
def _fb_backup_all(params: dict, devices: list) -> dict:
"""Sichert ALLE FritzBoxen mit hinterlegtem Login, eine nach der anderen.
Nicht erreichbar / Fehler / Taste noetig → notieren und WEITER, nie abbrechen."""
names = {d.get("ip"): d.get("name") for d in devices if d.get("ip")}
boxes = [ip for ip, e in sorted(_CREDS.items()) if (e.get("fritzbox") or {}).get("pass")]
if not boxes:
return {"ok": False, "error": "An diesem Satelliten ist fuer keine FritzBox ein Login hinterlegt."}
report = []
for ip in boxes:
name = names.get(ip) or ip
if not _fb_reachable(ip):
report.append({"box": name, "ip": ip, "ok": False, "grund": "nicht erreichbar (aus/weg?) — uebersprungen"})
continue
fb = _creds_for(ip).get("fritzbox", {})
try:
r = _fb_backup(ip, str(fb.get("user") or ""), str(fb.get("pass") or ""),
{**params, "name": name, "no_file": True})
except Exception as exc:
r = {"ok": False, "error": f"Fehler: {exc}"}
if r.get("ok"):
report.append({"box": name, "ip": ip, "ok": True, "datei": os.path.basename(r["result"]["datei"])})
else:
grund = ("braucht eine Taste an der Box — uebersprungen (mit button_wait einzeln nachholen)"
if r.get("needs_button") else r.get("error") or "Fehler")
report.append({"box": name, "ip": ip, "ok": False, "grund": grund})
ok = sum(1 for x in report if x["ok"])
return {"ok": True, "result": {"gesichert": ok, "von": len(report), "boxen": report}}
# ─── Zusatzbestaetigung (Taste) der Box: Status + Umschalten ───
def _fb_second_factor_info(ip: str, user: str, pw: str) -> dict:
"""Ist die Tastenbestaetigung an? Und laesst sie sich per TR-064 schalten?
Umschalten geht nur, wenn die Firmware dafuer eine Aktion anbietet (eine
Set-Aktion des Auth-Dienstes mit einem Enable-Argument) — sonst nur in der
FritzBox-Oberflaeche."""
auth = _fb_find_service(ip, "X_AVM-DE_Auth")
if not auth:
return {"ok": True, "result": {"vorhanden": False, "an": False, "schaltbar": False,
"hinweis": "Die Box kennt keine Zusatzbestaetigung per Schnittstelle — Aktionen laufen ohne Taste."}}
acts = _fb_actions(ip, auth)
setter = next((a for a in acts if not _fb_is_read(a["name"]) and a["name"] != "SetConfig"
and any("enable" in x.lower() for x in a["in"])), None)
info = _tr064(ip, user, pw, auth["type"], auth["control"], "GetInfo") \
if any(a["name"] == "GetInfo" for a in acts) else {"ok": False}
f = info.get("fields", {}) if info.get("ok") else {}
en = f.get("NewEnabled", f.get("NewEnable"))
return {"ok": True, "result": {
"vorhanden": True, "an": None if en is None else en in ("1", "true"),
"schaltbar": bool(setter), "aktion": setter["name"] if setter else "",
"hinweis": ("" if setter else
"Laesst sich ueber die Schnittstelle nicht umschalten (Firmware bietet es nicht an). "
"Nur in der FritzBox-Oberflaeche: System → FRITZ!Box-Benutzer → Anmeldung im Heimnetz → "
"\"Ausfuehrung bestimmter Einstellungen und Funktionen zusaetzlich bestaetigen\". "
"Manche Firmware erlaubt das Abschalten gar nicht mehr.")}}
def _fb_second_factor_set(ip: str, user: str, pw: str, params: dict) -> dict:
inf = _fb_second_factor_info(ip, user, pw)["result"]
if not inf["vorhanden"]:
return {"ok": False, "error": inf["hinweis"]}
if not inf["schaltbar"]:
return {"ok": False, "not_switchable": True, "error": inf["hinweis"]}
auth = _fb_find_service(ip, "X_AVM-DE_Auth")
act = next(a for a in _fb_actions(ip, auth) if a["name"] == inf["aktion"])
arg = next(x for x in act["in"] if "enable" in x.lower())
on = str(params.get("enable")).lower() in ("1", "true", "an", "on", "ja")
return _fb_call(ip, user, pw, {"service": auth["short"], "action": act["name"],
"args": {arg: 1 if on else 0}, "confirmed": params.get("confirmed"),
"button_wait": params.get("button_wait")})
def _fb_backup_get(params: dict) -> dict:
"""Eine vorhandene Sicherung als Datei zurueckgeben (Download im Diagnostic,
bei ARIA landet sie unter /shared/uploads/ fuer den Chat)."""
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)."}
data = open(path, "rb").read()
if len(data) > 20 * 1024 * 1024:
return {"ok": False, "error": "Sicherung zu gross fuer die Uebertragung (> 20 MB)."}
return {"ok": True, "result": {"datei": os.path.basename(path), "bytes": len(data),
"filename": os.path.basename(path),
"file_b64": base64.b64encode(data).decode()}}
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
Verbindungsart (LAN/WLAN), LAN-Port und Geschwindigkeit.
params.find: nur Geraete, deren Name/IP/MAC das enthaelt."""
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)
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)
if action == "fritzbox.backup_get":
return _fb_backup_get(params)
if action == "fritzbox.backup_password": # nur fuers Diagnostic (Brain sperrt es)
return _fb_backup_pw_action(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 "")
if not pw:
return {"ok": False, "error":
"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.backup":
return _fb_backup(ip, user, pw, params)
if action == "fritzbox.restore":
return _fb_restore(ip, user, pw, params)
if action == "fritzbox.second_factor":
return _fb_second_factor_info(ip, user, pw)
if action == "fritzbox.second_factor_set":
return _fb_second_factor_set(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")
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 = []
find = str(params.get("find") or "").strip().lower()
for item in re.finditer(r"<Item>(.*?)</Item>", r.text, re.DOTALL):
blk = item.group(1)
def _t(tag):
m = re.search(rf"<{tag}>(.*?)</{tag}>", blk, re.DOTALL)
return m.group(1).strip() if m else ""
h = {"name": _t("HostName"), "ip": _t("IPAddress"),
"mac": _t("MACAddress"),
"active": _t("Active") in ("1", "true")}
# AVM-Zusatzfelder (je nach FRITZ!OS vorhanden): Verbindungsart,
# LAN-Port der Box (0 = kein direkter LAN-Port, z.B. WLAN/Mesh),
# Geschwindigkeit in Mbit/s, Gast-Netz.
iface = _t("InterfaceType")
if iface:
h["interface"] = iface
port = _to_int(_t("X_AVM-DE_Port"))
if port:
h["lan_port"] = port
speed = _to_int(_t("X_AVM-DE_Speed"))
if speed:
h["speed_mbit"] = speed
if _t("X_AVM-DE_Guest") in ("1", "true"):
h["guest"] = True
if find and not any(find in str(h.get(k, "")).lower() for k in ("name", "ip", "mac")):
continue
hosts.append(h)
return {"ok": True, "result": {"ip": ip, "count": len(hosts), "hosts": hosts,
"hinweis": "lan_port = LAN-Buchse DIESER Box. Haengt ein Geraet "
"hinter einem Switch/Repeater, steht hier dessen Port."}}
# 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 <paket>')."}
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}>(.*?)</{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 — bis er fertig ist (hoechstens SCAN_MAX_SEC).
# Frueher nur 6 s: direkt nach dem Start (Scan ~25 s, Liste noch
# leer) kam so "keine Geraete gefunden" zurueck.
deadline = time.time() + SCAN_MAX_SEC + 5
while self._scanning and time.time() < deadline:
await asyncio.sleep(0.2)
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 _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 in ("dial.launch", "fritzbox.backup_all") 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)
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
# 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):
return
ip = (payload.get("ip") or "").strip()
creds = payload.get("creds") or {}
ok = False
weak = _strong_password_problem((creds.get("fritzbox") or {}).get("backupPass")) \
if isinstance(creds, dict) and (creds.get("fritzbox") or {}).get("backupPass") else ""
if weak:
await self._send({"type": "sat_creds_result",
"payload": {"requestId": payload.get("requestId", ""),
"satellite": SATELLITE_ID, "ok": False, "ip": ip,
"error": f"Sicherungs-Kennwort zu schwach — es fehlt: {weak}."},
"timestamp": int(time.time() * 1000)})
return
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_keep_secrets(creds[t], entry.get(t) or {})
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
ip = (payload.get("ip") or "").strip()
reveal = bool(payload.get("reveal")) and bool(ip)
await self._send({"type": "sat_creds_list_result",
"payload": {"requestId": payload.get("requestId", ""),
"satellite": SATELLITE_ID,
"location": SATELLITE_LOCATION,
"ip": ip, "revealed": reveal,
"items": _creds_public_summary(ip, reveal)},
"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()