Overhaul of Reverse Tunnel Code

Data/Engine/services/API/devices/tunnel.py

@@ -15,7 +15,7 @@ from typing import Any, Dict, Optional, Tuple
 from flask import Blueprint, jsonify, request, session
 from itsdangerous import BadSignature, SignatureExpired, URLSafeTimedSerializer
 
-from ...WebSocket.Agent.ReverseTunnel import ReverseTunnelService
+from ...WebSocket.Agent.reverse_tunnel_orchestrator import ReverseTunnelService
 
 if False:  # pragma: no cover - import cycle hint for type checkers
     from .. import EngineServiceAdapters
@@ -103,6 +103,8 @@ def register_tunnel(app, adapters: "EngineServiceAdapters") -> None:
         agent_id = _normalize_text(body.get("agent_id"))
         protocol = _normalize_text(body.get("protocol") or "ps").lower() or "ps"
         domain = _normalize_text(body.get("domain") or protocol).lower() or protocol
+        if protocol == "ps" and domain == "ps":
+            domain = "remote-interactive-shell"
 
         if not agent_id:
             return jsonify({"error": "agent_id_required"}), 400
@@ -135,4 +137,33 @@ def register_tunnel(app, adapters: "EngineServiceAdapters") -> None:
         )
         return jsonify(summary), 200
 
+    @blueprint.route("/api/tunnel/<tunnel_id>", methods=["DELETE"])
+    def stop_tunnel(tunnel_id: str):
+        requirement = _require_login(app)
+        if requirement:
+            payload, status = requirement
+            return jsonify(payload), status
+
+        tunnel_id_norm = _normalize_text(tunnel_id)
+        if not tunnel_id_norm:
+            return jsonify({"error": "tunnel_id_required"}), 400
+
+        body = request.get_json(silent=True) or {}
+        reason = _normalize_text(body.get("reason") or "operator_stop")
+
+        tunnel_service = _get_tunnel_service(adapters)
+        stopped = False
+        try:
+            stopped = tunnel_service.stop_tunnel(tunnel_id_norm, reason=reason)
+        except Exception as exc:  # pragma: no cover - defensive guard
+            logger.debug("stop_tunnel failed tunnel_id=%s: %s", tunnel_id_norm, exc, exc_info=True)
+        if not stopped:
+            return jsonify({"error": "not_found"}), 404
+
+        service_log(
+            "reverse_tunnel",
+            f"lease stopped tunnel_id={tunnel_id_norm} reason={reason or '-'}",
+        )
+        return jsonify({"status": "stopped", "tunnel_id": tunnel_id_norm}), 200
+
     app.register_blueprint(blueprint)

Data/Engine/services/WebSocket/Agent/ReverseTunnelProtocols/__init__.py

@@ -1,6 +0,0 @@
-"""Protocol-specific helpers for Reverse Tunnel (Engine side)."""
-
-from .Powershell import PowershellChannelServer
-
-__all__ = ["PowershellChannelServer"]
-

Data/Engine/services/WebSocket/Agent/Reverse_Tunnels/__init__.py

@@ -0,0 +1,3 @@
+"""Namespace package for reverse tunnel domain handlers (Engine side)."""
+
+__all__ = ["remote_interactive_shell", "remote_management", "remote_video"]

Data/Engine/services/WebSocket/Agent/Reverse_Tunnels/remote_interactive_shell/Protocols/Bash.py

