Hacker-Software/ARIA-AGENT
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 267 Wiki Activity
Files
v0.2.5.7
ARIA-AGENT/satellite/satellite.py
T
duffyduckandClaude Opus 5 6404b99510 feat(satellite): der Satelliten-Rechner steht selbst mit in der Geraeteliste 
Der Satellit laeuft auf einem ganz normalen Rechner — bei Stefan auf
duffy-desktop. Dort laufen typischerweise noch andere Dienste (FreeCAD-MCP,
Web-App, Dateiserver). Bisher war ausgerechnet dieser Rechner unsichtbar: man
findet sich nicht per mDNS/ARP selbst.

Er erscheint jetzt als normales Geraet — ansprechbar wie jedes andere, nur
zusaetzlich mit 🛰️ markiert und blau abgesetzt, damit klar ist, dass dort der
Satellit traegt. Abschaltbar ueber LIST_SELF=false, falls man den Traeger
bewusst heraushalten will.

Nebeneffekt, der ein echtes Problem loest: der eigene Rechner gilt damit als
"gefunden" und ist ohne Eintrag in TUNNEL_ALLOW tunnelbar. Genau der Fall, um
den es ging — der FreeCAD-MCP laeuft auf dem Satelliten-Host selbst.

Beim Zusammenfuehren bleibt die Markierung erhalten: findet mDNS denselben
Rechner nochmal, haette der Merge isSatellite sonst verschluckt. hostname kommt
aus demselben Grund mit in die erhaltenen Felder.

Dazu ein ✎-Knopf pro Geraet in der Diagnostic: Namen vergeben, damit ARIA
"duffy-desktop" ansprechen kann statt einer IP. Der Name haengt an der MAC und
ueberlebt einen DHCP-Wechsel.

8 Pruefungen: Selbst-Eintrag vollstaendig, Markierung ueberlebt den Merge,
mDNS-Infos bleiben erhalten, eigener Rechner ohne TUNNEL_ALLOW erlaubt, fremde
IP weiterhin blockiert.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-10-04 18:49:49 +02:00

2252 lines
96 KiB
Python
Raw Permalink Blame History

"""
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 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.info,fritzbox.hosts,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")
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 {}
def _creds_public_summary() -> list:
"""Fuer sat_creds_list: welche Geraete welche Cred-Typen haben — OHNE Secrets."""
out = []
for ip, entry in sorted(_CREDS.items()):
types = [t for t in ("snmp", "http", "fritzbox", "ssh") if entry.get(t)]
out.append({"ip": ip, "types": 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
try:
zc = Zeroconf()
listener = _Listener()
for t in MDNS_TYPES:
try:
ServiceBrowser(zc, t, listener)
except Exception:
pass
time.sleep(timeout)
except Exception as exc:
logger.warning("mDNS-Sweep-Fehler: %s", exc)
finally:
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"]
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
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 ────────────────────────────────────────────────────────
async def _control(action: str, params: dict, devices: list[dict]) -> dict:
"""Fuehrt eine Steuer-Aktion aus. Guards: CONTROL_ENABLED + Allowlist."""
if not CONTROL_ENABLED:
return {"ok": False, "error": "Steuerung ist an diesem Satelliten deaktiviert (CONTROL_ENABLED=false)."}
if action not in CONTROL_ALLOWLIST:
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 in ("fritzbox.info", "fritzbox.hosts"):
return await loop.run_in_executor(None, _do_fritzbox, action, params)
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) -> 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/"><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:
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}
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)."""
ip = _snmp_target(params)
if not ip:
return {"ok": False, "error": "ip/host erforderlich."}
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.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 = []
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) if m else ""
hosts.append({"name": _t("HostName"), "ip": _t("IPAddress"),
"mac": _t("MACAddress"),
"active": _t("Active") in ("1", "true")})
return {"ok": True, "result": {"ip": ip, "count": len(hosts), "hosts": hosts}}
# 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
# 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 (grob)
for _ in range(30):
await asyncio.sleep(0.2)
if not self._scanning:
break
return self._devices
self._scanning = True
try:
loop = asyncio.get_event_loop()
mdns = await loop.run_in_executor(None, _discover_mdns, DISCOVER_TIMEOUT_SEC)
ssdp = await loop.run_in_executor(None, _discover_ssdp, DISCOVER_TIMEOUT_SEC)
arp = await loop.run_in_executor(None, _discover_arp)
_self = [_self_device()] if LIST_SELF else []
_self = [d for d in _self if d]
self._devices = _apply_names(_merge_devices(mdns, ssdp, arp, _self))
if SNMP_DISCOVERY:
await self._enrich_snmp(self._devices)
await self._enrich_adb(self._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)
finally:
self._scanning = False
return self._devices
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 _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
req_id = payload.get("requestId", "")
devices = await self._scan(force=bool(payload.get("force")))
await self._send({
"type": "sat_devices",
"payload": {"requestId": req_id, "satellite": SATELLITE_ID,
"location": SATELLITE_LOCATION, "devices": devices,
"net": NET},
"timestamp": int(time.time() * 1000),
})
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
req_id = payload.get("requestId", "")
action = (payload.get("action") or "").strip()
params = payload.get("params") or {}
if payload.get("device") and "device" not in params:
params["device"] = payload.get("device")
# Geraeteliste nur scannen, wenn die Aktion sie wirklich braucht
# (dial.launch loest ein Geraet auf, oder es wurde ein device-Ref
# mitgegeben). http.get/http.post/wol arbeiten direkt mit url/mac —
# ein voller LAN-Scan davor kostete nur unnoetig viele Sekunden.
needs_devices = action == "dial.launch" or bool(params.get("device"))
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),
})
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
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[t]
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
await self._send({"type": "sat_creds_list_result",
"payload": {"requestId": payload.get("requestId", ""),
"satellite": SATELLITE_ID,
"location": SATELLITE_LOCATION,
"items": _creds_public_summary()},
"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()
Reference in New Issue View Git Blame Copy Permalink