feat(satellite): Port-Tunnel ins fremde Netz + Geraete-Namen

ARIA sitzt woanders und kommt nicht ins Buero-Netz. Der Satellit kann 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 einen FreeCAD-MCP, eine Web-App, ein NAS-Interface.

Entscheidend fuer die Brauchbarkeit: die Bridge macht daraus einen GANZ
NORMALEN lokalen Port (aria-bridge:19000...). Dadurch funktioniert alles, was
sowieso TCP spricht — MCP-Client, curl, Browser — ohne Sonder-API. Jede
eingehende Verbindung bekommt einen eigenen Tunnel, parallele Verbindungen
sind also kein Problem (ein Browser oeffnet gern sechs).

Geraete-Namen: "duffy-desktop" ist ansprechbar, "192.168.1.47" nicht. Stefan
kann jedem gefundenen Geraet einen Namen geben, ARIA adressiert danach.
Schluessel ist die MAC, nicht die IP — per DHCP wandert die IP, die MAC bleibt.

Absicherung, weil das ein Weg in ein fremdes Netz ist:
- TUNNEL_ENABLED default AUS, pro Satellit.
- Ohne TUNNEL_ALLOW nur Ziele, die bei der Suche GEFUNDEN wurden. Der Tunnel
  kann damit nichts ansteuern, was nicht ohnehin im LAN sichtbar ist — und
  nichts im Internet.
- Obergrenze gleichzeitiger Tunnel + Reaper fuer ungenutzte (sonst bleibt eine
  vergessene Verbindung offen, bis der Satellit neu startet).
- Faehigkeit 'tunnel' wird im sat_hello gemeldet, damit Diagnostic und ARIA
  sehen, ob der Satellit das ueberhaupt erlaubt.

Neue RVS-Typen eingetragen (sat_tunnel_*, sat_name_*) — ohne das verwirft die
Allow-List sie still, und nichts funktioniert ohne Fehlermeldung.

16 Pruefungen gegen BEIDE echten Implementierungen (nur der RVS dazwischen ist
kurzgeschlossen): Namensaufloesung, Freigabe-Regeln, fremder Satellit
ignoriert, abgeschaltet blockt, echter Byte-Durchlauf (PONG:HALLO), unbekanntes
Ziel schliesst sauber statt zu haengen, Liste/Schliessen/Doppel-Schliessen.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

bridge/aria_bridge.py

@@ -765,6 +765,17 @@ class ARIABridge:
         # "task": asyncio.Task}. Wir bruecken rohes RFB-TCP (QEMU-VNC auf dem
         # Host) <-> RVS (vnc_data/vnc_input, Base64-in-JSON).
         self._vnc_sessions: dict[str, dict] = {}