@@ -0,0 +1,78 @@
+"""Placeholder Bash channel server (Engine side)."""
+from __future__ import annotations
+
+from collections import deque
+
+
+class BashChannelServer:
+    """Stub Bash handler until the agent-side channel is implemented."""
+
+    protocol_name = "bash"
+
+    def __init__(self, bridge, service, *, channel_id: int = 1, frame_cls=None, close_frame_fn=None):
+        self.bridge = bridge
+        self.service = service
+        self.channel_id = channel_id
+        self.logger = service.logger.getChild(f"bash.{bridge.lease.tunnel_id}")
+        self._open_sent = False
+        self._ack_received = False
+        self._closed = False
+        self._output = deque()
+        self._frame_cls = frame_cls
+        self._close_frame_fn = close_frame_fn
+
+    def handle_agent_frame(self, frame) -> None:
+        # No-op placeholder; output collection for future Bash support.
+        try:
+            if frame.msg_type == 0x04:  # MSG_CHANNEL_ACK
+                self._ack_received = True
+        except Exception:
+            return
+
+    def open_channel(self, *, cols: int = 120, rows: int = 32) -> None:
+        self._open_sent = True
+        self.logger.info(
+            "bash channel placeholder open sent tunnel_id=%s channel_id=%s cols=%s rows=%s",
+            self.bridge.lease.tunnel_id,
+            self.channel_id,
+            cols,
+            rows,
+        )
+
+    def send_input(self, data: str) -> None:
+        # Placeholder: no agent-side Bash yet.
+        self.logger.info("bash placeholder send_input ignored tunnel_id=%s", self.bridge.lease.tunnel_id)
+
+    def send_resize(self, cols: int, rows: int) -> None:
+        # Placeholder: not implemented.
+        return
+
+    def close(self, code: int = 6, reason: str = "operator_close") -> None:
+        if self._closed:
+            return
+        self._closed = True
+        if callable(self._close_frame_fn):
+            try:
+                frame = self._close_frame_fn(self.channel_id, code, reason)
+                self.bridge.operator_to_agent(frame)
+            except Exception:
+                pass
+
+    def drain_output(self):
+        items = []
+        while self._output:
+            items.append(self._output.popleft())
+        return items
+
+    def status(self):
+        return {
+            "channel_id": self.channel_id,
+            "open_sent": self._open_sent,
+            "ack": self._ack_received,
+            "closed": self._closed,
+            "close_reason": None,
+            "close_code": None,
+        }
+
+
+__all__ = ["BashChannelServer"]

Data/Engine/services/WebSocket/Agent/Reverse_Tunnels/remote_interactive_shell/Protocols/Powershell.py

@@ -1,4 +1,4 @@
-"""Engine-side PowerShell tunnel channel helper."""
+"""Engine-side PowerShell tunnel channel helper (remote interactive shell domain)."""
 from __future__ import annotations
 
 import json
@@ -134,3 +134,6 @@ class PowershellChannelServer:
             "close_reason": self._close_reason,
             "close_code": self._close_code,
         }
+
+
+__all__ = ["PowershellChannelServer"]

Data/Engine/services/WebSocket/Agent/Reverse_Tunnels/remote_interactive_shell/Protocols/__init__.py

@@ -0,0 +1,6 @@
+"""Protocol handlers for remote interactive shell tunnels (Engine side)."""
+
+from .Powershell import PowershellChannelServer
+from .Bash import BashChannelServer
+
+__all__ = ["PowershellChannelServer", "BashChannelServer"]

Data/Engine/services/WebSocket/Agent/Reverse_Tunnels/remote_interactive_shell/__init__.py

@@ -0,0 +1,3 @@
+"""Domain handlers for remote interactive shells (PowerShell/Bash)."""
+
+__all__ = ["Protocols"]

Data/Engine/services/WebSocket/Agent/Reverse_Tunnels/remote_management/Protocols/SSH.py

