Keep remote-desktop receive threads alive and handshake off the accept thread

Malformed input or a bad WebSocket frame killed the host's receive thread
without stop(), leaving a handler that held a client slot, and the viewer's
thread died without telling the GUI. The TLS, WS and auth handshake ran on
the single accept thread, so one slow peer blocked every other viewer; each
connection now gets its own bounded handshake thread. Signaling timeouts and
hang-ups are raised as SignalingError.

Author: JE-Chen

CHANGELOG.md

@@ -150,6 +150,13 @@ it shipped into a `## [x.y.z] - date` section of their own; the tag's
 
 ### Fixed
 
+- **Remote desktop keeps serving after bad input.** A malformed INPUT
+  message or WebSocket frame no longer kills the host's receive thread while
+  its viewer keeps a client slot, nor the viewer's thread without an error
+  callback; a slow or silent peer no longer blocks other viewers from
+  connecting (each connection is handshaken on its own thread); and a
+  signaling timeout or hang-up is reported as `SignalingError`.
+
 - **Generated code cannot run what an action file smuggles in.** Codegen
   emits a parameter as a keyword argument only when its name is a plain
   identifier, and the Robot target carries the actions base64-encoded.

architecture_explore.md

@@ -20,7 +20,7 @@ iOS（WebDriverAgent）。核心能力是滑鼠／鍵盤控制、影像辨識、
 | 指標 | 數值 |
 | --- | ---: |
 | Python 模組總數（含周邊子專案） | 1,045 |
-| 程式碼總行數 | 143,227 |
+| 程式碼總行數 | 143,290 |
 | `je_auto_control/utils/` 子套件數 | 310 |
 | `AC_*` 動作指令數（`known_commands()` 實測） | 773 |
 | 套件門面 `__all__` 公開名稱數 | 1,239 |
@@ -512,14 +512,14 @@ socket server 有 8 MiB 讀取上限與 30 秒 handler timeout。
 
 ### 5.4.10 遠端桌面與 USB
 
-> 6 個套件、約 18,363 行。
+> 6 個套件、約 18,426 行。
 
 | 模組 | 行數 | 職責 |
 | --- | ---: | --- |
 | `utils/admin/` | 349 | 多主機管理主控台：平行輪詢 N 個 AutoControl REST 端點 |
 | `utils/config_sync/` | 246 | 透過訊令伺服器做跨機器設定同步 |
 | `utils/device_matrix/` | 138 | 行動裝置矩陣：同一 action list 於多台裝置平行執行 |
-| `utils/remote_desktop/` | 12,265 | **遠端桌面子系統**（56 檔／11.7K LOC）：TCP／WebSocket／WebRTC 三條傳輸路徑、主機與檢視端、訊令伺服器、TURN／中繼、多檢視者、錄影、信任清單、TOTP、稽核鏈 |
+| `utils/remote_desktop/` | 12,328 | **遠端桌面子系統**（56 檔／11.7K LOC）：TCP／WebSocket／WebRTC 三條傳輸路徑、主機與檢視端、訊令伺服器、TURN／中繼、多檢視者、錄影、信任清單、TOTP、稽核鏈 |
 | `utils/usb/` | 4,422 | 跨平台 USB 列舉／熱插拔／裝置直通（WinUSB、IOKit、libusb 後端 + ACL + WebRTC DataChannel 通道） |
 | `utils/usbip/` | 943 | USB/IP 線路協定主機端（協定封包、TCP 伺服器、libusb URB 後端） |
 
@@ -739,25 +739,25 @@ socket server 有 8 MiB 讀取上限與 30 秒 handler timeout。
 | `rate_limit.py` | 48 | 工具呼叫的 token bucket 限流。 |
 | `__main__.py` | 88 | `je_auto_control_mcp` console script 進入點。 |
 
-#### `utils/remote_desktop/`（12,265 行／56 檔）
+#### `utils/remote_desktop/`（12,328 行／56 檔）
 
 三條傳輸路徑並存：**TCP**（JPEG 影格）、**WebSocket**（同協定換傳輸）、**WebRTC**（aiortc 視訊 + DataChannel）。
 
 | 檔案 | 行數 | 職責 |
 | --- | ---: | --- |
 | `webrtc_host.py` | 716 | WebRTC 主機：串流螢幕視訊並接受檢視端輸入;session 生命週期、DataChannel 接線、檔案收發。 |
 | `webrtc_viewer.py` | 672 | WebRTC 檢視端：接收視訊並送出輸入。 |
