feat(web): move connection downlinks to WebSocket
This commit is contained in:
imccyu
@@ -2,5 +2,5 @@
|
||||
# side as of the last confirmed-consistent state. Both languages carry equal authority;
|
||||
# after editing either side, bring the other along and re-record with:
|
||||
# pnpm run verify-translation-pairing --write packages/client/connection/README.md
|
||||
README.md: f537fee3273e3b5d2411197cf1a1a6e0d34af5f9
|
||||
README.zh.md: a29d2c00e7df3f6290a03ffdad59b70b43702aca
|
||||
README.md: 11bfc950b7d4f09f1a3075e0f444966de841be70
|
||||
README.zh.md: 4b346de9e0dbc468d6b94546aa963a5b57b62127
|
||||
|
||||
@@ -2,11 +2,15 @@
|
||||
|
||||
English | [中文](README.zh.md)
|
||||
|
||||
Wire consumer layer: the client plugin's apply mounts `ctx.connection` (shared api client + current-page loopback state + single-consumer stream-loop starter); the export face carries the wire contract types, the `AbstractApiClient` seam, and the loop's sink/config types. Loopback hostname classification stays package-internal: the `/api` Host fence uses it directly, while other client plugins consume the derived `ctx.connection.isLoopback` state. The node half's `/api` route pins the privileged method set (`host.pickDirectory`, `host.openPath`, and the whole configuration plane — `settings.describe`/`update`/`replace`/`mutate` and `credentials.describe`/`set`/`unset`, reads included, since describing returns the exposed configuration and probing an arbitrary reference reports where a credential comes from) to loopback by passing the trust fence with an empty trust list — a declared `trustedHosts` authority reaches every other method, while these stay loopback-local until a real authentication layer exists. The platform subclasses (WebApiClient/FixtureApiClient), the ConnectionController loop, and the fixture data source are package-internal — apply selects and drives them; tests reach them via src. Contract: api-contracts v3 §3.
|
||||
Wire consumer layer: the client plugin's apply mounts `ctx.connection` (shared api client + current-page loopback state + single-consumer stream-loop starter); the export face carries the wire contract types, the `AbstractApiClient` seam, and the loop's sink/config types. The real browser carrier uses HTTP POST for unary and respond operations and opens one downlink-only WebSocket each for `events.mux` and `events.host`; the fixture and in-process carriers continue to satisfy the same two-stream abstraction. Loopback hostname classification stays package-internal: the `/api` Host fence and WebSocket upgrades use it directly, while other client plugins consume the derived `ctx.connection.isLoopback` state. The node half's `/api` route pins the privileged method set (`host.pickDirectory`, `host.openPath`, and the whole configuration plane — `settings.describe`/`update`/`replace`/`mutate` and `credentials.describe`/`set`/`unset`, reads included, since describing returns the exposed configuration and probing an arbitrary reference reports where a credential comes from) to loopback by passing the trust fence with an empty trust list — a declared `trustedHosts` authority reaches every other method, while these stay loopback-local until a real authentication layer exists. The platform subclasses (WebApiClient/FixtureApiClient), the ConnectionController loop, and the fixture data source are package-internal — apply selects and drives them; tests reach them via src. The downlink boundary is documented in the [WebSocket downlink carrier Agent Note](../../../.agents/notes/implemented/architecture/2026-08-04-websocket-downlink-carrier.md); the protocol contract is api-contracts v3 §3.
|
||||
|
||||
## /api browser-trust fence
|
||||
|
||||
The node half guards every request under `/api` before bridging (`src/api-request-trust.ts`). Every request — browser-marked or not — must present a `Host` that is a loopback authority or matches a `trustedHosts` entry: exact on `host:port` entries, any port on port-less entries, both sides compared through WHATWG normalization (DNS-rebinding defense). There is deliberately no shortcut for requests without browser markers: over plain HTTP a browser attaches neither `Origin` nor Fetch-Metadata to reads (EventSource, images, navigations — those headers go only to trustworthy destinations), so an unmarked request may still be a rebound browser read with a readable response, and Host is the one header rebinding cannot forge; non-browser clients pass the same fence via loopback, the CLI-derived LAN IP literals, or a declared authority. When markers are present, an attached `Origin` must equal the Host authority, and an explicit `sec-fetch-site: cross-site` marker is refused. A `trustedHosts` entry that is not a bare, canonical `host[:port]` authority — one WHATWG parsing reads back exactly as written — fails the plugin load loudly: parsing would otherwise quietly authorize the hostname inside `harness.internal/path`, or broaden a dangling-colon or zero-padded port to an any-port grant. Failures answer plain 403 before any RPC dispatch. A non-loopback (`--host 0.0.0.0`) deployment therefore needs its serving authorities trusted: the dsh CLI derives the machine's LAN IP literals itself and its `--trusted-host` flag declares named ones, so `trustedHosts` in cordis.yml is for compositions the CLI does not boot. The fence is deliberately not an authentication layer — reachability policy stays with the webserver binding, and auth remains deferred work. Decision record: [the api browser-trust boundary Agent Note](../../../.agents/notes/implemented/architecture/2026-07-28-api-browser-trust-boundary.md).
|
||||
The node half guards every entry under `/api` before bridging or upgrading (`src/api-request-trust.ts`). Every request — browser-marked or not — must present a `Host` that is a loopback authority or matches a `trustedHosts` entry: exact on `host:port` entries, any port on port-less entries, both sides compared through WHATWG normalization (DNS-rebinding defense). There is deliberately no shortcut for unmarked HTTP requests: over plain HTTP a browser attaches neither `Origin` nor Fetch-Metadata to image and navigation reads, so an unmarked request may still be a rebound browser read with a readable response, and Host is the one header rebinding cannot forge; a browser WebSocket handshake carries `Origin` and passes the same comparison. Non-browser clients pass the same fence via loopback, the CLI-derived LAN IP literals, or a declared authority. When markers are present, an attached `Origin` must equal the Host authority, and an explicit `sec-fetch-site: cross-site` marker is refused. A `trustedHosts` entry that is not a bare, canonical `host[:port]` authority — one WHATWG parsing reads back exactly as written — fails the plugin load loudly: parsing would otherwise quietly authorize the hostname inside `harness.internal/path`, or broaden a dangling-colon or zero-padded port to an any-port grant. HTTP failures answer plain 403 before any RPC dispatch; upgrade failures reject the handshake before any event stream starts. A non-loopback (`--host 0.0.0.0`) deployment therefore needs its serving authorities trusted: the dsh CLI derives the machine's LAN IP literals itself and its `--trusted-host` flag declares named ones, so `trustedHosts` in cordis.yml is for compositions the CLI does not boot. The fence is deliberately not an authentication layer — reachability policy stays with the webserver binding, and auth remains deferred work. Decision record: [the api browser-trust boundary Agent Note](../../../.agents/notes/implemented/architecture/2026-07-28-api-browser-trust-boundary.md).
|
||||
|
||||
## `/api` WebSocket downlinks
|
||||
|
||||
`/api/events.mux` and `/api/events.host` each accept a WebSocket upgrade and send only the corresponding `ServerRequest` text messages to the browser; the client sends no application data over these sockets. If either socket ends, the current connection generation fails and rebuilds both streams; readiness still requires both sockets to be open and the `host.describe` HTTP call to succeed. Ordinary network GETs to these paths return 426 with no SSE fallback; `toFetchHandler`'s SSE codec serves only the isomorphic in-process carrier.
|
||||
|
||||
## Keyless fixture
|
||||
|
||||
|
||||
@@ -2,11 +2,15 @@
|
||||
|
||||
[English](README.md) | 中文
|
||||
|
||||
协议消费层：客户端插件的 apply 会挂载 `ctx.connection`（共享 API 客户端 + 当前页面的 loopback 状态 + 单消费方流循环启动器）；导出表层携带协议契约类型、`AbstractApiClient` seam，以及循环的 sink／配置类型。Loopback hostname 判定逻辑留在包内部：`/api` Host fence 会直接使用它，其他客户端插件则消费派生的 `ctx.connection.isLoopback` 状态。node 半侧的 `/api` 路由让特权方法集（`host.pickDirectory`、`host.openPath`，以及整个配置面——`settings.describe`/`update`/`replace`/`mutate` 与 `credentials.describe`/`set`/`unset`，读取也在内，因为 describe 会返回已暴露的配置，而探测任意引用会报出某条凭据来自何处）以空信任表过信任 fence，从而钉在回环——已声明的 `trustedHosts` 授权可达其余全部方法，而这些方法在真正的认证层出现之前仍只限回环本机。平台子类（WebApiClient/FixtureApiClient）、ConnectionController 循环和 fixture 数据源都属于包内部：apply 负责选择并驱动它们，测试则通过 src 访问。契约：api-contracts v3 §3。
|
||||
协议消费层：客户端插件的 apply 会挂载 `ctx.connection`（共享 API 客户端 + 当前页面的 loopback 状态 + 单消费方流循环启动器）；导出表层携带协议契约类型、`AbstractApiClient` seam，以及循环的 sink／配置类型。真实浏览器载体以 HTTP POST 发送 unary／respond，并为 `events.mux` 与 `events.host` 各开一条只下行的 WebSocket；fixture 与进程内载体继续满足同一双流抽象。Loopback hostname 判定逻辑留在包内部：`/api` Host fence 与 WebSocket upgrade 会直接使用它，其他客户端插件则消费派生的 `ctx.connection.isLoopback` 状态。node 半侧的 `/api` 路由让特权方法集（`host.pickDirectory`、`host.openPath`，以及整个配置面——`settings.describe`/`update`/`replace`/`mutate` 与 `credentials.describe`/`set`/`unset`，读取也在内，因为 describe 会返回已暴露的配置，而探测任意引用会报出某条凭据来自何处）以空信任表过信任 fence，从而钉在回环——已声明的 `trustedHosts` 授权可达其余全部方法，而这些方法在真正的认证层出现之前仍只限回环本机。平台子类（WebApiClient/FixtureApiClient）、ConnectionController 循环和 fixture 数据源都属于包内部：apply 负责选择并驱动它们，测试则通过 src 访问。下行边界见 [WebSocket 下行载体 Agent Note](../../../.agents/notes/implemented/architecture/2026-08-04-websocket-downlink-carrier.md)；协议契约见 api-contracts v3 §3。
|
||||
|
||||
## /api 浏览器信任栅栏
|
||||
|
||||
node 半侧在桥接前守卫 `/api` 下的每个请求（`src/api-request-trust.ts`）。每个请求——无论是否带浏览器标记——`Host` 都必须是回环地址权威，或与某个 `trustedHosts` 条目匹配：带端口的 `host:port` 条目精确匹配，不带端口的条目匹配任意端口，两侧均经 WHATWG 归一化后比较（DNS rebinding 防御）。刻意不为无浏览器标记的请求开捷径：明文 HTTP 下浏览器的读取（EventSource、图片、导航——这些头只发给可信目标）既不带 `Origin` 也不带 Fetch-Metadata，因此无标记请求仍可能是被重绑页面发起的、响应可被读走的读取，而 Host 是重绑唯一伪造不了的请求头；非浏览器客户端经由回环地址、CLI 推导的 LAN IP 字面量或已声明的权威通过同一道栅栏。当标记存在时，`Origin` 必须与 Host 权威完全一致；显式的 `sec-fetch-site: cross-site` 标记一律拒绝。不是纯的、规范形 `host[:port]` 权威的 `trustedHosts` 条目——即 WHATWG 解析读回后与原文不完全一致的——会让插件加载大声失败：否则解析会悄悄授权 `harness.internal/path` 这类笔误里的 hostname，或把悬空冒号、补零端口放大成任意端口授权。失败在任何 RPC 分发之前以纯 403 应答。因此非回环（`--host 0.0.0.0`）部署需要让自己的服务权威被信任：dsh CLI 会自行推导本机的 LAN IP 字面量，其 `--trusted-host` flag 用于声明具名权威，所以 cordis.yml 中的 `trustedHosts` 面向 CLI 不参与引导的组合。这道栅栏刻意不承担认证职责——可达性策略归 webserver 绑定配置，认证仍是延期工作。决策记录：[api 浏览器信任边界 Agent Note](../../../.agents/notes/implemented/architecture/2026-07-28-api-browser-trust-boundary.md)。
|
||||
node 半侧在桥接或 upgrade 前守卫 `/api` 下的每个入口（`src/api-request-trust.ts`）。每个请求——无论是否带浏览器标记——`Host` 都必须是回环地址权威，或与某个 `trustedHosts` 条目匹配：带端口的 `host:port` 条目精确匹配，不带端口的条目匹配任意端口，两侧均经 WHATWG 归一化后比较（DNS rebinding 防御）。刻意不为无浏览器标记的 HTTP 请求开捷径：明文 HTTP 下浏览器的图片与导航读取既不带 `Origin` 也不带 Fetch-Metadata，因此无标记请求仍可能是被重绑页面发起的、响应可被读走的读取，而 Host 是重绑唯一伪造不了的请求头；WebSocket 浏览器握手会带 `Origin` 并通过同一道比较。非浏览器客户端经由回环地址、CLI 推导的 LAN IP 字面量或已声明的权威通过同一道栅栏。当标记存在时，`Origin` 必须与 Host 权威完全一致；显式的 `sec-fetch-site: cross-site` 标记一律拒绝。不是纯的、规范形 `host[:port]` 权威的 `trustedHosts` 条目——即 WHATWG 解析读回后与原文不完全一致的——会让插件加载大声失败：否则解析会悄悄授权 `harness.internal/path` 这类笔误里的 hostname，或把悬空冒号、补零端口放大成任意端口授权。HTTP 失败在任何 RPC 分发之前以纯 403 应答，upgrade 失败在启动任何 event stream 前拒绝握手。因此非回环（`--host 0.0.0.0`）部署需要让自己的服务权威被信任：dsh CLI 会自行推导本机的 LAN IP 字面量，其 `--trusted-host` flag 用于声明具名权威，所以 cordis.yml 中的 `trustedHosts` 面向 CLI 不参与引导的组合。这道栅栏刻意不承担认证职责——可达性策略归 webserver 绑定配置，认证仍是延期工作。决策记录：[api 浏览器信任边界 Agent Note](../../../.agents/notes/implemented/architecture/2026-07-28-api-browser-trust-boundary.md)。
|
||||
|
||||
## `/api` WebSocket 下行
|
||||
|
||||
`/api/events.mux` 与 `/api/events.host` 各接受一条 WebSocket upgrade，并只向浏览器发送对应的 `ServerRequest` text message；客户端不会在这些 socket 上发送业务数据。任一 socket 结束都会使当前 connection generation 失败并重建两条流，连接就绪仍要求两条 socket open 且 `host.describe` HTTP 调用成功。普通网络 GET 这些路径会返回 426，不保留 SSE 回退；`toFetchHandler` 的 SSE 编解码只服务进程内同构载体。
|
||||
|
||||
## 无密钥 fixture
|
||||
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@deepseek-ai/dsh-client-connection",
|
||||
"description": "Wire consumer layer: IApiClient subclasses, ConnectionController (SSE dual-stream + reconnect), fixture api (no cordis)",
|
||||
"description": "Wire consumer layer: HTTP-up/WebSocket-down client, ConnectionController dual streams with reconnect, and fixture api",
|
||||
"version": "0.0.1",
|
||||
"private": true,
|
||||
"type": "module",
|
||||
@@ -34,7 +34,8 @@
|
||||
"@deepseek-ai/dsh-llm": "workspace:^",
|
||||
"@deepseek-ai/dsh-session": "workspace:^",
|
||||
"@deepseek-ai/dsh-tools": "workspace:^",
|
||||
"schemastery": "^3.18.0"
|
||||
"schemastery": "^3.18.0",
|
||||
"ws": "^8.21.0"
|
||||
},
|
||||
"files": [
|
||||
"lib/index.js",
|
||||
@@ -52,6 +53,7 @@
|
||||
"devDependencies": {
|
||||
"@deepseek-ai/dsh-host-webserver": "workspace:^",
|
||||
"@deepseek-ai/dsh-invariants": "workspace:^",
|
||||
"@types/ws": "^8.18.1",
|
||||
"cordis": "^4.0.0-rc.7"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,8 +1,14 @@
|
||||
/**
|
||||
* The /api URL prefix — single source for both halves of the web transport.
|
||||
* The node half registers this prefix on the web server; browser-side path
|
||||
* literals currently live in the apiproxy client layer (out of scope here).
|
||||
* The node half registers this prefix on the web server; both halves share the
|
||||
* event paths below for the browser WebSocket downlinks.
|
||||
*/
|
||||
|
||||
/** Route prefix owning every api request (`/api` and `/api/<anything>`). */
|
||||
export const API_PATH = '/api'
|
||||
|
||||
/** Browser mux-frame WebSocket pathname. */
|
||||
export const MUX_EVENTS_PATH = `${API_PATH}/events.mux`
|
||||
|
||||
/** Browser host-frame WebSocket pathname. */
|
||||
export const HOST_EVENTS_PATH = `${API_PATH}/events.host`
|
||||
|
||||
@@ -4,7 +4,7 @@
|
||||
* the attacker's domain while the socket reaches this server) and cross-site
|
||||
* requests fired from a malicious page. The Host fence binds every request,
|
||||
* browser-looking or not: over plain HTTP a browser attaches neither Origin
|
||||
* nor Fetch-Metadata to reads (EventSource, images, navigations — those
|
||||
* nor Fetch-Metadata to reads (images and navigations — those
|
||||
* headers go only to trustworthy destinations), so an unmarked request may
|
||||
* still be a rebound browser read and Host is the one header rebinding cannot
|
||||
* forge. Non-browser and remote clients pass the same fence via loopback, the
|
||||
@@ -97,7 +97,7 @@ export function isTrustedApiRequest(request: ApiTrustRequest, trustedHosts: read
|
||||
// fills Host from the URL it believes it is talking to, so a rebound page
|
||||
// carries the attacker's domain here even though the socket lands on this
|
||||
// server. There is no marker shortcut — a browser read over plain HTTP
|
||||
// (EventSource, images, navigations) arrives with neither Origin nor
|
||||
// (images and navigations) arrives with neither Origin nor
|
||||
// Fetch-Metadata, indistinguishable from curl, and its response is readable
|
||||
// by the rebound page.
|
||||
const host = header(request.headers, 'host')
|
||||
|
||||
@@ -126,7 +126,7 @@ export class ConnectionController {
|
||||
|
||||
try {
|
||||
// Strict readiness handshake (audit C2): describe proves unary reachability, onOpen
|
||||
// proves each SSE transport is established (response headers in, before any frame) —
|
||||
// proves each physical stream is established before any frame —
|
||||
// only then may onConnected fire, so the resync it triggers cannot outrun the
|
||||
// subscribed baseline. The timeout guards against a carrier that never fires onOpen
|
||||
// (see ConnectionConfig.streamOpenTimeoutMs).
|
||||
|
||||
@@ -1,12 +1,91 @@
|
||||
// WebApiClient: the browser platform subclass — transport = global fetch over same-origin
|
||||
// /api/* (base resolution handled by AbstractApiClient). Envelope observation comes from the
|
||||
// base batching aspect; subscribers attach via subscribeEnvelopes (see boot).
|
||||
/** Browser API carrier: HTTP upstream plus one WebSocket per downstream event stream. */
|
||||
|
||||
import type { ApiProxy, HostFrame, MuxFrame, RpcRequest, ServerRequest } from './api.ts'
|
||||
import { AbstractApiClient } from './api.ts'
|
||||
import { hostFrameSchema, muxFrameSchema } from '@deepseek-ai/dsh-host-apiproxy/api/events.schema'
|
||||
import { serverRequestSchema } from '@deepseek-ai/dsh-host-apiproxy/api/rpc.schema'
|
||||
import { HOST_EVENTS_PATH, MUX_EVENTS_PATH } from '../api-path.ts'
|
||||
|
||||
/** Browser platform subclass: transport = global fetch over same-origin /api/*. */
|
||||
type SocketItem<F> = { kind: 'frame'; envelope: RpcRequest<F> } | { kind: 'end' }
|
||||
type Parser<F> = { parse(value: unknown): F }
|
||||
|
||||
/** Browser platform subclass: unary/respond use fetch; mux/host use downlink-only WebSockets. */
|
||||
export class WebApiClient extends AbstractApiClient {
|
||||
protected doFetch(input: URL, init?: RequestInit): Promise<Response> {
|
||||
return globalThis.fetch(input, init)
|
||||
}
|
||||
|
||||
protected override openMux(
|
||||
_payload: Parameters<ApiProxy['events']['mux']>[0]['payload'],
|
||||
signal: AbortSignal,
|
||||
onOpen?: () => void,
|
||||
): AsyncIterable<RpcRequest<MuxFrame>> {
|
||||
return this.readWebSocket(MUX_EVENTS_PATH, signal, muxFrameSchema, onOpen)
|
||||
}
|
||||
|
||||
protected override openHost(
|
||||
_payload: Parameters<ApiProxy['events']['host']>[0]['payload'],
|
||||
signal: AbortSignal,
|
||||
onOpen?: () => void,
|
||||
): AsyncIterable<RpcRequest<HostFrame>> {
|
||||
return this.readWebSocket(HOST_EVENTS_PATH, signal, hostFrameSchema, onOpen)
|
||||
}
|
||||
|
||||
private async *readWebSocket<F extends MuxFrame | HostFrame>(
|
||||
path: string,
|
||||
signal: AbortSignal,
|
||||
frameSchema: Parser<F>,
|
||||
onOpen?: () => void,
|
||||
): AsyncGenerator<RpcRequest<F>> {
|
||||
const url = new URL(path, this.resolveBase())
|
||||
url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:'
|
||||
const socket = new WebSocket(url)
|
||||
const inbox: SocketItem<F>[] = []
|
||||
let wake: (() => void) | undefined
|
||||
const enqueue = (item: SocketItem<F>): void => {
|
||||
inbox.push(item)
|
||||
wake?.()
|
||||
wake = undefined
|
||||
}
|
||||
const handleOpen = (): void => { onOpen?.() }
|
||||
const handleMessage = (event: MessageEvent): void => {
|
||||
let full: ServerRequest
|
||||
let frame: F
|
||||
try {
|
||||
if (typeof event.data !== 'string') throw new Error('binary WebSocket frame')
|
||||
full = serverRequestSchema.parse(JSON.parse(event.data))
|
||||
frame = frameSchema.parse(full.payload)
|
||||
} catch (error) {
|
||||
console.error(`[client-connection] dropping malformed WebSocket frame on ${path}:`, error)
|
||||
return
|
||||
}
|
||||
this.onEnvelope(full)
|
||||
enqueue({ kind: 'frame', envelope: { rpcId: full.rpcId, payload: frame } })
|
||||
}
|
||||
const handleClose = (): void => { enqueue({ kind: 'end' }) }
|
||||
const handleAbort = (): void => {
|
||||
if (socket.readyState === WebSocket.CONNECTING || socket.readyState === WebSocket.OPEN) socket.close()
|
||||
}
|
||||
socket.addEventListener('open', handleOpen)
|
||||
socket.addEventListener('message', handleMessage)
|
||||
socket.addEventListener('close', handleClose, { once: true })
|
||||
signal.addEventListener('abort', handleAbort, { once: true })
|
||||
if (signal.aborted) handleAbort()
|
||||
try {
|
||||
while (true) {
|
||||
while (inbox.length > 0) {
|
||||
const item = inbox.shift() as SocketItem<F>
|
||||
if (item.kind === 'end') return
|
||||
yield item.envelope
|
||||
}
|
||||
await new Promise<void>((resolve) => { wake = resolve })
|
||||
}
|
||||
} finally {
|
||||
signal.removeEventListener('abort', handleAbort)
|
||||
socket.removeEventListener('open', handleOpen)
|
||||
socket.removeEventListener('message', handleMessage)
|
||||
socket.removeEventListener('close', handleClose)
|
||||
handleAbort()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,13 +2,14 @@
|
||||
import type { Context } from 'cordis'
|
||||
import z from 'schemastery'
|
||||
// Activates the httpServer Context merge used below.
|
||||
import type { WebRoute } from '@deepseek-ai/dsh-host-webserver'
|
||||
import type { WebRoute, WebUpgradeRoute } from '@deepseek-ai/dsh-host-webserver'
|
||||
import { toFetchHandler } from '@deepseek-ai/dsh-host-apiproxy'
|
||||
import { API_PATH } from './api-path.ts'
|
||||
import { API_PATH, HOST_EVENTS_PATH, MUX_EVENTS_PATH } from './api-path.ts'
|
||||
import { bridge } from './http-bridge.ts'
|
||||
import { assertTrustedAuthority, isTrustedApiRequest } from './api-request-trust.ts'
|
||||
import { rejectWebSocketUpgrade, WebSocketDownlinks } from './websocket-downlink.ts'
|
||||
|
||||
export { API_PATH } from './api-path.ts'
|
||||
export { API_PATH, HOST_EVENTS_PATH, MUX_EVENTS_PATH } from './api-path.ts'
|
||||
|
||||
/** Stable Cordis plugin name. */
|
||||
export const name = 'client-connection'
|
||||
@@ -76,6 +77,7 @@ export function apply(ctx: Context, config?: ConnectionConfig): void {
|
||||
// silently authorizing its hostname prefix at request time.
|
||||
for (const entry of trustedHosts) assertTrustedAuthority(entry)
|
||||
const apiHandler = toFetchHandler(ctx.apiProxy)
|
||||
const downlinks = new WebSocketDownlinks(ctx.apiProxy)
|
||||
const route: WebRoute = {
|
||||
kind: 'prefix',
|
||||
path: API_PATH,
|
||||
@@ -92,8 +94,31 @@ export function apply(ctx: Context, config?: ConnectionConfig): void {
|
||||
res.end('forbidden')
|
||||
return
|
||||
}
|
||||
if (req.method === 'GET' && (pathname === MUX_EVENTS_PATH || pathname === HOST_EVENTS_PATH)) {
|
||||
res.writeHead(426, { connection: 'Upgrade', upgrade: 'websocket' })
|
||||
res.end('upgrade required')
|
||||
return
|
||||
}
|
||||
await bridge(req, res, apiHandler)
|
||||
},
|
||||
}
|
||||
ctx.effect(() => ctx.httpServer.register(route), 'client-connection: /api route')
|
||||
const registerDownlink = (
|
||||
path: string,
|
||||
handle: WebUpgradeRoute['handler'],
|
||||
): void => {
|
||||
ctx.effect(() => ctx.httpServer.registerUpgrade({
|
||||
path,
|
||||
handler: (req, socket, head) => {
|
||||
if (!isTrustedApiRequest(req, trustedHosts)) {
|
||||
rejectWebSocketUpgrade(socket)
|
||||
return
|
||||
}
|
||||
return handle(req, socket, head)
|
||||
},
|
||||
}), `client-connection: ${path} WebSocket`)
|
||||
}
|
||||
ctx.effect(() => () => downlinks.close(), 'client-connection: WebSocket downlinks')
|
||||
registerDownlink(MUX_EVENTS_PATH, (req, socket, head) => { downlinks.handleMux(req, socket, head) })
|
||||
registerDownlink(HOST_EVENTS_PATH, (req, socket, head) => { downlinks.handleHost(req, socket, head) })
|
||||
}
|
||||
|
||||
150
packages/client/connection/src/websocket-downlink.ts
Normal file
150
packages/client/connection/src/websocket-downlink.ts
Normal file
@@ -0,0 +1,150 @@
|
||||
/** Host-side WebSocket carrier for the two server-to-browser event streams. */
|
||||
|
||||
import { randomUUID } from 'node:crypto'
|
||||
import type { IncomingMessage } from 'node:http'
|
||||
import type { Duplex } from 'node:stream'
|
||||
import WebSocket, { WebSocketServer } from 'ws'
|
||||
import type {
|
||||
ApiProxy, HostFrame, MuxFrame, RpcRequest, ServerRequest,
|
||||
} from '@deepseek-ai/dsh-host-apiproxy/api'
|
||||
import { RpcId } from '@deepseek-ai/dsh-host-apiproxy/api'
|
||||
|
||||
type Frame = MuxFrame | HostFrame
|
||||
|
||||
function serverRequest(frame: RpcRequest<Frame>): ServerRequest {
|
||||
return {
|
||||
type: 'server-request',
|
||||
rpcId: frame.rpcId,
|
||||
method: frame.payload.type,
|
||||
payload: frame.payload,
|
||||
}
|
||||
}
|
||||
|
||||
function send(socket: WebSocket, frame: RpcRequest<Frame>): Promise<void> {
|
||||
return new Promise((resolve, reject) => {
|
||||
if (socket.readyState !== WebSocket.OPEN) {
|
||||
reject(new Error('websocket downlink closed before frame delivery'))
|
||||
return
|
||||
}
|
||||
socket.send(JSON.stringify(serverRequest(frame)), (error) => {
|
||||
if (error === undefined) resolve()
|
||||
else reject(error)
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
function failureFrame(error: unknown): RpcRequest<Frame> {
|
||||
return {
|
||||
rpcId: RpcId(randomUUID()),
|
||||
payload: {
|
||||
type: 'stream/error',
|
||||
error: { code: 'internal', message: String(error), details: {} },
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Owns WebSocket negotiation and frame pumping for the connection plugin's
|
||||
* two downlinks. Client messages are a protocol violation: upstream traffic
|
||||
* remains on HTTP.
|
||||
*/
|
||||
export class WebSocketDownlinks {
|
||||
private readonly server = new WebSocketServer({ noServer: true })
|
||||
|
||||
/** @param api - host API supplying the typed event streams. */
|
||||
constructor(private readonly api: ApiProxy) {}
|
||||
|
||||
/**
|
||||
* Upgrade one socket and pump the mux stream until either side closes.
|
||||
* @param req - HTTP upgrade request.
|
||||
* @param socket - Raw socket transferred by the HTTP server.
|
||||
* @param head - Bytes already read after the upgrade headers.
|
||||
*/
|
||||
handleMux(req: IncomingMessage, socket: Duplex, head: Buffer): void {
|
||||
this.upgrade(req, socket, head, signal => this.api.events.mux({
|
||||
rpcId: RpcId(randomUUID()),
|
||||
payload: {},
|
||||
}, signal))
|
||||
}
|
||||
|
||||
/**
|
||||
* Upgrade one socket and pump the host stream until either side closes.
|
||||
* @param req - HTTP upgrade request.
|
||||
* @param socket - Raw socket transferred by the HTTP server.
|
||||
* @param head - Bytes already read after the upgrade headers.
|
||||
*/
|
||||
handleHost(req: IncomingMessage, socket: Duplex, head: Buffer): void {
|
||||
this.upgrade(req, socket, head, signal => this.api.events.host({
|
||||
rpcId: RpcId(randomUUID()),
|
||||
payload: {},
|
||||
}, signal))
|
||||
}
|
||||
|
||||
/**
|
||||
* Terminate owned sockets and await the no-server acceptor's close.
|
||||
* @returns A promise resolving after every accepted socket has closed.
|
||||
*/
|
||||
close(): Promise<void> {
|
||||
for (const socket of this.server.clients) socket.terminate()
|
||||
return new Promise((resolve, reject) => {
|
||||
this.server.close((error) => {
|
||||
if (error === undefined) resolve()
|
||||
else reject(error)
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
private upgrade<F extends Frame>(
|
||||
req: IncomingMessage,
|
||||
socket: Duplex,
|
||||
head: Buffer,
|
||||
open: (signal: AbortSignal) => AsyncIterable<RpcRequest<F>>,
|
||||
): void {
|
||||
this.server.handleUpgrade(req, socket, head, (websocket) => {
|
||||
const abort = new AbortController()
|
||||
websocket.once('close', () => { abort.abort() })
|
||||
websocket.once('error', () => { abort.abort() })
|
||||
websocket.once('message', () => {
|
||||
websocket.close(1008, 'downlink only')
|
||||
})
|
||||
void this.pump(websocket, open(abort.signal), abort)
|
||||
})
|
||||
}
|
||||
|
||||
private async pump<F extends Frame>(
|
||||
socket: WebSocket,
|
||||
frames: AsyncIterable<RpcRequest<F>>,
|
||||
abort: AbortController,
|
||||
): Promise<void> {
|
||||
try {
|
||||
for await (const frame of frames) await send(socket, frame)
|
||||
} catch (error) {
|
||||
if (!abort.signal.aborted) {
|
||||
try {
|
||||
await send(socket, failureFrame(error))
|
||||
} catch {
|
||||
// Socket loss won the race; no downstream remains to receive the failure frame.
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
abort.abort()
|
||||
if (socket.readyState === WebSocket.OPEN) socket.close()
|
||||
else if (socket.readyState === WebSocket.CONNECTING) socket.terminate()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Reject an untrusted upgrade before protocol negotiation.
|
||||
* @param socket - Raw HTTP socket that remains owned by the caller.
|
||||
*/
|
||||
export function rejectWebSocketUpgrade(socket: Duplex): void {
|
||||
socket.end([
|
||||
'HTTP/1.1 403 Forbidden',
|
||||
'Connection: close',
|
||||
'Content-Type: text/plain; charset=utf-8',
|
||||
'Content-Length: 9',
|
||||
'',
|
||||
'forbidden',
|
||||
].join('\r\n'))
|
||||
}
|
||||
@@ -3,15 +3,55 @@
|
||||
* selection off the page URL, and the single-consumer stream-loop ownership.
|
||||
*/
|
||||
import { Context } from 'cordis'
|
||||
import { afterEach, describe, expect, it } from 'vitest'
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import { apply, type ConnectionHandle } from '../src/client/index.ts'
|
||||
import type { RpcMessage } from '../src/client/api.ts'
|
||||
import { RpcId } from '../src/client/api.ts'
|
||||
import { FixtureApiClient } from '../src/client/fixture.ts'
|
||||
import { WebApiClient } from '../src/client/web-api-client.ts'
|
||||
|
||||
type Win = { location?: { hostname: string; search: string } }
|
||||
type Win = { location?: { hostname: string; search: string; origin?: string } }
|
||||
type WebSocketGlobal = { WebSocket?: typeof WebSocket }
|
||||
|
||||
const originalWebSocket = globalThis.WebSocket
|
||||
const sockets: FakeWebSocket[] = []
|
||||
|
||||
class FakeWebSocket extends EventTarget {
|
||||
static readonly CONNECTING = 0
|
||||
static readonly OPEN = 1
|
||||
static readonly CLOSING = 2
|
||||
static readonly CLOSED = 3
|
||||
|
||||
readonly url: string
|
||||
readyState = FakeWebSocket.CONNECTING
|
||||
|
||||
constructor(url: string | URL) {
|
||||
super()
|
||||
this.url = String(url)
|
||||
sockets.push(this)
|
||||
queueMicrotask(() => {
|
||||
if (this.readyState !== FakeWebSocket.CONNECTING) return
|
||||
this.readyState = FakeWebSocket.OPEN
|
||||
this.dispatchEvent(new Event('open'))
|
||||
})
|
||||
}
|
||||
|
||||
close(): void {
|
||||
if (this.readyState === FakeWebSocket.CLOSED) return
|
||||
this.readyState = FakeWebSocket.CLOSED
|
||||
this.dispatchEvent(new Event('close'))
|
||||
}
|
||||
|
||||
receive(data: unknown): void {
|
||||
this.dispatchEvent(new MessageEvent('message', { data }))
|
||||
}
|
||||
}
|
||||
|
||||
afterEach(() => {
|
||||
delete (globalThis as Win).location
|
||||
sockets.length = 0
|
||||
if (originalWebSocket === undefined) delete (globalThis as WebSocketGlobal).WebSocket
|
||||
else globalThis.WebSocket = originalWebSocket
|
||||
})
|
||||
|
||||
async function mount(): Promise<ConnectionHandle> {
|
||||
@@ -53,7 +93,7 @@ describe('connection client apply', () => {
|
||||
loop.stop() // teardown must not throw; the fixture streams abort quietly
|
||||
})
|
||||
|
||||
it('WebApiClient carries requests over globalThis.fetch', async () => {
|
||||
it('WebApiClient keeps unary calls and respond on globalThis.fetch', async () => {
|
||||
;(globalThis as Win).location = { hostname: 'localhost', search: '' }
|
||||
const handle = await mount()
|
||||
const original = globalThis.fetch
|
||||
@@ -65,9 +105,88 @@ describe('connection client apply', () => {
|
||||
try {
|
||||
// Schema rejection is fine — the transport hop is the assertion.
|
||||
await (handle.api as WebApiClient).host.describe({}).catch(() => undefined)
|
||||
await handle.api.respond({
|
||||
type: 'client-response',
|
||||
rpcId: RpcId('response-over-http'),
|
||||
result: { ok: true, value: {} },
|
||||
}).catch(() => undefined)
|
||||
} finally {
|
||||
globalThis.fetch = original
|
||||
}
|
||||
expect(seen.some(u => u.includes('/api/'))).toBe(true)
|
||||
expect(seen.some(u => u.includes('/api/host.describe'))).toBe(true)
|
||||
expect(seen.some(u => u.includes('/api/respond'))).toBe(true)
|
||||
})
|
||||
|
||||
it('opens one WebSocket per downlink, parses frames, and aborts both without using fetch', async () => {
|
||||
;(globalThis as Win).location = {
|
||||
hostname: 'localhost', search: '', origin: 'http://localhost:3080',
|
||||
}
|
||||
;(globalThis as WebSocketGlobal).WebSocket = FakeWebSocket as unknown as typeof WebSocket
|
||||
const fetch = vi.spyOn(globalThis, 'fetch')
|
||||
const client = (await mount()).api as WebApiClient
|
||||
const envelopes: RpcMessage[][] = []
|
||||
client.subscribeEnvelopes(batch => { envelopes.push([...batch]) })
|
||||
const opened: string[] = []
|
||||
const muxAbort = new AbortController()
|
||||
const hostAbort = new AbortController()
|
||||
const mux = client.events.mux({}, muxAbort.signal, () => { opened.push('mux') })[Symbol.asyncIterator]()
|
||||
const host = client.events.host({}, hostAbort.signal, () => { opened.push('host') })[Symbol.asyncIterator]()
|
||||
const muxFrame = mux.next()
|
||||
const hostFrame = host.next()
|
||||
await vi.waitFor(() => { expect(sockets).toHaveLength(2) })
|
||||
expect(sockets.map(socket => socket.url)).toEqual([
|
||||
'ws://localhost:3080/api/events.mux',
|
||||
'ws://localhost:3080/api/events.host',
|
||||
])
|
||||
await vi.waitFor(() => { expect(opened).toEqual(['mux', 'host']) })
|
||||
|
||||
const errors = vi.spyOn(console, 'error').mockImplementation(() => {})
|
||||
sockets[0]!.receive(new Uint8Array([1, 2, 3]))
|
||||
sockets[1]!.receive(JSON.stringify({ type: 'server-request', rpcId: 'bad', method: 'host/session-status', payload: {} }))
|
||||
sockets[0]!.receive(JSON.stringify({
|
||||
type: 'server-request',
|
||||
rpcId: 'mux-browser',
|
||||
method: 'session/subscribed',
|
||||
payload: { type: 'session/subscribed', sessionId: 'session-browser', lastSeq: 8 },
|
||||
}))
|
||||
sockets[1]!.receive(JSON.stringify({
|
||||
type: 'server-request',
|
||||
rpcId: 'host-browser',
|
||||
method: 'host/commands-changed',
|
||||
payload: { type: 'host/commands-changed' },
|
||||
}))
|
||||
expect(await muxFrame).toMatchObject({
|
||||
value: { rpcId: 'mux-browser', payload: { type: 'session/subscribed', lastSeq: 8 } },
|
||||
})
|
||||
expect(await hostFrame).toMatchObject({
|
||||
value: { rpcId: 'host-browser', payload: { type: 'host/commands-changed' } },
|
||||
})
|
||||
expect(errors).toHaveBeenCalledTimes(2)
|
||||
await vi.waitFor(() => { expect(envelopes.flat()).toHaveLength(2) })
|
||||
expect(fetch).not.toHaveBeenCalled()
|
||||
|
||||
const muxEnd = mux.next()
|
||||
const hostEnd = host.next()
|
||||
muxAbort.abort()
|
||||
hostAbort.abort()
|
||||
await expect(muxEnd).resolves.toMatchObject({ done: true })
|
||||
await expect(hostEnd).resolves.toMatchObject({ done: true })
|
||||
expect(sockets.every(socket => socket.readyState === FakeWebSocket.CLOSED)).toBe(true)
|
||||
errors.mockRestore()
|
||||
fetch.mockRestore()
|
||||
})
|
||||
|
||||
it('maps an HTTPS page origin to a secure WebSocket URL', async () => {
|
||||
;(globalThis as Win).location = {
|
||||
hostname: 'harness.example', search: '', origin: 'https://harness.example',
|
||||
}
|
||||
;(globalThis as WebSocketGlobal).WebSocket = FakeWebSocket as unknown as typeof WebSocket
|
||||
const client = (await mount()).api
|
||||
const abort = new AbortController()
|
||||
const iterator = client.events.mux({}, abort.signal)[Symbol.asyncIterator]()
|
||||
const pending = iterator.next()
|
||||
await vi.waitFor(() => { expect(sockets[0]?.url).toBe('wss://harness.example/api/events.mux') })
|
||||
abort.abort()
|
||||
await expect(pending).resolves.toMatchObject({ done: true })
|
||||
})
|
||||
})
|
||||
|
||||
@@ -1,22 +1,29 @@
|
||||
/** Node half: registers the /api prefix route bridging to the api gateway. */
|
||||
import { EventEmitter } from 'node:events'
|
||||
import { EventEmitter, once } from 'node:events'
|
||||
import { createServer, request as httpRequest } from 'node:http'
|
||||
import { Readable } from 'node:stream'
|
||||
import { PassThrough, Readable } from 'node:stream'
|
||||
import { Context } from 'cordis'
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import type { AddressInfo } from 'node:net'
|
||||
import type { IncomingMessage, ServerResponse } from 'node:http'
|
||||
import type { ApiProxy } from '@deepseek-ai/dsh-host-apiproxy/api'
|
||||
import type { HttpServerService, WebRoute } from '@deepseek-ai/dsh-host-webserver'
|
||||
import { API_PATH, apply, inject } from '../src/index.ts'
|
||||
import type { HttpServerService, WebRoute, WebUpgradeRoute } from '@deepseek-ai/dsh-host-webserver'
|
||||
import { API_PATH, apply, HOST_EVENTS_PATH, inject, MUX_EVENTS_PATH } from '../src/index.ts'
|
||||
|
||||
/** Structural httpServer fake: the plugin only touches register(). */
|
||||
function fakeHttpServer(routes: WebRoute[]): Pick<HttpServerService, 'register' | 'tapIndex' | 'port'> {
|
||||
/** Structural httpServer fake recording both route registries. */
|
||||
function fakeHttpServer(
|
||||
routes: WebRoute[],
|
||||
upgrades: WebUpgradeRoute[],
|
||||
): Pick<HttpServerService, 'register' | 'registerUpgrade' | 'tapIndex' | 'port'> {
|
||||
return {
|
||||
register(route) {
|
||||
routes.push(route)
|
||||
return () => { routes.splice(routes.indexOf(route), 1) }
|
||||
},
|
||||
registerUpgrade(route) {
|
||||
upgrades.push(route)
|
||||
return () => { upgrades.splice(upgrades.indexOf(route), 1) }
|
||||
},
|
||||
tapIndex: () => () => {},
|
||||
port: 0,
|
||||
}
|
||||
@@ -45,33 +52,67 @@ function fakeResponse(): { response: ServerResponse; state: { status?: number; b
|
||||
return { response, state }
|
||||
}
|
||||
|
||||
async function mounted(config?: { trustedHosts?: string[] }): Promise<{ routes: WebRoute[]; dispose: () => Promise<void> }> {
|
||||
async function mounted(config?: { trustedHosts?: string[] }): Promise<{
|
||||
routes: WebRoute[]
|
||||
upgrades: WebUpgradeRoute[]
|
||||
dispose: () => Promise<void>
|
||||
}> {
|
||||
const ctx = new Context()
|
||||
const routes: WebRoute[] = []
|
||||
ctx.provide('httpServer', fakeHttpServer(routes) as HttpServerService)
|
||||
const upgrades: WebUpgradeRoute[] = []
|
||||
ctx.provide('httpServer', fakeHttpServer(routes, upgrades) as HttpServerService)
|
||||
ctx.provide('apiProxy', {} as unknown as ApiProxy)
|
||||
const fiber = ctx.plugin({ inject: [...inject], apply }, config)
|
||||
await fiber.await()
|
||||
return { routes, dispose: () => fiber.dispose() }
|
||||
return { routes, upgrades, dispose: () => fiber.dispose() }
|
||||
}
|
||||
|
||||
describe('connection node half', () => {
|
||||
it('fails the load on a trustedHosts entry that is not a bare authority', async () => {
|
||||
const routes: WebRoute[] = []
|
||||
const upgrades: WebUpgradeRoute[] = []
|
||||
const ctx = new Context()
|
||||
ctx.provide('httpServer', fakeHttpServer(routes) as HttpServerService)
|
||||
ctx.provide('httpServer', fakeHttpServer(routes, upgrades) as HttpServerService)
|
||||
ctx.provide('apiProxy', {} as unknown as ApiProxy)
|
||||
const fiber = ctx.plugin({ inject: [...inject], apply }, { trustedHosts: ['harness.internal/path'] })
|
||||
await expect(fiber).rejects.toThrow(/not a bare host\[:port\] authority/)
|
||||
expect(routes).toHaveLength(0)
|
||||
expect(upgrades).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('registers the /api prefix route and removes it with the fiber', async () => {
|
||||
const { routes, dispose } = await mounted()
|
||||
it('registers one HTTP route plus one upgrade route per downlink and removes all three with the fiber', async () => {
|
||||
const { routes, upgrades, dispose } = await mounted()
|
||||
expect(routes).toHaveLength(1)
|
||||
expect(routes[0]).toMatchObject({ kind: 'prefix', path: API_PATH })
|
||||
expect(upgrades.map(route => route.path)).toEqual([MUX_EVENTS_PATH, HOST_EVENTS_PATH])
|
||||
await dispose()
|
||||
expect(routes).toHaveLength(0)
|
||||
expect(upgrades).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('requires WebSocket upgrade for network GETs to either event path', async () => {
|
||||
const { routes, dispose } = await mounted()
|
||||
for (const path of [MUX_EVENTS_PATH, HOST_EVENTS_PATH]) {
|
||||
const { response, state } = fakeResponse()
|
||||
await routes[0]!.handler(fakeRequest({ host: '127.0.0.1:3080' }, path), response)
|
||||
expect(state.status).toBe(426)
|
||||
expect(state.body).toBe('upgrade required')
|
||||
}
|
||||
await dispose()
|
||||
})
|
||||
|
||||
it('rejects an untrusted WebSocket upgrade before protocol negotiation', async () => {
|
||||
const { upgrades, dispose } = await mounted()
|
||||
const socket = new PassThrough()
|
||||
const chunks: Buffer[] = []
|
||||
socket.on('data', (chunk: Buffer) => { chunks.push(chunk) })
|
||||
const ended = once(socket, 'end')
|
||||
await upgrades[0]!.handler(fakeRequest({
|
||||
host: 'harness.example', origin: 'http://harness.example', 'sec-fetch-site': 'same-origin',
|
||||
}, MUX_EVENTS_PATH), socket, Buffer.alloc(0))
|
||||
await ended
|
||||
expect(Buffer.concat(chunks).toString()).toContain('HTTP/1.1 403 Forbidden')
|
||||
await dispose()
|
||||
})
|
||||
|
||||
it('refuses an untrusted Host on any /api path before the bridge runs', async () => {
|
||||
|
||||
164
packages/client/connection/tests/websocket-downlink.spec.ts
Normal file
164
packages/client/connection/tests/websocket-downlink.spec.ts
Normal file
@@ -0,0 +1,164 @@
|
||||
import { once } from 'node:events'
|
||||
import { createServer } from 'node:http'
|
||||
import type { AddressInfo } from 'node:net'
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import WebSocket from 'ws'
|
||||
import type {
|
||||
ApiProxy, HostFrame, MuxFrame, RpcRequest, ServerRequest,
|
||||
} from '@deepseek-ai/dsh-host-apiproxy/api'
|
||||
import { RpcId } from '@deepseek-ai/dsh-host-apiproxy/api'
|
||||
import { HOST_EVENTS_PATH, MUX_EVENTS_PATH } from '../src/api-path.ts'
|
||||
import { WebSocketDownlinks } from '../src/websocket-downlink.ts'
|
||||
|
||||
type MuxSource = (signal: AbortSignal) => AsyncIterable<RpcRequest<MuxFrame>>
|
||||
type HostSource = (signal: AbortSignal) => AsyncIterable<RpcRequest<HostFrame>>
|
||||
|
||||
const running: (() => Promise<void>)[] = []
|
||||
|
||||
afterEach(async () => {
|
||||
await Promise.all(running.splice(0).map(close => close()))
|
||||
})
|
||||
|
||||
function untilAbort(signal: AbortSignal): Promise<void> {
|
||||
if (signal.aborted) return Promise.resolve()
|
||||
return new Promise(resolve => signal.addEventListener('abort', () => { resolve() }, { once: true }))
|
||||
}
|
||||
|
||||
async function * idle<F>(signal: AbortSignal): AsyncGenerator<RpcRequest<F>> {
|
||||
await untilAbort(signal)
|
||||
}
|
||||
|
||||
function api(mux: MuxSource, host: HostSource): ApiProxy {
|
||||
return {
|
||||
events: {
|
||||
mux: (_request, signal) => mux(signal),
|
||||
host: (_request, signal) => host(signal),
|
||||
},
|
||||
} as ApiProxy
|
||||
}
|
||||
|
||||
async function serve(downlinks: WebSocketDownlinks): Promise<{
|
||||
origin: string
|
||||
close: () => Promise<void>
|
||||
}> {
|
||||
const server = createServer()
|
||||
server.on('upgrade', (request, socket, head) => {
|
||||
const pathname = new URL(request.url ?? '/', 'http://dsh.internal').pathname
|
||||
if (pathname === MUX_EVENTS_PATH) downlinks.handleMux(request, socket, head)
|
||||
else if (pathname === HOST_EVENTS_PATH) downlinks.handleHost(request, socket, head)
|
||||
else socket.destroy()
|
||||
})
|
||||
await new Promise<void>(resolve => server.listen(0, '127.0.0.1', resolve))
|
||||
const port = (server.address() as AddressInfo).port
|
||||
return {
|
||||
origin: `ws://127.0.0.1:${String(port)}`,
|
||||
close: async () => {
|
||||
await downlinks.close()
|
||||
await new Promise<void>(resolve => server.close(() => { resolve() }))
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
function read(socket: WebSocket): Promise<ServerRequest> {
|
||||
return once(socket, 'message').then(([data]) => JSON.parse(String(data)) as ServerRequest)
|
||||
}
|
||||
|
||||
describe('WebSocket downlinks', () => {
|
||||
it('carries mux and host over independent downstream sockets and cancels each source on close', async () => {
|
||||
let muxAborted = false
|
||||
let hostAborted = false
|
||||
const downlinks = new WebSocketDownlinks(api(
|
||||
async function * (signal) {
|
||||
try {
|
||||
yield {
|
||||
rpcId: RpcId('mux-1'),
|
||||
payload: { type: 'session/subscribed', sessionId: 'session-1' as never, lastSeq: 4 },
|
||||
}
|
||||
await untilAbort(signal)
|
||||
} finally {
|
||||
muxAborted = true
|
||||
}
|
||||
},
|
||||
async function * (signal) {
|
||||
try {
|
||||
yield { rpcId: RpcId('host-1'), payload: { type: 'host/commands-changed' } }
|
||||
await untilAbort(signal)
|
||||
} finally {
|
||||
hostAborted = true
|
||||
}
|
||||
},
|
||||
))
|
||||
const host = await serve(downlinks)
|
||||
running.push(host.close)
|
||||
|
||||
const mux = new WebSocket(`${host.origin}${MUX_EVENTS_PATH}`)
|
||||
const hostSocket = new WebSocket(`${host.origin}${HOST_EVENTS_PATH}`)
|
||||
const muxFrame = read(mux)
|
||||
const hostFrame = read(hostSocket)
|
||||
expect(await muxFrame).toEqual({
|
||||
type: 'server-request',
|
||||
rpcId: 'mux-1',
|
||||
method: 'session/subscribed',
|
||||
payload: { type: 'session/subscribed', sessionId: 'session-1', lastSeq: 4 },
|
||||
})
|
||||
expect(await hostFrame).toEqual({
|
||||
type: 'server-request',
|
||||
rpcId: 'host-1',
|
||||
method: 'host/commands-changed',
|
||||
payload: { type: 'host/commands-changed' },
|
||||
})
|
||||
|
||||
const muxClosed = once(mux, 'close')
|
||||
const hostClosed = once(hostSocket, 'close')
|
||||
mux.close()
|
||||
hostSocket.close()
|
||||
await Promise.all([muxClosed, hostClosed])
|
||||
await vi.waitFor(() => {
|
||||
expect(muxAborted).toBe(true)
|
||||
expect(hostAborted).toBe(true)
|
||||
})
|
||||
})
|
||||
|
||||
it('rejects client messages because upstream remains HTTP', async () => {
|
||||
let aborted = false
|
||||
const downlinks = new WebSocketDownlinks(api(
|
||||
async function * (signal) {
|
||||
try {
|
||||
await untilAbort(signal)
|
||||
} finally {
|
||||
aborted = true
|
||||
}
|
||||
},
|
||||
idle,
|
||||
))
|
||||
const host = await serve(downlinks)
|
||||
running.push(host.close)
|
||||
const socket = new WebSocket(`${host.origin}${MUX_EVENTS_PATH}`)
|
||||
await once(socket, 'open')
|
||||
const closed = once(socket, 'close')
|
||||
socket.send('upstream payload')
|
||||
const [code, reason] = await closed
|
||||
expect(code).toBe(1008)
|
||||
expect(String(reason)).toBe('downlink only')
|
||||
await vi.waitFor(() => { expect(aborted).toBe(true) })
|
||||
})
|
||||
|
||||
it('sends stream/error before closing when a source fails', async () => {
|
||||
const downlinks = new WebSocketDownlinks(api(
|
||||
async function * () {
|
||||
throw new Error('mux source failed')
|
||||
},
|
||||
idle,
|
||||
))
|
||||
const host = await serve(downlinks)
|
||||
running.push(host.close)
|
||||
const socket = new WebSocket(`${host.origin}${MUX_EVENTS_PATH}`)
|
||||
const failure = read(socket)
|
||||
const closed = once(socket, 'close')
|
||||
expect((await failure).payload).toEqual({
|
||||
type: 'stream/error',
|
||||
error: { code: 'internal', message: 'Error: mux source failed', details: {} },
|
||||
})
|
||||
await closed
|
||||
})
|
||||
})
|
||||
Reference in New Issue
Block a user