fix(device): isolate and bound receiver pre-authentication admission

docs/i18n-manifest.json

@@ -21,7 +21,7 @@
     "docs/download-metrics.md": "ded60e7b84d14ce779f8d639516b4e390358a06ef76b677dc9c1f997fb626a9f",
     "docs/mcp.md": "d8db1c66a5a2f500c8916d7a739616a5fe36af3a89ef8de631af4e427c8334de",
     "docs/multi-home-list.md": "f8d8e3adf727e4842ea7c4d45e8f6523b11d0513bf2509d4d1204645875e6489",
-    "docs/paired-devices.md": "10c11a0fe37602de896f780297196edaf37a82fe34c5d2250172404da245a636",
+    "docs/paired-devices.md": "47d58364cbd2e83e496d7319e0887d4ea7bdedb2a29ca20a06e842b3de73266a",
     "docs/relay-auth.md": "959945f5fe1fe8a5d402fb9a289620ada08cf063ce9709698db11476d4fb3e4c",
     "docs/relay-development-plan.md": "0abf750b36275ea4e467a0020e560277f08689f2d574aed7674133705383b1ff",
     "docs/relay-edge-policy.md": "c1c7f7a1c4303943c78ca334dab4034f66c0c272e40c007d07c3da2a496155eb",

docs/ja/paired-devices.md

@@ -2,6 +2,8 @@
 
 このオプションの Unix/Python 3.11+ トランスポートは、オペレーターが定義した Claude、Codex、または登録済みの Antigravity エンドポイントへリクエストを転送します。通常のローカル/SSH コマンドは依存関係のない状態を維持します。これは明示的な CLI ワークフローであり、モバイルアプリケーション、NAT トラバーサル、WireGuard トンネル、自動公開サービスなどはインストールされません。
 
+受信側はデバイス認証前の直接接続を最大 16 本、Relay に attach された接続を最大 2 本まで許可し、認証済み接続 8 本とは別に管理します。TLS はネットワークとスケジューリングの遅延を含めて 2 秒以内に完了する必要があります。TLS 後、クライアント証明書のない初期ペアリング接続には、単一のペアリング要求と応答を完了するためにさらに 2 秒が与えられます。認証済み要求の期限は 45 秒のままです。この短い期限は正常な遅い接続も拒否する場合があります。アイドル状態または TLS が遅い直接ソケット 8 本があっても、ペアリング済みの直接接続と Relay 要求を受け付ける余地が残ります。ただし、認証前の直接接続枠 16 本がすべて埋まると、ペアリング済みの直接接続も拒否される可能性があり、予約された Relay 枠 2 本も飽和する場合があります。任意のフラッド攻撃に対する可用性は保証されません。超過ソケットや受信側のシャットダウン開始後のソケットには処理タスクを作成しません。集計した入場カウンターは ID を含めずに拒否とクリーンアップ失敗を保持します。明示的に有効化した `receiver_admission` イベントは、固定カウンターと許可リスト内の理由を 1 分に最大 1 回報告します。アプリケーションの自動再試行は有効になりません。
+
 ## 接続失敗の診断と安全な接続再試行
 
 `no_authenticated_route` は `retryAllowed:false` と `consumptionConfirmed:false` を維持します。`routeFailures` は経路ごとの最後の `stage`、許可リスト内の `reason`、`attempts` を示します。限定された `attemptHistory` は各試行の段階・理由・`elapsedMs` を保持し、HTTP 拒否には `httpStatus`、WebSocket の切断には `closeCode` が含まれる場合があります。制御、入場、アップグレード、attach、ピア TLS、probe の失敗を区別し、URL・認証情報・生の例外・メッセージ本文は記録しません。2 回目で成功した場合は `setupDegraded:true`、`setupAttempts:2`、`setupFailureHistory` が付き、正常な安定性試験の合格とは見なしません。

docs/ko/paired-devices.md

@@ -2,6 +2,8 @@
 
 이 선택형 Unix/Python 3.11+ 트랜스포트는 운영자가 정의한 Claude, Codex 또는 등록된 Antigravity 엔드포인트로 요청을 전달합니다. 일반적인 로컬/SSH 명령은 의존성 없는 상태로 유지됩니다. 이는 명시적인 CLI 워크플로이며, 모바일 애플리케이션, NAT 통과, WireGuard 터널 또는 자동 공용 서비스는 설치되지 않습니다.
 