-| `host.py` | 634 | TCP 主機：接受迴圈、TLS 包裝、連線／認證握手、音訊與剪貼簿廣播、檔案推送、單次 token。 |
+| `host.py` | 667 | TCP 主機：接受迴圈、TLS 包裝、連線／認證握手、音訊與剪貼簿廣播、檔案推送、單次 token。 |
 | `viewer.py` | 634 | TCP 檢視端。 |
 | `host_service.py` | 558 | 無頭 WebRTC 主機執行器 + 多平台服務安裝器。 |
-| `host_client.py` | 406 | TCP 主機的每連線處理器：一個檢視端一個實例,擁有它的認證交換、sender／audio／receiver 三條執行緒,以及入站訊息的路由表。 |
+| `host_client.py` | 417 | TCP 主機的每連線處理器：一個檢視端一個實例,擁有它的認證交換、sender／audio／receiver 三條執行緒,以及入站訊息的路由表。 |
 | `registry.py` | 370 | `AC_remote_*` 指令使用的行程級單例。 |
 | `webrtc_transport.py` | 369 | 共用 WebRTC 管線：asyncio 橋接執行緒、螢幕視訊軌、設定。 |
 | `multi_viewer.py` | 339 | 每個連入檢視端各跑一個 `WebRTCDesktopHost` 的協調器。 |
 | `signaling_server.py` | 313 | 獨立的 WebRTC SDP 交換 rendezvous 服務。 |
 | `audit_log.py` | 302 | SQLite 雜湊鏈稽核記錄。 |
 | `host_capture.py` | 297 | TCP 主機的影格與游標產生：螢幕列舉、監視器索引轉擷取區域、預設 JPEG／游標 provider,以及 `FrameProductionMixin`（游標輪詢、擷取迴圈、上線編碼）。 |
-| `ws_protocol.py` | 277 | 最小 RFC 6455 WebSocket 框架與握手。 |
+| `ws_protocol.py` | 284 | 最小 RFC 6455 WebSocket 框架與握手。 |
 | `file_transfer.py` | 323 | 分塊檔案傳輸。 |
 | `relay.py` | 307 | NAT 穿透失敗時的 TCP 中繼。 |
 | `fingerprint.py` | 246 | TOFU 主機指紋驗證。 |
@@ -775,10 +775,10 @@ socket server 有 8 MiB 讀取上限與 30 秒 handler timeout。
 | `webrtc_stats.py` | 167 | 把 aiortc 的 `RTCStats` 報告輪詢成精簡 dict。 |
 | `connect_coordinator.py` | 149 | 由使用者輸入的目標決定該用哪條傳輸。 |
 | `adaptive_bitrate.py` | 148 | 依統計調整主機擷取 FPS。 |
-| `signaling_client.py` | 145 | 純標準庫的訊令用戶端。 |
+| `signaling_client.py` | 151 | 純標準庫的訊令用戶端。 |
 | `trust_list.py` | 140 | 自動接受的檢視端信任清單。 |
 | `webrtc_inspector.py` | 138 | 行程級的 `StatsSnapshot` 滾動視窗。 |
-| `input_dispatch.py` | 133 | 在主機端套用輸入訊息。 |
+| `input_dispatch.py` | 139 | 在主機端套用輸入訊息。 |
 | `session_recorder.py` | 134 | 以 PyAV 把 WebRTC 影格錄成 mp4。 |
 | `totp.py` | 140 | RFC 6238 TOTP（零外部相依）。 |
 | `file_sync.py` | 141 | 輪詢式資料夾鏡像。 |
@@ -1061,7 +1061,7 @@ socket 預設綁 `127.0.0.1`；資源一律用 `with`。
 | --- | ---: | ---: |
 | `gui/` | 91 | 26,794 |
 | `utils/mcp_server/` | 31 | 17,538 |
-| `utils/remote_desktop/` | 56 | 12,265 |
+| `utils/remote_desktop/` | 56 | 12,328 |
 | `utils/executor/` | 7 | 9,238 |
 | `utils/usb/` | 17 | 4,422 |
 | `je_auto_control/`（頂層 3 檔） | 3 | 2,388 |