@@ -0,0 +1,73 @@
+"""Placeholder SSH channel server (Engine side)."""
+from __future__ import annotations
+
+from collections import deque
+
+
+class SSHChannelServer:
+    protocol_name = "ssh"
+
+    def __init__(self, bridge, service, *, channel_id: int = 1, frame_cls=None, close_frame_fn=None):
+        self.bridge = bridge
+        self.service = service
+        self.channel_id = channel_id
+        self.logger = service.logger.getChild(f"ssh.{bridge.lease.tunnel_id}")
+        self._open_sent = False
+        self._ack_received = False
+        self._closed = False
+        self._output = deque()
+        self._frame_cls = frame_cls
+        self._close_frame_fn = close_frame_fn
+
+    def handle_agent_frame(self, frame) -> None:
+        try:
+            if frame.msg_type == 0x04:  # MSG_CHANNEL_ACK
+                self._ack_received = True
+        except Exception:
+            return
+
+    def open_channel(self, *, cols: int = 120, rows: int = 32) -> None:
+        self._open_sent = True
+        self.logger.info(
+            "ssh channel placeholder open sent tunnel_id=%s channel_id=%s cols=%s rows=%s",
+            self.bridge.lease.tunnel_id,
+            self.channel_id,
+            cols,
+            rows,
+        )
+
+    def send_input(self, data: str) -> None:
+        self.logger.info("ssh placeholder send_input ignored tunnel_id=%s", self.bridge.lease.tunnel_id)
+
+    def send_resize(self, cols: int, rows: int) -> None:
+        return
+
+    def close(self, code: int = 6, reason: str = "operator_close") -> None:
+        if self._closed:
+            return
+        self._closed = True
+        if callable(self._close_frame_fn):
+            try:
+                frame = self._close_frame_fn(self.channel_id, code, reason)
+                self.bridge.operator_to_agent(frame)
+            except Exception:
+                pass
+
+    def drain_output(self):
+        items = []
+        while self._output:
+            items.append(self._output.popleft())
+        return items
+
+    def status(self):
+        return {
+            "channel_id": self.channel_id,
+            "open_sent": self._open_sent,
+            "ack": self._ack_received,
+            "closed": self._closed,
+            "close_reason": None,
+            "close_code": None,
+        }
+
+
+__all__ = ["SSHChannelServer"]

Data/Engine/services/WebSocket/Agent/Reverse_Tunnels/remote_management/Protocols/WinRM.py

@@ -0,0 +1,73 @@
+"""Placeholder WinRM channel server (Engine side)."""
+from __future__ import annotations
+
+from collections import deque
+
+
+class WinRMChannelServer:
+    protocol_name = "winrm"
+
+    def __init__(self, bridge, service, *, channel_id: int = 1, frame_cls=None, close_frame_fn=None):
+        self.bridge = bridge
+        self.service = service
+        self.channel_id = channel_id
+        self.logger = service.logger.getChild(f"winrm.{bridge.lease.tunnel_id}")
+        self._open_sent = False
+        self._ack_received = False
+        self._closed = False
+        self._output = deque()
+        self._frame_cls = frame_cls
+        self._close_frame_fn = close_frame_fn
+
+    def handle_agent_frame(self, frame) -> None:
+        try:
+            if frame.msg_type == 0x04:  # MSG_CHANNEL_ACK
+                self._ack_received = True
+        except Exception:
+            return
+
+    def open_channel(self, *, cols: int = 120, rows: int = 32) -> None:
+        self._open_sent = True
+        self.logger.info(
+            "winrm channel placeholder open sent tunnel_id=%s channel_id=%s cols=%s rows=%s",
+            self.bridge.lease.tunnel_id,
+            self.channel_id,
+            cols,
+            rows,
+        )
+
+    def send_input(self, data: str) -> None:
+        self.logger.info("winrm placeholder send_input ignored tunnel_id=%s", self.bridge.lease.tunnel_id)
+
+    def send_resize(self, cols: int, rows: int) -> None:
+        return
+
+    def close(self, code: int = 6, reason: str = "operator_close") -> None:
+        if self._closed:
+            return
+        self._closed = True
+        if callable(self._close_frame_fn):
+            try:
+                frame = self._close_frame_fn(self.channel_id, code, reason)
+                self.bridge.operator_to_agent(frame)
+            except Exception:
+                pass
+
+    def drain_output(self):
+        items = []
+        while self._output:
+            items.append(self._output.popleft())
+        return items
+
+    def status(self):
+        return {
+            "channel_id": self.channel_id,
+            "open_sent": self._open_sent,
+            "ack": self._ack_received,
+            "closed": self._closed,
+            "close_reason": None,
+            "close_code": None,
+        }
+
+
+__all__ = ["WinRMChannelServer"]