+        # ── Satelliten-Port-Tunnel ───────────────────────────────────
+        # ARIA sitzt hier, der Dienst steht in einem fremden LAN. Der Satellit
+        # haelt dorthin die Verbindung; wir machen daraus einen GANZ NORMALEN
+        # lokalen Port. Damit kann alles im Stack einfach draufzugreifen
+        # (MCP-Client, curl, Browser) statt eine Sonder-API zu brauchen.
+        #   _tunnel_listeners: port -> {"server", "satellite", "target", "port"}
+        #   _tunnel_conns:     tunnelId -> {"writer", "opened" (Future)}
+        self._tunnel_listeners: dict[int, dict] = {}
+        self._tunnel_conns: dict[str, dict] = {}
+        self._TUNNEL_PORT_BASE = int(os.environ.get("TUNNEL_PORT_BASE", "19000"))
+        self._TUNNEL_PORT_MAX = int(os.environ.get("TUNNEL_PORT_MAX", "19099"))
         # VNC-Host: das aria-net-Gateway (dort bindet der Brain QEMUs VNC).
         # host.docker.internal zeigt faelschlich auf docker0 (172.17.0.1) → refused.
         self._vnc_host: str = (os.environ.get("ARIA_VNC_HOST")
@@ -3599,6 +3610,42 @@ class ARIABridge:
             asyncio.create_task(self._check_desktop(session, port))
             return
 
+        elif msg_type == "sat_tunnel_opened":
+            # Antwort des Satelliten auf unseren Verbindungswunsch.
+            tid = (payload.get("tunnelId") or "").strip()
+            conn = self._tunnel_conns.get(tid)
+            fut = conn.get("opened") if conn else None
+            if fut is not None and not fut.done():
+                fut.set_result(payload)
+            return
+
+        elif msg_type == "sat_tunnel_data":
+            # LAN -> lokaler Socket. Der Satellit schickt dasselbe Feld in die
+            # andere Richtung; unterschieden wird ueber die tunnelId, die nur
+            # hier bekannt ist.
+            tid = (payload.get("tunnelId") or "").strip()
+            conn = self._tunnel_conns.get(tid)
+            if conn:
+                try:
+                    conn["writer"].write(base64.b64decode(payload.get("b64") or ""))
+                except Exception as exc:
+                    logger.warning("[tunnel] schreiben (%s) fehlgeschlagen: %s", tid[:8], exc)
+            return
+
+        elif msg_type == "sat_tunnel_closed":
+            tid = (payload.get("tunnelId") or "").strip()
+            conn = self._tunnel_conns.pop(tid, None)
+            if conn:
+                fut = conn.get("opened")
+                if fut is not None and not fut.done():
+                    fut.set_result({"ok": False,
+                                    "error": payload.get("reason") or "geschlossen"})
+                try:
+                    conn["writer"].close()
+                except Exception:
+                    pass
+            return
+
         elif msg_type == "vnc_open":
             session = str(payload.get("session") or "main")
             try:
@@ -4700,6 +4747,45 @@ class ARIABridge:
                 elif method in ("GET", "POST") and path == "/internal/worker-list":
                     # Diagnostic/Brain fragt: welche Compute-Worker sind online (Flotte).
                     await _send_response(writer, 200, {"ok": True, "workers": self._worker_list()})
+                elif method == "POST" and path == "/internal/tunnel":
+                    # Brain-Tool: Port aus einem Satelliten-LAN lokal verfuegbar
+                    # machen. body: {op: open|close|list, satellite, target,
+                    #                port, localPort?}
+                    try:
+                        data = json.loads(body.decode("utf-8", "ignore"))
+                    except Exception as exc:
+                        await _send_response(writer, 400, {"error": f"bad json: {exc}"})
+                        return
+                    op = (data.get("op") or "open").strip()
+                    if op == "list":
+                        await _send_response(writer, 200, {"ok": True, "tunnels": self.tunnel_list()})
+                    elif op == "close":
+                        res = await self.tunnel_stop(int(data.get("localPort") or 0))
+                        await _send_response(writer, 200, res)
+                    else:
+                        res = await self.tunnel_listen(
+                            (data.get("satellite") or "").strip(),
+                            (data.get("target") or "").strip(),
+                            int(data.get("port") or 0),
+                            int(data.get("localPort") or 0))
+                        await _send_response(writer, 200, res)
+                elif method == "POST" and path == "/internal/satellite-name":
+                    # Brain/Diagnostic: Geraetenamen vergeben. body: {satellite, mac|ip, name}
+                    try:
+                        data = json.loads(body.decode("utf-8", "ignore"))
+                    except Exception as exc:
+                        await _send_response(writer, 400, {"error": f"bad json: {exc}"})
+                        return
+                    await self._send_to_rvs({
+                        "type": "sat_name_set",
+                        "payload": {"satellite": data.get("satellite") or "",
+                                    "mac": data.get("mac") or "",
+                                    "ip": data.get("ip") or "",
+                                    "name": data.get("name") or "",
+                                    "requestId": uuid.uuid4().hex},
+                        "timestamp": int(time.time() * 1000),
+                    })
+                    await _send_response(writer, 200, {"ok": True})
                 elif method == "POST" and path == "/internal/host-list":
                     # Brain fragt: welche Host-Agenten (Rechner) sind online + Capabilities.
                     await _send_response(writer, 200, {"ok": True, "hosts": self._host_list()})
@@ -4903,6 +4989,107 @@ class ARIABridge:
             "timestamp": int(time.time() * 1000),
         })
 