+수신자는 기기 인증 전 직접 연결을 최대 16개, Relay에 attach된 연결을 최대 2개 허용하며, 인증된 연결 8개와 별도로 관리합니다. TLS는 네트워크 및 스케줄링 지연을 포함해 2초 안에 완료해야 합니다. TLS 이후 클라이언트 인증서가 없는 초기 페어링 연결에는 단일 페어링 요청과 응답을 완료할 추가 2초가 주어지며, 인증된 요청의 제한 시간은 45초로 유지됩니다. 이 짧은 제한 시간은 정상적인 느린 연결도 거부할 수 있습니다. 유휴 또는 느린 TLS 직접 소켓 8개가 있어도 페어링된 직접 연결과 Relay 요청을 받을 여유가 남습니다. 그러나 직접 연결의 인증 전 슬롯 16개가 모두 차면 페어링된 직접 연결도 거부될 수 있고, 예약된 Relay 슬롯 2개도 포화될 수 있으므로 임의의 홍수 공격에서 가용성을 보장하지 않습니다. 초과 소켓과 수신자 종료가 시작된 뒤의 소켓에는 처리 태스크를 만들지 않습니다. 집계 입장 카운터는 신원 없이 거부 및 정리 실패를 보존하며, 명시적으로 켠 `receiver_admission` 이벤트는 고정 카운터와 허용 목록의 사유를 분당 최대 한 번 보고합니다. 애플리케이션 자동 재시도는 활성화되지 않습니다.
+
 ## 연결 실패 진단과 안전한 연결 재시도
 
 `no_authenticated_route`는 `retryAllowed:false`, `consumptionConfirmed:false`를 유지합니다. `routeFailures`는 경로별 마지막 `stage`, 허용 목록의 `reason`, `attempts`를 표시합니다. 제한된 `attemptHistory`는 각 시도의 단계·사유·`elapsedMs`를 보존하며 HTTP 거부에는 `httpStatus`, WebSocket 종료에는 `closeCode`가 포함될 수 있습니다. 제어, 입장, 업그레이드, attach, 피어 TLS 및 probe 실패를 구분하되 URL·자격증명·원문 예외·메시지 본문은 기록하지 않습니다. 두 번째 연결이 성공하면 `setupDegraded:true`, `setupAttempts:2`, `setupFailureHistory`가 표시되며 깨끗한 안정성 통과로 계산하지 않습니다.

docs/paired-devices.md

@@ -5,6 +5,21 @@ Claude, Codex or registered Antigravity endpoints. The usual local/SSH commands
 stay dependency-free. This is an explicit CLI workflow; no mobile application,
 NAT traversal, WireGuard tunnel or automatic public service is installed.
 
+The receiver admits up to 16 direct and two Relay-attached connections before
+device authentication, separately from its eight authenticated connections.
+TLS must finish within two seconds, including network and scheduling delays.
+After TLS, a cert-less bootstrap connection has another two seconds to complete
+its single pairing request and response; authenticated request deadlines remain
+45 seconds. These short deadlines may refuse a legitimate slow connection.
+Eight idle or slow-TLS direct sockets leave room for paired direct and Relay
+requests. Saturating all 16 direct pre-authentication slots can still refuse a
+paired direct connection, and both reserved Relay slots can also saturate;
+availability under arbitrary floods is not guaranteed. Admission creates no
+handler task for an excess socket or once receiver shutdown starts. Aggregate
+admission counters retain refusals and cleanup failures without identities;
+opt-in `receiver_admission` events report fixed counters and allowlisted reasons
+at most once per minute. No automatic application retry is enabled.
+
 ## Connection failures and safe setup retry
 
 `no_authenticated_route` retains `retryAllowed:false` and

docs/zh-CN/paired-devices.md

@@ -5,6 +5,8 @@ Claude、Codex 或已注册的 Antigravity 端点。通常的本地/SSH 命令
 保持无外部依赖。这是一个显式的 CLI 工作流；不会安装移动端应用程序、
 NAT 穿透、WireGuard 隧道或自动公开服务。
 
+接收端在设备认证前最多允许 16 条直接连接和 2 条已 attach 的 Relay 连接，与 8 条已认证连接分别管理。TLS 必须在两秒内完成，包含网络和调度延迟。TLS 完成后，没有客户端证书的初始配对连接还有两秒用于完成唯一一次配对请求及响应；已认证请求的期限仍为 45 秒。这些较短的期限也可能拒绝合法的慢速连接。存在 8 个空闲或慢速 TLS 直接套接字时，仍有空间接收已配对设备的直接连接和 Relay 请求。但全部 16 个直接连接认证前名额耗尽时，仍可能拒绝已配对设备的直接连接，两个预留的 Relay 名额也可能饱和；不保证在任意洪水攻击下的可用性。超额套接字或接收端开始关闭后的套接字不会创建处理任务。汇总准入计数器保留拒绝和清理失败数量，不包含身份；显式启用的 `receiver_admission` 事件每分钟最多报告一次固定计数器和允许列表内的原因。不会启用应用程序自动重试。
+
 ## 连接失败诊断与安全的连接重试
 
 `no_authenticated_route` 保留 `retryAllowed:false` 和 `consumptionConfirmed:false`。`routeFailures` 给出每条失败路径最后一次的 `stage`、允许列表中的 `reason` 和 `attempts`。有界的 `attemptHistory` 保留每次尝试的阶段、原因和 `elapsedMs`；HTTP 拒绝还可能包含 `httpStatus`，WebSocket 关闭可能包含 `closeCode`。控制、准入、升级、attach、对端 TLS 和 probe 故障可以区分，但不会记录 URL、凭据、原始异常或消息正文。第二次尝试成功时会有 `setupDegraded:true`、`setupAttempts:2` 和 `setupFailureHistory`，不能算作干净的稳定性测试通过。

