""" 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 json import logging import os import re import socket import struct import time 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) RVS_TOKEN = os.environ.get("RVS_TOKEN", "") 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").split(",") if a.strip() ] 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") 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) ] CAPABILITIES = ["discover"] if CONTROL_ENABLED: CAPABILITIES += CONTROL_ALLOWLIST # ─── 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, "warning": warning} 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", } 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 for key in ("dialAppUrl", "mac", "model", "manufacturer"): 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 # ─── 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) 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= (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 _do_http(action: str, params: dict) -> dict: 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" r = requests.request(method, url, data=params.get("body"), headers=params.get("headers"), timeout=6) return {"ok": True, "result": {"status": r.status_code, "body": r.text[:2000]}} # ─── 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 _xml_tag(xml: str, tag: str) -> str: m = re.search(rf"<{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 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._devices = _merge_devices(mdns, ssdp, arp) self._devices_ts = time.time() logger.info("[scan] %d Geraete (mdns=%d ssdp=%d arp=%d)", len(self._devices), len(mdns), len(ssdp), len(arp)) finally: self._scanning = False return self._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) 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_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") devices = await self._scan() 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 _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 asyncio.create_task(self._periodic_scan()) backoff = 1 while True: proto = "wss" if RVS_TLS else "ws" url = f"{proto}://{RVS_HOST}:{RVS_PORT}?token={RVS_TOKEN}" try: logger.info("Verbinde mit RVS %s://%s:%s …", proto, RVS_HOST, RVS_PORT) async with websockets.connect(url, max_size=8 * 1024 * 1024, ping_interval=20, ping_timeout=20) as ws: self.ws = ws backoff = 1 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) finally: self.ws = None await asyncio.sleep(backoff) backoff = min(backoff * 2, 30) 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 alle=%s", NET.get("primary_ip"), NET.get("ips")) if NET.get("warning"): logger.warning("⚠ %s", NET["warning"]) try: asyncio.run(Satellite().run()) except KeyboardInterrupt: pass if __name__ == "__main__": main()