Data/Engine/services/WebSocket/Agent/Reverse_Tunnels/remote_management/Protocols/__init__.py

@@ -0,0 +1,6 @@
+"""Protocol handlers for remote management tunnels (Engine side)."""
+
+from .SSH import SSHChannelServer
+from .WinRM import WinRMChannelServer
+
+__all__ = ["SSHChannelServer", "WinRMChannelServer"]

Data/Engine/services/WebSocket/Agent/Reverse_Tunnels/remote_management/__init__.py

@@ -0,0 +1,3 @@
+"""Domain handlers for remote management tunnels (SSH/WinRM)."""
+
+__all__ = ["Protocols"]

Data/Engine/services/WebSocket/Agent/Reverse_Tunnels/remote_video/Protocols/RDP.py

@@ -0,0 +1,73 @@
+"""Placeholder RDP channel server (Engine side)."""
+from __future__ import annotations
+
+from collections import deque
+
+
+class RDPChannelServer:
+    protocol_name = "rdp"
+
+    def __init__(self, bridge, service, *, channel_id: int = 1, frame_cls=None, close_frame_fn=None):
+        self.bridge = bridge
+        self.service = service
+        self.channel_id = channel_id
+        self.logger = service.logger.getChild(f"rdp.{bridge.lease.tunnel_id}")
+        self._open_sent = False
+        self._ack_received = False
+        self._closed = False
+        self._output = deque()
+        self._frame_cls = frame_cls
+        self._close_frame_fn = close_frame_fn
+
+    def handle_agent_frame(self, frame) -> None:
+        try:
+            if frame.msg_type == 0x04:  # MSG_CHANNEL_ACK
+                self._ack_received = True
+        except Exception:
+            return
+
+    def open_channel(self, *, cols: int = 120, rows: int = 32) -> None:
+        self._open_sent = True
+        self.logger.info(
+            "rdp channel placeholder open sent tunnel_id=%s channel_id=%s cols=%s rows=%s",
+            self.bridge.lease.tunnel_id,
+            self.channel_id,
+            cols,
+            rows,
+        )
+
+    def send_input(self, data: str) -> None:
+        self.logger.info("rdp placeholder send_input ignored tunnel_id=%s", self.bridge.lease.tunnel_id)
+
+    def send_resize(self, cols: int, rows: int) -> None:
+        return
+
+    def close(self, code: int = 6, reason: str = "operator_close") -> None:
+        if self._closed:
+            return
+        self._closed = True
+        if callable(self._close_frame_fn):
+            try:
+                frame = self._close_frame_fn(self.channel_id, code, reason)
+                self.bridge.operator_to_agent(frame)
+            except Exception:
+                pass
+
+    def drain_output(self):
+        items = []
+        while self._output:
+            items.append(self._output.popleft())
+        return items
+
+    def status(self):
+        return {
+            "channel_id": self.channel_id,
+            "open_sent": self._open_sent,
+            "ack": self._ack_received,
+            "closed": self._closed,
+            "close_reason": None,
+            "close_code": None,
+        }
+
+
+__all__ = ["RDPChannelServer"]

Data/Engine/services/WebSocket/Agent/Reverse_Tunnels/remote_video/Protocols/VNC.py

@@ -0,0 +1,73 @@
+"""Placeholder VNC channel server (Engine side)."""
+from __future__ import annotations
+
+from collections import deque
+
+
+class VNCChannelServer:
+    protocol_name = "vnc"
+
+    def __init__(self, bridge, service, *, channel_id: int = 1, frame_cls=None, close_frame_fn=None):
+        self.bridge = bridge
+        self.service = service
+        self.channel_id = channel_id
+        self.logger = service.logger.getChild(f"vnc.{bridge.lease.tunnel_id}")
+        self._open_sent = False
+        self._ack_received = False
+        self._closed = False
+        self._output = deque()
+        self._frame_cls = frame_cls
+        self._close_frame_fn = close_frame_fn
+
+    def handle_agent_frame(self, frame) -> None:
+        try:
+            if frame.msg_type == 0x04:  # MSG_CHANNEL_ACK
+                self._ack_received = True
+        except Exception:
+            return
+
+    def open_channel(self, *, cols: int = 120, rows: int = 32) -> None:
+        self._open_sent = True
+        self.logger.info(
+            "vnc channel placeholder open sent tunnel_id=%s channel_id=%s cols=%s rows=%s",
+            self.bridge.lease.tunnel_id,
+            self.channel_id,
+            cols,
+            rows,
+        )
+
+    def send_input(self, data: str) -> None:
+        self.logger.info("vnc placeholder send_input ignored tunnel_id=%s", self.bridge.lease.tunnel_id)
+
+    def send_resize(self, cols: int, rows: int) -> None:
+        return
+
+    def close(self, code: int = 6, reason: str = "operator_close") -> None:
+        if self._closed:
+            return
+        self._closed = True
+        if callable(self._close_frame_fn):
+            try:
+                frame = self._close_frame_fn(self.channel_id, code, reason)
+                self.bridge.operator_to_agent(frame)
+            except Exception:
+                pass
+
+    def drain_output(self):
+        items = []
+        while self._output:
+            items.append(self._output.popleft())
+        return items
+
+    def status(self):
+        return {
+            "channel_id": self.channel_id,
+            "open_sent": self._open_sent,
+            "ack": self._ack_received,
+            "closed": self._closed,
+            "close_reason": None,
+            "close_code": None,
+        }
+
+
+__all__ = ["VNCChannelServer"]

Data/Engine/services/WebSocket/Agent/Reverse_Tunnels/remote_video/Protocols/WebRTC.py

@@ -0,0 +1,73 @@
+"""Placeholder WebRTC channel server (Engine side)."""
+from __future__ import annotations
+
+from collections import deque
+
+
+class WebRTCChannelServer:
+    protocol_name = "webrtc"
+
+    def __init__(self, bridge, service, *, channel_id: int = 1, frame_cls=None, close_frame_fn=None):
+        self.bridge = bridge
+        self.service = service
+        self.channel_id = channel_id
+        self.logger = service.logger.getChild(f"webrtc.{bridge.lease.tunnel_id}")
+        self._open_sent = False
+        self._ack_received = False
+        self._closed = False
+        self._output = deque()
+        self._frame_cls = frame_cls
+        self._close_frame_fn = close_frame_fn
+
+    def handle_agent_frame(self, frame) -> None:
+        try:
+            if frame.msg_type == 0x04:  # MSG_CHANNEL_ACK
+                self._ack_received = True
+        except Exception:
+            return
+
+    def open_channel(self, *, cols: int = 120, rows: int = 32) -> None:
+        self._open_sent = True
+        self.logger.info(
+            "webrtc channel placeholder open sent tunnel_id=%s channel_id=%s cols=%s rows=%s",
+            self.bridge.lease.tunnel_id,
+            self.channel_id,
+            cols,
+            rows,
+        )
+
+    def send_input(self, data: str) -> None:
+        self.logger.info("webrtc placeholder send_input ignored tunnel_id=%s", self.bridge.lease.tunnel_id)
+
+    def send_resize(self, cols: int, rows: int) -> None:
+        return
+
+    def close(self, code: int = 6, reason: str = "operator_close") -> None:
+        if self._closed:
+            return
+        self._closed = True
+        if callable(self._close_frame_fn):
+            try:
+                frame = self._close_frame_fn(self.channel_id, code, reason)
+                self.bridge.operator_to_agent(frame)
+            except Exception:
+                pass
+
+    def drain_output(self):
+        items = []
+        while self._output:
+            items.append(self._output.popleft())
+        return items
+
+    def status(self):
+        return {
+            "channel_id": self.channel_id,
+            "open_sent": self._open_sent,
+            "ack": self._ack_received,
+            "closed": self._closed,
+            "close_reason": None,
+            "close_code": None,
+        }
+
+
+__all__ = ["WebRTCChannelServer"]