+    # ── Satelliten-Port-Tunnel ──────────────────────────────────────
+
+    async def tunnel_listen(self, satellite: str, target: str, port: int,
+                            local_port: int = 0) -> dict:
+        """Macht aus <satellite>/<target>:<port> einen lokalen Port.
+
+        Jede eingehende Verbindung auf dem lokalen Port wird einzeln durch den
+        Satelliten zum Ziel gefuehrt — mehrere parallele Verbindungen sind also
+        kein Problem (ein Browser oeffnet gern sechs).
+        """
+        if local_port and not (self._TUNNEL_PORT_BASE <= local_port <= self._TUNNEL_PORT_MAX):
+            return {"ok": False, "error":
+                    f"localPort muss zwischen {self._TUNNEL_PORT_BASE} und "
+                    f"{self._TUNNEL_PORT_MAX} liegen."}
+        if not local_port:
+            for p in range(self._TUNNEL_PORT_BASE, self._TUNNEL_PORT_MAX + 1):
+                if p not in self._tunnel_listeners:
+                    local_port = p
+                    break
+            if not local_port:
+                return {"ok": False, "error": "Keine freien Tunnel-Ports mehr."}
+        if local_port in self._tunnel_listeners:
+            await self.tunnel_stop(local_port)
+
+        async def on_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+            tid = uuid.uuid4().hex
+            opened: asyncio.Future = asyncio.get_event_loop().create_future()
+            self._tunnel_conns[tid] = {"writer": writer, "opened": opened}
+            await self._send_to_rvs({
+                "type": "sat_tunnel_open",
+                "payload": {"satellite": satellite, "tunnelId": tid,
+                            "target": target, "port": port},
+                "timestamp": int(time.time() * 1000),
+            })
+            try:
+                res = await asyncio.wait_for(opened, timeout=20)
+            except asyncio.TimeoutError:
+                res = {"ok": False, "error": "Satellit antwortet nicht"}
+            if not res.get("ok"):
+                logger.warning("[tunnel] %s:%d nicht erreichbar: %s",
+                               target, port, res.get("error"))
+                self._tunnel_conns.pop(tid, None)
+                try:
+                    writer.close()
+                except Exception:
+                    pass
+                return
+            try:
+                while True:
+                    data = await reader.read(32768)
+                    if not data:
+                        break
+                    await self._send_to_rvs({
+                        "type": "sat_tunnel_data",
+                        "payload": {"tunnelId": tid,
+                                    "b64": base64.b64encode(data).decode("ascii")},
+                        "timestamp": int(time.time() * 1000),
+                    })
+            except Exception:
+                pass
+            finally:
+                self._tunnel_conns.pop(tid, None)
+                await self._send_to_rvs({
+                    "type": "sat_tunnel_close",
+                    "payload": {"satellite": satellite, "tunnelId": tid},
+                    "timestamp": int(time.time() * 1000),
+                })
+                try:
+                    writer.close()
+                except Exception:
+                    pass
+
+        try:
+            server = await asyncio.start_server(on_client, "0.0.0.0", local_port)
+        except Exception as exc:
+            return {"ok": False, "error": f"Port {local_port} nicht belegbar: {exc}"}
+        self._tunnel_listeners[local_port] = {
+            "server": server, "satellite": satellite, "target": target, "port": port}
+        logger.info("[tunnel] lokal :%d -> %s/%s:%d", local_port, satellite, target, port)
+        # Adresse aus Sicht der ANDEREN Container (Brain/Proxy liegen im selben
+        # Docker-Netz) — damit ARIA weiss, was sie ansprechen soll.
+        return {"ok": True, "localPort": local_port,
+                "address": f"aria-bridge:{local_port}",
+                "satellite": satellite, "target": target, "port": port}
+
+    async def tunnel_stop(self, local_port: int) -> dict:
+        entry = self._tunnel_listeners.pop(int(local_port), None)
+        if not entry:
+            return {"ok": False, "error": f"Auf Port {local_port} laeuft kein Tunnel."}
+        try:
+            entry["server"].close()
+        except Exception:
+            pass
+        logger.info("[tunnel] lokal :%s geschlossen", local_port)
+        return {"ok": True, "localPort": int(local_port)}
+
+    def tunnel_list(self) -> list[dict]:
+        return [{"localPort": p, "address": f"aria-bridge:{p}",
+                 "satellite": e["satellite"], "target": e["target"], "port": e["port"]}
+                for p, e in sorted(self._tunnel_listeners.items())]
+
     async def _vnc_open(self, session: str, port: int) -> None:
         """Oeffnet die TCP-Verbindung zur QEMU-VNC und streamt RFB-Bytes als
         vnc_data ueber RVS. Bestehende Session wird vorher geschlossen."""