session_peer_relay/app.py

@@ -18,13 +18,30 @@
 from .transport_errors import failure, NoAuthenticatedRoute, TransportFailure
 
 
+# Pre-authentication admission is bounded separately from paired connections.
+# A reserved relay lane prevents direct-port traffic from consuming its budget.
+DIRECT_PREAUTH_LIMIT = 16
+RELAY_PREAUTH_LIMIT = 2
+AUTHENTICATED_LIMIT = 8
+PREAUTH_TLS_TIMEOUT = 2
+CERTLESS_REQUEST_TIMEOUT = 2
+
+
 class Receiver:
     def __init__(self, store, policy, *, diagnostic_events=False):
         self.store = store
         self.policy = Policy(policy)
         self.native = Native()
         self.tasks = set()
         self.connections = set()
+        self.direct_preauth = set()
+        self.relay_preauth = set()
+        self.authenticated = set()
+        self.closing = False
+        self.admission_counters = dict.fromkeys(('directPreauthRefused', 'relayPreauthRefused',
+                                                'authenticatedRefused', 'closingRefused',
+                                                'abortFailed', 'handlerFailed'), 0)
+        self.admission_last_logged = float('-inf')
         self.rejected = 0
         self.relay_failure_last_logged = float('-inf')
         self.diagnostic_events = diagnostic_events
@@ -35,6 +52,24 @@ def lifecycle_event(self, event, **values):
                 json.dumps({'event': event, 'eventTimeUtcMs': int(time.time()*1000),
                             **values}, sort_keys=True))
 
+    def admission_event(self, reason):
+        # Fixed counters retain aggregate evidence; flood diagnostics emit at
+        # most one allowlisted event per minute, without connection metadata.
+        if self.diagnostic_events and time.monotonic()-self.admission_last_logged >= 60:
+            self.admission_last_logged = time.monotonic()
+            with contextlib.suppress(Exception):
+                self.lifecycle_event('receiver_admission', reason=reason,
+                                     counters=dict(self.admission_counters))
+
+    def abort(self, raw):
+        try:
+            raw.abort()
+        except Exception:
+            self.admission_counters['abortFailed'] += 1
+            self.admission_event('abort_failed')
+        finally:
+            self.release(raw)
+
     async def watch_peer(self, channel):
         while True:
             await asyncio.sleep(.2)
@@ -128,17 +163,14 @@ async def dispatch(self, peer, value):
         return self.store.finish(peer, value['id'], result)
 
     async def handle(self, raw):
-        if len(self.connections) >= 8:
-            await raw.close()
-            return
-        self.connections.add(raw)
         channel = None
         watch = None
         try:
             channel = Secure(raw, context(self.store.identity_root, True, self.store.trusted()), server=True)
             started = time.monotonic()
             try:
-                await channel.handshake()
+                async with asyncio.timeout(PREAUTH_TLS_TIMEOUT):
+                    await channel.handshake()
             except Exception as exc:
                 if isinstance(raw, Ws):
                     self.lifecycle_event('peer_tls', outcome='failed',
@@ -151,39 +183,88 @@ async def handle(self, raw):
                     elapsedMs=round((time.monotonic()-started)*1000),
                     attemptId=getattr(raw, 'attempt_id', None))
             if channel.peer:
+                if len(self.authenticated) >= AUTHENTICATED_LIMIT:
+                    self.admission_counters['authenticatedRefused'] += 1
+                    self.admission_event('authenticated_full')
+                    return
+                self.authenticated.add(raw)
+                self.direct_preauth.discard(raw)
+                self.relay_preauth.discard(raw)
                 watch = asyncio.create_task(self.watch_peer(channel))
             # Bound unauthenticated pairing attempts per TLS connection.
             budget = 100 if channel.peer else 1
-            for _ in range(budget):
-                request = await channel.recv()
-                try:
-                    result = await self.dispatch(channel.peer, request)
-                except Rejected as exc:
-                    self.rejected += 1
-                    result = {'ok': False, 'status': 'refused', 'reason': str(exc),
-                              'consumptionConfirmed': False, 'retryAllowed': False}
-                except Exception:
-                    self.rejected += 1
-                    result = {'ok': False, 'status': 'unknown', 'reason': 'unknown',
-                              'retryAllowed': False, 'consumptionConfirmed': False}
-                if len(json.dumps(result, ensure_ascii=False).encode()) > 60*1024:
-                    result = {'ok': False, 'reason': 'result_too_large', 'retryAllowed': False,
-                              'consumptionConfirmed': False}
-                await channel.send(result)
+            async with asyncio.timeout(None if channel.peer else CERTLESS_REQUEST_TIMEOUT):
+                for _ in range(budget):
+                    request = await channel.recv()
+                    try:
+                        result = await self.dispatch(channel.peer, request)
+                    except Rejected as exc:
+                        self.rejected += 1
+                        result = {'ok': False, 'status': 'refused', 'reason': str(exc),
+                                  'consumptionConfirmed': False, 'retryAllowed': False}
+                    except Exception:
+                        self.rejected += 1
+                        result = {'ok': False, 'status': 'unknown', 'reason': 'unknown',
+                                  'retryAllowed': False, 'consumptionConfirmed': False}
+                    if len(json.dumps(result, ensure_ascii=False).encode()) > 60*1024:
+                        result = {'ok': False, 'reason': 'result_too_large', 'retryAllowed': False,
+                                  'consumptionConfirmed': False}
+                    await channel.send(result)
         except Exception:
             self.rejected += 1
         finally:
-            if watch:
-                watch.cancel()
-                with contextlib.suppress(asyncio.CancelledError):
-                    await watch
-            await (channel.close() if channel is not None else raw.close())
-            self.connections.discard(raw)
+            try:
+                if watch:
+                    watch.cancel()
+                    with contextlib.suppress(asyncio.CancelledError):
+                        await watch
+            finally:
+                try:
+                    await (channel.close() if channel is not None else raw.close())
+                finally:
+                    self.abort(raw)
+
+    def release(self, raw):
+        self.connections.discard(raw)
+        self.direct_preauth.discard(raw)
+        self.relay_preauth.discard(raw)
+        self.authenticated.discard(raw)
 
     def spawn(self, raw):
+        if self.closing:
+            self.admission_counters['closingRefused'] += 1
+            self.admission_event('closing')
+            self.abort(raw)
+            return
+        preauth, limit = ((self.relay_preauth, RELAY_PREAUTH_LIMIT) if isinstance(raw, Ws)
+                          else (self.direct_preauth, DIRECT_PREAUTH_LIMIT))
+        # Reserve synchronously, before scheduling: a burst cannot create an
+        # unbounded queue of handling or rejection/close tasks.
+        if len(preauth) >= limit:
+            counter, reason = (('relayPreauthRefused', 'relay_preauth_full') if isinstance(raw, Ws)
+                               else ('directPreauthRefused', 'direct_preauth_full'))
+            self.admission_counters[counter] += 1
+            self.admission_event(reason)
+            self.abort(raw)
+            return
+        preauth.add(raw)
+        self.connections.add(raw)
         task = asyncio.create_task(self.handle(raw))
         self.tasks.add(task)
-        task.add_done_callback(self.tasks.discard)
+
+        def completed(task):
+            # Cancellation before handle starts never executes its finally.
+            try:
+                if raw in self.connections:
+                    self.abort(raw)
+                if not task.cancelled() and task.exception() is not None:
+                    self.admission_counters['handlerFailed'] += 1
+                    self.admission_event('handler_failed')
+            finally:
+                self.release(raw)
+                self.tasks.discard(task)
+
+        task.add_done_callback(completed)
 
     async def listen(self, host, port):
         return await asyncio.start_server(lambda r, w: self.spawn(Tcp(r, w)), host, port, limit=65536)
@@ -240,6 +321,7 @@ def observed(event, *, elapsed_ms, attempt_id=None):
                 delay = .5 if expired else min(5, delay*2)
 
     async def close(self):
+        self.closing = True
         tasks = list(self.tasks)
         for task in tasks:
             task.cancel()

session_peer_relay/wire.py

@@ -38,6 +38,13 @@ async def close(self):
         except (OSError, asyncio.TimeoutError):
             pass
 
+    def abort(self):
+        try:
+            self.writer.transport.abort()
+        except Exception:
+            self.writer.close()
+            raise
+
 
 class Ws:
     def __init__(self, ws, attempt_id=None):
@@ -57,6 +64,13 @@ async def send(self, data):
     async def close(self):
         await self.ws.close()
 
+    def abort(self):
+        try:
+            self.ws.transport.abort()
+        except Exception:
+            self.ws.transport.close()
+            raise
+
 
 class PinnedConnect(connect):
     def process_redirect(self, exc):