Data/Engine/services/WebSocket/Agent/Reverse_Tunnels/remote_video/Protocols/__init__.py

@@ -0,0 +1,7 @@
+"""Protocol handlers for remote video tunnels (Engine side)."""
+
+from .WebRTC import WebRTCChannelServer
+from .RDP import RDPChannelServer
+from .VNC import VNCChannelServer
+
+__all__ = ["WebRTCChannelServer", "RDPChannelServer", "VNCChannelServer"]

Data/Engine/services/WebSocket/Agent/Reverse_Tunnels/remote_video/__init__.py

@@ -0,0 +1,3 @@
+"""Domain handlers for remote video/desktop tunnels (RDP/VNC/WebRTC)."""
+
+__all__ = ["Protocols"]

Data/Engine/services/WebSocket/Agent/reverse_tunnel_orchestrator.py

@@ -1,5 +1,5 @@
 # ======================================================
-# Data\Engine\services\WebSocket\Agent\ReverseTunnel.py
+# Data\Engine\services\WebSocket\Agent\reverse_tunnel_orchestrator.py
 # Description: Async reverse tunnel scaffolding (Engine side) providing lease management, domain limits, and placeholders for WebSocket listeners.
 #
 # API Endpoints (if applicable): None
@@ -29,7 +29,13 @@ from typing import Callable, Deque, Dict, Iterable, List, Optional, Tuple
 from collections import deque
 from threading import Thread
 
-from .ReverseTunnelProtocols import PowershellChannelServer
+from .Reverse_Tunnels.remote_interactive_shell.Protocols.Powershell import PowershellChannelServer
+from .Reverse_Tunnels.remote_interactive_shell.Protocols.Bash import BashChannelServer
+from .Reverse_Tunnels.remote_management.Protocols.SSH import SSHChannelServer
+from .Reverse_Tunnels.remote_management.Protocols.WinRM import WinRMChannelServer
+from .Reverse_Tunnels.remote_video.Protocols.VNC import VNCChannelServer
+from .Reverse_Tunnels.remote_video.Protocols.RDP import RDPChannelServer
+from .Reverse_Tunnels.remote_video.Protocols.WebRTC import WebRTCChannelServer
 
 try:  # websockets is added to engine requirements
     import websockets
@@ -234,10 +240,15 @@ class DomainPolicy:
     """Enforce per-domain concurrency and defaults."""
 
     DEFAULT_LIMITS = {
-        "ps": 1,
+        # New domain lanes
+        "remote-interactive-shell": 2,
+        "remote-management": 1,
+        "remote-video": 2,
+        # Protocol-specific fallbacks (for backward compatibility / legacy callers)
+        "ps": 2,
         "rdp": 1,
         "vnc": 1,
-        "webrtc": 1,
+        "webrtc": 2,
         "ssh": None,  # Unlimited
         "winrm": None,  # Unlimited
     }
@@ -459,7 +470,17 @@ class ReverseTunnelService:
         self._bridges: Dict[str, "TunnelBridge"] = {}
         self._port_servers: Dict[int, asyncio.AbstractServer] = {}
         self._agent_sockets: Dict[str, "websockets.WebSocketServerProtocol"] = {}
-        self._ps_servers: Dict[str, PowershellChannelServer] = {}
+        self.protocol_registry = {
+            "ps": PowershellChannelServer,
+            "powershell": PowershellChannelServer,
+            "bash": BashChannelServer,
+            "ssh": SSHChannelServer,
+            "winrm": WinRMChannelServer,
+            "vnc": VNCChannelServer,
+            "rdp": RDPChannelServer,
+            "webrtc": WebRTCChannelServer,
+        }
+        self._protocol_servers: Dict[str, object] = {}
 
     def _ensure_loop(self) -> None:
         if self._running and self._loop:
@@ -504,6 +525,12 @@ class ReverseTunnelService:
                 self._loop.call_soon_threadsafe(asyncio.create_task, websocket.close())
             except Exception:
                 pass
+        for tunnel_id in list(self._bridges.keys()):
+            try:
+                self.release_bridge(tunnel_id, reason="service_stop")
+            except Exception:
+                pass
+        self._protocol_servers.clear()
         for lease in list(self.lease_manager.all_leases()):
             self.lease_manager.release(lease.tunnel_id, reason="service_stop")
         if self._sweeper_task:
@@ -561,9 +588,9 @@ class ReverseTunnelService:
             raise ValueError("unknown_tunnel")
         bridge = self.ensure_bridge(lease)
         bridge.attach_operator(operator_id)
-        if lease.domain.lower() == "ps":
+        if (lease.protocol or "").lower() in {"ps", "powershell"}:
             try:
-                server = self.ensure_ps_server(tunnel_id)
+                server = self.ensure_protocol_server(tunnel_id)
                 if server:
                     server.open_channel()
             except Exception:
@@ -629,6 +656,22 @@ class ReverseTunnelService:
         lease.expires_at = expires_at
         return token
 
+    def stop_tunnel(self, tunnel_id: str, *, reason: str = "operator_stop", code: int = CLOSE_AGENT_SHUTDOWN) -> bool:
+        """Request a graceful stop for a tunnel (operator-driven)."""
+
+        lease = self.lease_manager.get(tunnel_id)
+        if lease is None:
+            return False
+        server = self.get_protocol_server(tunnel_id)
+        if server and hasattr(server, "close"):
+            try:
+                server.close(code=code, reason=reason)
+            except Exception:
+                self.logger.debug("protocol server close failed tunnel_id=%s", tunnel_id, exc_info=True)
+        self._push_stop_to_agent(lease, reason=reason)
+        self.release_bridge(tunnel_id, reason=reason)
+        return True
+
     def _push_start_to_agent(self, lease: TunnelLease) -> None:
         """Notify the target agent about the new lease over Socket.IO (best-effort)."""
 
@@ -658,6 +701,26 @@ class ReverseTunnelService:
         except Exception:
             self.logger.debug("Failed to emit reverse_tunnel_start for tunnel_id=%s", lease.tunnel_id, exc_info=True)
 