@@ -1080,5 +1080,5 @@ socket 預設綁 `127.0.0.1`；資源一律用 `with`。
 | `autocontrol-lsp/` | 8 | 744 |
 | `utils/hotkey/` | 7 | 783 |
 | 其餘模組（約 286 個 `utils/` 子套件 + `android/`／`ios/`／周邊小工具） | 674 | 48,508 |
-| **總計** | **1,039** | **143,162** |
+| **總計** | **1,039** | **143,225** |
 

docs/updates/2026-09.md

@@ -398,3 +398,13 @@ Index and query commands: [README.md](README.md). New entries go at the end.
 - **Tests**: `test_codegen_injection.py` (new, 4; all fail on the previous commit), `test_network_edge_audit.py` (new, 6; it does not import on the previous commit, `atomic_write_bytes` being new). `test_codegen.py`, `test_usbip.py` and `test_r3_rdusb_usbip.py` updated as described.
 - **Files**: `je_auto_control/utils/{codegen/codegen,usbip/server,tls_acme/keys,json_store/json_store,remote_desktop/signaling_server,remote_desktop/relay}.py`, `test/unit_test/headless/{test_codegen_injection,test_network_edge_audit,test_codegen,test_usbip,test_r3_rdusb_usbip}.py`, `CHANGELOG.md`, `architecture_explore.md` (figures).
 - **Open items**: none.