+    def _push_stop_to_agent(self, lease: TunnelLease, *, reason: str = "operator_stop") -> None:
+        """Notify the agent to tear down a tunnel (best-effort)."""
+
+        if not self._socketio:
+            return
+        try:
+            self._socketio.emit(
+                "reverse_tunnel_stop",
+                {"tunnel_id": lease.tunnel_id, "reason": reason},
+                namespace="/",
+            )
+            self.audit_logger.info(
+                "lease_push_stop tunnel_id=%s agent_id=%s reason=%s",
+                lease.tunnel_id,
+                lease.agent_id,
+                reason or "-",
+            )
+        except Exception:
+            self.logger.debug("Failed to emit reverse_tunnel_stop for tunnel_id=%s", lease.tunnel_id, exc_info=True)
+
     def lease_summary(self, lease: TunnelLease) -> Dict[str, object]:
         return {
             "tunnel_id": lease.tunnel_id,
@@ -892,7 +955,7 @@ class ReverseTunnelService:
                     pass
 
     def _dispatch_agent_frame(self, tunnel_id: str, frame: TunnelFrame) -> None:
-        server = self._ps_servers.get(tunnel_id)
+        server = self._protocol_servers.get(tunnel_id)
         if not server:
             return
         try:
@@ -1136,33 +1199,39 @@ class ReverseTunnelService:
             self._bridges[lease.tunnel_id] = bridge
         return bridge
 
-    def ensure_ps_server(self, tunnel_id: str) -> Optional[PowershellChannelServer]:
-        server = self._ps_servers.get(tunnel_id)
+    def ensure_protocol_server(self, tunnel_id: str) -> Optional[object]:
+        server = self._protocol_servers.get(tunnel_id)
         if server:
             return server
         lease = self.lease_manager.get(tunnel_id)
         if lease is None:
             return None
+        handler_cls = self.protocol_registry.get((lease.protocol or "").lower())
+        if handler_cls is None:
+            return None
         bridge = self.ensure_bridge(lease)
-        server = PowershellChannelServer(
-            bridge=bridge,
-            service=self,
-            frame_cls=TunnelFrame,
-            close_frame_fn=close_frame,
-        )
-        self._ps_servers[tunnel_id] = server
+        try:
+            server = handler_cls(
+                bridge=bridge,
+                service=self,
+                frame_cls=TunnelFrame,
+                close_frame_fn=close_frame,
+            )
+        except TypeError:
+            server = handler_cls(bridge=bridge, service=self)
+        self._protocol_servers[tunnel_id] = server
         return server
 
-    def get_ps_server(self, tunnel_id: str) -> Optional[PowershellChannelServer]:
-        return self._ps_servers.get(tunnel_id)
+    def get_protocol_server(self, tunnel_id: str) -> Optional[object]:
+        return self._protocol_servers.get(tunnel_id)
 
     def release_bridge(self, tunnel_id: str, *, reason: str = "bridge_released") -> None:
         bridge = self._bridges.pop(tunnel_id, None)
         if bridge:
             bridge.stop(reason=reason)
-        if tunnel_id in self._ps_servers:
+        if tunnel_id in self._protocol_servers:
             try:
-                self._ps_servers.pop(tunnel_id, None)
+                self._protocol_servers.pop(tunnel_id, None)
             except Exception:
                 pass
 

Data/Engine/services/WebSocket/__init__.py

@@ -20,7 +20,7 @@ from flask_socketio import SocketIO
 
 from ...database import initialise_engine_database
 from ...server import EngineContext
-from .Agent.ReverseTunnel import (
+from .Agent.reverse_tunnel_orchestrator import (
     ReverseTunnelService,
     TunnelBridge,
     decode_frame,
@@ -406,8 +406,8 @@ def register_realtime(socket_server: SocketIO, context: EngineContext) -> None:
         tunnel_id = _operator_sessions.get(sid)
         if not tunnel_id:
             return None, None, {"error": "not_joined"}
-        server = tunnel_service.ensure_ps_server(tunnel_id)
-        if server is None:
+        server = tunnel_service.ensure_protocol_server(tunnel_id)
+        if server is None or not hasattr(server, "open_channel"):
             return None, tunnel_id, {"error": "ps_unsupported"}
         return server, tunnel_id, None
 
@@ -489,4 +489,9 @@ def register_realtime(socket_server: SocketIO, context: EngineContext) -> None:
     @socket_server.on("disconnect", namespace=tunnel_namespace)
     def _ws_tunnel_disconnect():
         sid = request.sid
-        _operator_sessions.pop(sid, None)
+        tunnel_id = _operator_sessions.pop(sid, None)
+        if tunnel_id and tunnel_id not in _operator_sessions.values():
+            try:
+                tunnel_service.stop_tunnel(tunnel_id, reason="operator_socket_disconnect")
+            except Exception as exc:
+                logger.debug("ws_tunnel_disconnect stop_tunnel failed tunnel_id=%s: %s", tunnel_id, exc, exc_info=True)