+
+## U-20260923-30 · 2026-09-23 · Remote-desktop robustness: dead receive threads, blocked accept thread, signaling errors · #incident #remote-desktop
+
+- **What**: The robustness findings of the remote-desktop network audit (the security ones are U-20260923-29).
+- **Receive threads**: (1) an INPUT message with a missing field (`KeyError`), an infinite or non-numeric coordinate (`OverflowError` / `ValueError` from `int()`), or an unknown key or button name (the wrappers' `AutoControlException`) escaped the host's catch; the receive thread died without `stop()`, so the handler still counted as an authenticated client, ignored all later input and — with `max_clients=1` — locked out every new viewer. `dispatch_input` turns conversion errors into `InputDispatchError`, the host catches the wrappers' framework errors, and `_recv_loop` calls `stop()` before letting anything else propagate. (2) `WsProtocolError` was a bare `RuntimeError`, so one TEXT or fragmented frame killed the WebSocket host's receive thread the same way and the viewer's silently (`on_error` never ran); it is a `ProtocolError`, which both loops already end a connection on.
+- **Accept thread**: the TLS handshake, the WS upgrade and authentication ran on the one accept thread, each read bounded by a 60 s *per-read* timeout, so a peer sending a byte just inside each timeout kept everyone else from connecting (8.6 s and counting in the reproduction). Each connection is handshaken on its own `rd-handshake` thread, at most 32 at once; beyond that new connections are closed.
+- **Signaling client**: a read timeout (`TimeoutError`) or a server hanging up (`RemoteDisconnected`) is no `URLError`, so they left `_request` as themselves while the GUI workers catch only `SignalingError`; they are wrapped too.
+- **Tests**: `test_rd_robustness_audit.py` (new, 8; all fail on the previous commit). The 360 existing remote-desktop tests pass unchanged, including the two that pin the stop-during-handshake window.
+- **Files**: `je_auto_control/utils/remote_desktop/{host,host_client,input_dispatch,ws_protocol,signaling_client}.py`, `test/unit_test/headless/test_rd_robustness_audit.py`, `CHANGELOG.md`, `architecture_explore.md` (figures).
+- **Open items**: none.

docs/updates/README.md

@@ -58,6 +58,7 @@ In the same commit: delete the item from `Progress.md`, add a `#done` entry here
 
 | ID | Date | Title | Tags | Batch |
 |---|---|---|---|---|
+| U-20260923-30 | 2026-09-23 | Remote-desktop robustness: dead receive threads, blocked accept thread, signaling errors | #incident #remote-desktop | [2026-09](2026-09.md) |
 | U-20260923-29 | 2026-09-23 | Codegen injection, USB/IP device scope, TLS key writes, signaling and relay limits | #incident #security | [2026-09](2026-09.md) |
 | U-20260923-28 | 2026-09-23 | Shared JSON stores lock and re-read across processes | #done #stores | [2026-09](2026-09.md) |
 | U-20260923-27 | 2026-09-23 | CLI / hotkey / recording audit: exit codes, lost recordings, wrong virtual keys | #incident #cli #hotkey | [2026-09](2026-09.md) |
@@ -134,7 +135,7 @@ In the same commit: delete the item from `Progress.md`, add a `#done` entry here
 
 | File | Period | Entries |
 |---|---|---:|
-| [2026-09.md](2026-09.md) | 2026-09 | 45 |
+| [2026-09.md](2026-09.md) | 2026-09 | 46 |
 | [2026-08-f.md](2026-08-f.md) | 2026-08 | 1 |
 | [2026-08-e.md](2026-08-e.md) | 2026-08 | 2 |
 | [2026-08-d.md](2026-08-d.md) | 2026-08 | 2 |

je_auto_control/utils/remote_desktop/host.py

@@ -69,6 +69,17 @@ def _validate_host_args(token: str, fps: float, quality: int) -> None:
         raise ValueError("quality must be in [1, 95]")
 
 
+#: Connections being handshaken at once; beyond this new ones are closed.
+_MAX_HANDSHAKES = 32
+
+
+def _close_quietly(sock: socket.socket) -> None:
+    try:
+        sock.close()
+    except OSError:
+        pass
+
+
 class RemoteDesktopHost(FrameProductionMixin):
     """Stream the screen to authenticated viewers and apply their input.
 
@@ -176,6 +187,7 @@ def __init__(
         self._lifecycle_lock = threading.RLock()
         self._clients: List[_ClientHandler] = []
         self._clients_lock = threading.Lock()
+        self._handshake_slots = threading.BoundedSemaphore(_MAX_HANDSHAKES)
         self._frame_cond = threading.Condition()
         self._latest_frame: Optional[bytes] = None
         self._latest_seq = 0
@@ -546,45 +558,66 @@ def _accept_loop(self, stop: threading.Event) -> None:
             except OSError:
                 return
             if not self._ip_allowed(address):
-                try:
-                    client_sock.close()
-                except OSError:
-                    pass
+                _close_quietly(client_sock)
                 continue
+            # The TLS handshake, the WS upgrade and authentication each wait
+            # up to _AUTH_TIMEOUT_S *per read*, and ran here, on the one
+            # accept thread: a peer sending a byte just inside each timeout
+            # kept every other viewer from connecting. Each connection now
+            # gets its own thread, and a bounded number of them at once.
+            if not self._handshake_slots.acquire(blocking=False):
+                autocontrol_logger.info(
+                    "remote_desktop dropping %s: too many handshakes in progress",
+                    address,
+                )
+                _close_quietly(client_sock)
+                continue
+            threading.Thread(
+                target=self._handshake, args=(client_sock, address, stop),
+                name="rd-handshake", daemon=True,
+            ).start()
+
+    def _handshake(self, client_sock: socket.socket, address,
+                   stop: threading.Event) -> None:
+        """Open the channel, register the client, then authenticate it."""
+        try:
             channel = self._open_channel(client_sock, address)
             if channel is None:
-                continue
+                return
             handler = _ClientHandler(self, channel, address)
-            # Prune handlers whose viewer already disconnected *before* the
-            # capacity check below. Otherwise a client table filled with
-            # max_clients dead handlers rejects every new connection forever:
-            # the reject branch continues without ever reaping (the reap used
-            # to run only on the accept-success path).
-            self._reap_dead_clients()
-            with self._clients_lock:
-                # Re-check under the lock. _open_channel above performs the
-                # auth/TLS handshake, which can take up to _AUTH_TIMEOUT_S; a
-                # stop() during that window sets _shutdown and then snapshots
-                # and clears _clients under this same lock. Without this check
-                # the handler is registered *after* that snapshot, so nothing
-                # ever stops it — leaving a viewer dispatching input to a host
-                # the operator has already stopped.
-                if stop.is_set():
-                    autocontrol_logger.info(
-                        "remote_desktop dropping %s: host stopped during "
-                        "handshake", address,
-                    )
-                    handler._close()
-                    return
-                if len(self._clients) >= self._max_clients:
-                    autocontrol_logger.info(
-                        "remote_desktop dropping %s: max_clients reached",
-                        address,
-                    )
-                    handler._close()
-                    continue
-                self._clients.append(handler)
-            handler.start()
+            if self._register_client(handler, address, stop):
+                handler.start()
+        finally:
+            self._handshake_slots.release()
+
+    def _register_client(self, handler: "_ClientHandler", address,
+                         stop: threading.Event) -> bool:
+        """Add ``handler`` to the client table unless stopped or full."""
+        # Prune handlers whose viewer already disconnected *before* the
+        # capacity check below. Otherwise a client table filled with
+        # max_clients dead handlers rejects every new connection forever.
+        self._reap_dead_clients()
+        with self._clients_lock:
+            # Re-check under the lock: a stop() during the handshake sets
+            # _shutdown and then snapshots and clears _clients under this same
+            # lock, so a handler registered after that would never be stopped
+            # -- a viewer dispatching input to a host the operator stopped.
+            if stop.is_set():
+                autocontrol_logger.info(
+                    "remote_desktop dropping %s: host stopped during "
+                    "handshake", address,
+                )
+                handler._close()
+                return False
+            if len(self._clients) >= self._max_clients:
+                autocontrol_logger.info(
+                    "remote_desktop dropping %s: max_clients reached",
+                    address,
+                )
+                handler._close()
+                return False
+            self._clients.append(handler)
+        return True
 
     def _build_channel(self, sock: socket.socket,
                        address) -> MessageChannel:

je_auto_control/utils/remote_desktop/host_client.py

@@ -12,6 +12,7 @@
 import threading
 from typing import TYPE_CHECKING, Deque, Optional
 
+from je_auto_control.utils.exception.exceptions import AutoControlException
 from je_auto_control.utils.logging.logging_instance import autocontrol_logger
 from je_auto_control.utils.remote_desktop.auth import make_nonce
 from je_auto_control.utils.remote_desktop.clipboard_sync import (
@@ -81,10 +82,10 @@ def start(self) -> None:
             return
         self.authenticated = True
         # The initial cursor + frame are seeded from _send_loop (the
-        # per-client sender thread), not here. start() runs on the shared
-        # accept thread with the socket timeout already cleared, so sending a
-        # full-screen JPEG to a viewer that authenticates then stops reading
-        # would block every new accept until its send buffer drains.
+        # per-client sender thread), not here: start() runs on this
+        # connection's handshake thread with the socket timeout already
+        # cleared, and a viewer that authenticates then stops reading would
+        # pin that thread -- and its handshake slot -- until its buffer drains.
         self._sender_thread = threading.Thread(
             target=self._send_loop, name="rd-sender", daemon=True,
         )
@@ -267,7 +268,14 @@ def _recv_loop(self) -> None:
                     )
                 self.stop()
                 return
-            self._route_incoming(msg_type, payload)
+            try:
+                self._route_incoming(msg_type, payload)
+            except BaseException:
+                # Dying here without stop() left a handler that still
+                # counted as an authenticated client: it ignored all later
+                # input and held a slot that locked out new viewers.
+                self.stop()
+                raise
 
     def _route_incoming(self, msg_type: MessageType, payload: bytes) -> None:
         """Dispatch one received message to the matching handler."""
@@ -396,7 +404,10 @@ def _handle_input_payload(self, payload: bytes) -> None:
                 "remote_desktop rejected INPUT from %s: %r",
                 self._address, error,
             )
-        except (OSError, RuntimeError, ValueError, TypeError) as error:
+        # AutoControlException: the wrappers raise it for an unknown key or
+        # button name, which is the peer's input, not the host failing.
+        except (OSError, RuntimeError, ValueError, TypeError,
+                AutoControlException) as error:
             autocontrol_logger.warning(
                 "remote_desktop input apply failed for %s: %r",
                 self._address, error,

je_auto_control/utils/remote_desktop/input_dispatch.py

@@ -57,7 +57,13 @@ def dispatch_input(message: Mapping[str, Any]) -> Any:
     if action == "ping":
         return None
     wrappers = _import_wrappers()
-    return _APPLIERS[action](message, wrappers)
+    try:
+        return _APPLIERS[action](message, wrappers)
+    except (KeyError, TypeError, ValueError, OverflowError) as error:
+        # A missing field (KeyError), a non-numeric or infinite coordinate
+        # (ValueError / OverflowError from int()) is the peer's malformed
+        # message, and the receive loop only expects this type for that.
+        raise InputDispatchError(f"malformed {action!r} message: {error!r}") from error
 
 
 def _apply_mouse_move(message: Mapping[str, Any], wrappers: Dict[str, Any]) -> Any: