From a630b8860eddb3cf762f35a01887886aecce33c2 Mon Sep 17 00:00:00 2001 From: shekohex Date: Sun, 8 Feb 2026 15:39:22 +0200 Subject: [PATCH] feat(terminal): add persistent websocket transport for low-latency input (#348) * fix(server): resolve terminal WebSocket proxy conflict and implement server-side transport - Disables proxy websocket handling conflict in server proxy config to fix 1006 abnormal closes. - Implements server-side WebSocket upgrade and connection handling for terminal input. - Adds server-side debug instrumentation for WS lifecycle events. - Includes terminal input WS protocol definition and unit tests. * feat(ui): implement hardened terminal WebSocket transport with idempotency and diagnostics - Adds client-side WebSocket transport manager with automatic reconnection and jitter. - Implements idempotency and debug instrumentation for terminal input WS. - Adds HMR dispose cleanup for terminal WS transport manager. - Primes terminal input transport when terminal view becomes active. - Updates terminal session types to include input capabilities. * refactor(terminal): remove temporary websocket debug instrumentation --- bun.lock | 3 + .../ui/src/components/views/TerminalView.tsx | 9 + packages/ui/src/lib/api/types.ts | 11 + packages/ui/src/lib/terminalApi.ts | 535 +++++++++++++++++- packages/web/package.json | 1 + .../web/server/TERMINAL_INPUT_WS_PROTOCOL.md | 44 ++ packages/web/server/index.js | 312 +++++++++- .../server/lib/terminal-input-ws-protocol.js | 66 +++ .../lib/terminal-input-ws-protocol.test.js | 138 +++++ 9 files changed, 1105 insertions(+), 14 deletions(-) create mode 100644 packages/web/server/TERMINAL_INPUT_WS_PROTOCOL.md create mode 100644 packages/web/server/lib/terminal-input-ws-protocol.js create mode 100644 packages/web/server/lib/terminal-input-ws-protocol.test.js diff --git a/bun.lock b/bun.lock index 19e12767..7b51cafe 100644 --- a/bun.lock +++ b/bun.lock @@ -266,6 +266,7 @@ "strip-json-comments": "^5.0.3", "tailwind-merge": "^3.3.1", "web-push": "^3.6.7", + "ws": "^8.18.3", "yaml": "^2.8.1", "zustand": "^5.0.8", }, @@ -3134,6 +3135,8 @@ "wrappy": ["wrappy@1.0.2", "", {}, "sha512-l4Sp/DRseor9wL6EvV2+TuQn63dMkPjZ/sp9XkghTEbV9KlPS1xUsZ3u7/IQO4wxtcFB4bgpQPRcR3QCvezPcQ=="], + "ws": ["ws@8.19.0", "", { "peerDependencies": { "bufferutil": "^4.0.1", "utf-8-validate": ">=5.0.2" }, "optionalPeers": ["bufferutil", "utf-8-validate"] }, "sha512-blAT2mjOEIi0ZzruJfIhb3nps74PRWTCz1IjglWEEpQl5XS/UNama6u2/rjFkDDouqr4L67ry+1aGIALViWjDg=="], + "wsl-utils": ["wsl-utils@0.1.0", "", { "dependencies": { "is-wsl": "^3.1.0" } }, "sha512-h3Fbisa2nKGPxCpm89Hk33lBLsnaGBvctQopaBSOW/uIs6FTe1ATyAnKFJrzVs9vpGdsTe73WF3V4lIsk4Gacw=="], "xml2js": ["xml2js@0.5.0", "", { "dependencies": { "sax": ">=0.6.0", "xmlbuilder": "~11.0.0" } }, "sha512-drPFnkQJik/O+uPKpqSgr22mpuFHqKdbS835iAQrUC73L2F5WkboIRd63ai/2Yg6I1jzifPFKH2NTK+cfglkIA=="], diff --git a/packages/ui/src/components/views/TerminalView.tsx b/packages/ui/src/components/views/TerminalView.tsx index bbf1ec87..d3973447 100644 --- a/packages/ui/src/components/views/TerminalView.tsx +++ b/packages/ui/src/components/views/TerminalView.tsx @@ -17,6 +17,7 @@ import { useUIStore } from '@/stores/useUIStore'; import { Button } from '@/components/ui/button'; import { useDeviceInfo } from '@/lib/device'; import { useRuntimeAPIs } from '@/hooks/useRuntimeAPIs'; +import { primeTerminalInputTransport } from '@/lib/terminalApi'; type Modifier = 'ctrl' | 'cmd'; type MobileKey = @@ -178,6 +179,14 @@ export const TerminalView: React.FC = () => { const activeMainTab = useUIStore((state) => state.activeMainTab); const isTerminalActive = activeMainTab === 'terminal'; + React.useEffect(() => { + if (!isTerminalActive || runtime.platform === 'vscode') { + return; + } + + primeTerminalInputTransport(); + }, [isTerminalActive, runtime.platform]); + React.useEffect(() => { terminalIdRef.current = terminalSessionId; }, [terminalSessionId]); diff --git a/packages/ui/src/lib/api/types.ts b/packages/ui/src/lib/api/types.ts index 67e65515..255d1a01 100644 --- a/packages/ui/src/lib/api/types.ts +++ b/packages/ui/src/lib/api/types.ts @@ -33,6 +33,17 @@ export interface TerminalSession { sessionId: string; cols: number; rows: number; + capabilities?: { + input?: { + preferred?: 'ws' | 'http'; + transports?: Array<'ws' | 'http'>; + ws?: { + path: string; + v?: number; + enc?: string; + }; + }; + }; } export interface TerminalStreamEvent { diff --git a/packages/ui/src/lib/terminalApi.ts b/packages/ui/src/lib/terminalApi.ts index 87a37e11..c11ae71c 100644 --- a/packages/ui/src/lib/terminalApi.ts +++ b/packages/ui/src/lib/terminalApi.ts @@ -4,6 +4,19 @@ export interface TerminalSession { sessionId: string; cols: number; rows: number; + capabilities?: { + input?: TerminalInputCapability; + }; +} + +export interface TerminalInputCapability { + preferred?: 'ws' | 'http'; + transports?: Array<'ws' | 'http'>; + ws?: { + path: string; + v?: number; + enc?: string; + }; } export interface TerminalStreamEvent { @@ -28,6 +41,454 @@ export interface ConnectStreamOptions { connectionTimeout?: number; } +type TerminalInputControlMessage = { + t: string; + s?: string; + c?: string; + f?: boolean; + v?: number; +}; + +const CONTROL_TAG_JSON = 0x01; +const WS_READY_STATE_OPEN = 1; +const DEFAULT_TERMINAL_INPUT_WS_PATH = '/api/terminal/input-ws'; +const WS_SEND_WAIT_MS = 1200; +const WS_RECONNECT_INITIAL_DELAY_MS = 1000; +const WS_RECONNECT_MAX_DELAY_MS = 30000; +const WS_RECONNECT_JITTER_MS = 250; +const WS_KEEPALIVE_INTERVAL_MS = 20000; +const WS_CONNECT_TIMEOUT_MS = 5000; +const GLOBAL_TERMINAL_INPUT_STATE_KEY = '__openchamberTerminalInputWsState'; + +const textEncoder = new TextEncoder(); +const textDecoder = new TextDecoder(); + +const normalizeWebSocketPath = (pathValue: string): string => { + if (/^wss?:\/\//i.test(pathValue)) { + return pathValue; + } + + if (/^https?:\/\//i.test(pathValue)) { + const url = new URL(pathValue); + url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:'; + return url.toString(); + } + + if (typeof window === 'undefined') { + return ''; + } + + const normalizedPath = pathValue.startsWith('/') ? pathValue : `/${pathValue}`; + const protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:'; + return `${protocol}//${window.location.host}${normalizedPath}`; +}; + +const encodeControlFrame = (payload: TerminalInputControlMessage): Uint8Array => { + const jsonBytes = textEncoder.encode(JSON.stringify(payload)); + const bytes = new Uint8Array(jsonBytes.length + 1); + bytes[0] = CONTROL_TAG_JSON; + bytes.set(jsonBytes, 1); + return bytes; +}; + +const isWsInputSupported = (capability: TerminalInputCapability | null): boolean => { + if (!capability) return false; + const transports = capability.transports ?? []; + const supportsTransport = transports.includes('ws') || capability.preferred === 'ws'; + return supportsTransport && typeof capability.ws?.path === 'string' && capability.ws.path.length > 0; +}; + +class TerminalInputWsManager { + private socket: WebSocket | null = null; + private socketUrl = ''; + private boundSessionId: string | null = null; + private openPromise: Promise | null = null; + private reconnectTimeout: ReturnType | null = null; + private reconnectAttempt = 0; + private keepaliveInterval: ReturnType | null = null; + private closed = false; + + configure(socketUrl: string): void { + if (!socketUrl) return; + + if (this.socketUrl === socketUrl) { + this.closed = false; + if (this.isConnectedOrConnecting()) { + return; + } + + this.ensureConnected(); + return; + } + + this.socketUrl = socketUrl; + this.closed = false; + this.resetConnection(); + this.ensureConnected(); + } + + async sendInput(sessionId: string, data: string): Promise { + if (!sessionId || !data || this.closed || !this.socketUrl) { + return false; + } + + const socket = await this.getOpenSocket(WS_SEND_WAIT_MS); + if (!socket || socket.readyState !== WS_READY_STATE_OPEN) { + return false; + } + + try { + if (this.boundSessionId !== sessionId) { + socket.send(encodeControlFrame({ t: 'b', s: sessionId, v: 1 })); + this.boundSessionId = sessionId; + } + socket.send(data); + return true; + } catch { + this.handleSocketFailure(); + return false; + } + } + + unbindSession(sessionId: string): void { + if (!sessionId) return; + if (this.boundSessionId === sessionId) { + this.boundSessionId = null; + } + } + + close(): void { + this.closed = true; + this.clearReconnectTimeout(); + this.resetConnection(); + this.socketUrl = ''; + } + + prime(): void { + if (this.closed || !this.socketUrl) { + return; + } + + if (this.isConnectedOrConnecting()) { + return; + } + + this.ensureConnected(); + } + + isConnectedOrConnecting(socketUrl?: string): boolean { + if (this.closed) { + return false; + } + + if (socketUrl && this.socketUrl !== socketUrl) { + return false; + } + + if (this.socket && (this.socket.readyState === WebSocket.OPEN || this.socket.readyState === WebSocket.CONNECTING)) { + return true; + } + + return this.openPromise !== null; + } + + private sendControl(payload: TerminalInputControlMessage): boolean { + if (!this.socket || this.socket.readyState !== WS_READY_STATE_OPEN) { + return false; + } + + try { + this.socket.send(encodeControlFrame(payload)); + return true; + } catch { + this.handleSocketFailure(); + return false; + } + } + + private startKeepalive(): void { + this.stopKeepalive(); + this.keepaliveInterval = setInterval(() => { + if (this.closed) { + return; + } + + this.sendControl({ t: 'p', v: 1 }); + }, WS_KEEPALIVE_INTERVAL_MS); + } + + private stopKeepalive(): void { + if (!this.keepaliveInterval) { + return; + } + + clearInterval(this.keepaliveInterval); + this.keepaliveInterval = null; + } + + private scheduleReconnect(): void { + if (this.closed || !this.socketUrl || this.reconnectTimeout) { + return; + } + + const baseDelay = Math.min( + WS_RECONNECT_INITIAL_DELAY_MS * Math.pow(2, this.reconnectAttempt), + WS_RECONNECT_MAX_DELAY_MS + ); + const jitter = Math.floor(Math.random() * WS_RECONNECT_JITTER_MS); + const delay = baseDelay + jitter; + + this.reconnectTimeout = setTimeout(() => { + this.reconnectTimeout = null; + this.reconnectAttempt += 1; + this.ensureConnected(); + }, delay); + } + + private clearReconnectTimeout(): void { + if (!this.reconnectTimeout) { + return; + } + + clearTimeout(this.reconnectTimeout); + this.reconnectTimeout = null; + } + + private async getOpenSocket(waitMs: number): Promise { + if (this.socket && this.socket.readyState === WS_READY_STATE_OPEN) { + return this.socket; + } + + this.ensureConnected(); + + if (this.socket && this.socket.readyState === WS_READY_STATE_OPEN) { + return this.socket; + } + + const opened = await Promise.race([ + this.openPromise ?? Promise.resolve(null), + new Promise((resolve) => { + setTimeout(() => resolve(null), waitMs); + }), + ]); + + if (opened && opened.readyState === WS_READY_STATE_OPEN) { + return opened; + } + + if (this.socket && this.socket.readyState === WS_READY_STATE_OPEN) { + return this.socket; + } + + return null; + } + + private ensureConnected(): void { + if (this.closed || !this.socketUrl) { + return; + } + + if (this.socket && (this.socket.readyState === WebSocket.OPEN || this.socket.readyState === WebSocket.CONNECTING)) { + return; + } + + if (this.openPromise) { + return; + } + + this.clearReconnectTimeout(); + + this.openPromise = new Promise((resolve) => { + let settled = false; + let connectTimeout: ReturnType | null = null; + + const settle = (value: WebSocket | null) => { + if (settled) { + return; + } + settled = true; + if (connectTimeout) { + clearTimeout(connectTimeout); + connectTimeout = null; + } + this.openPromise = null; + resolve(value); + }; + + try { + const socket = new WebSocket(this.socketUrl); + socket.binaryType = 'arraybuffer'; + + socket.onopen = () => { + this.socket = socket; + this.reconnectAttempt = 0; + this.startKeepalive(); + settle(socket); + }; + + socket.onmessage = (event) => { + void this.handleSocketMessage(event.data); + }; + + socket.onclose = () => { + if (this.socket === socket) { + this.socket = null; + this.boundSessionId = null; + this.stopKeepalive(); + if (!this.closed) { + this.scheduleReconnect(); + } + } + settle(null); + }; + + this.socket = socket; + + connectTimeout = setTimeout(() => { + if (socket.readyState === WebSocket.CONNECTING) { + socket.close(); + settle(null); + } + }, WS_CONNECT_TIMEOUT_MS); + } catch { + settle(null); + if (!this.closed) { + this.scheduleReconnect(); + } + } + }); + } + + private async handleSocketMessage(messageData: unknown): Promise { + const bytes = await this.asUint8Array(messageData); + if (!bytes || bytes.length < 2) { + return; + } + + if (bytes[0] !== CONTROL_TAG_JSON) { + return; + } + + try { + const payload = JSON.parse(textDecoder.decode(bytes.subarray(1))) as TerminalInputControlMessage; + if (payload.t === 'po') { + return; + } + + if (payload.t === 'e') { + if (payload.c === 'NOT_BOUND' || payload.c === 'SESSION_NOT_FOUND') { + this.boundSessionId = null; + } + if (payload.f === true) { + this.handleSocketFailure(); + } + } + } catch { + this.handleSocketFailure(); + } + } + + private async asUint8Array(messageData: unknown): Promise { + if (messageData instanceof ArrayBuffer) { + return new Uint8Array(messageData); + } + + if (messageData instanceof Uint8Array) { + return messageData; + } + + if (typeof Blob !== 'undefined' && messageData instanceof Blob) { + const buffer = await messageData.arrayBuffer(); + return new Uint8Array(buffer); + } + + return null; + } + + private handleSocketFailure(): void { + this.boundSessionId = null; + this.resetConnection(); + this.scheduleReconnect(); + } + + private resetConnection(): void { + this.openPromise = null; + this.stopKeepalive(); + if (this.socket) { + const socket = this.socket; + this.socket = null; + socket.onopen = null; + socket.onmessage = null; + socket.onerror = null; + socket.onclose = null; + if (socket.readyState === WebSocket.OPEN || socket.readyState === WebSocket.CONNECTING) { + socket.close(); + } + } + this.boundSessionId = null; + } +} + +type TerminalInputWsGlobalState = { + capability: TerminalInputCapability | null; + manager: TerminalInputWsManager | null; +}; + +const getTerminalInputWsGlobalState = (): TerminalInputWsGlobalState => { + const globalScope = globalThis as typeof globalThis & { + [GLOBAL_TERMINAL_INPUT_STATE_KEY]?: TerminalInputWsGlobalState; + }; + + if (!globalScope[GLOBAL_TERMINAL_INPUT_STATE_KEY]) { + globalScope[GLOBAL_TERMINAL_INPUT_STATE_KEY] = { + capability: null, + manager: null, + }; + } + + return globalScope[GLOBAL_TERMINAL_INPUT_STATE_KEY]; +}; + +const applyTerminalInputCapability = (capability: TerminalInputCapability | undefined): void => { + const globalState = getTerminalInputWsGlobalState(); + globalState.capability = capability ?? null; + + if (!isWsInputSupported(globalState.capability)) { + globalState.manager?.close(); + globalState.manager = null; + return; + } + + const wsPath = globalState.capability?.ws?.path; + if (!wsPath) { + return; + } + + const socketUrl = normalizeWebSocketPath(wsPath); + if (!socketUrl) { + return; + } + + if (!globalState.manager) { + globalState.manager = new TerminalInputWsManager(); + } + + globalState.manager.configure(socketUrl); +}; + +const sendTerminalInputHttp = async (sessionId: string, data: string): Promise => { + const response = await fetch(`/api/terminal/${sessionId}/input`, { + method: 'POST', + headers: { 'Content-Type': 'text/plain' }, + body: data, + }); + + if (!response.ok) { + const error = await response.json().catch(() => ({ error: 'Failed to send input' })); + throw new Error(error.error || 'Failed to send terminal input'); + } +}; + export async function createTerminalSession( options: CreateTerminalOptions ): Promise { @@ -46,7 +507,9 @@ export async function createTerminalSession( throw new Error(error.error || 'Failed to create terminal session'); } - return response.json(); + const session = await response.json() as TerminalSession; + applyTerminalInputCapability(session.capabilities?.input); + return session; } export function connectTerminalStream( @@ -127,6 +590,7 @@ export function connectTerminalStream( const data = JSON.parse(event.data) as TerminalStreamEvent; if (data.type === 'exit') { + getTerminalInputWsGlobalState().manager?.unbindSession(sessionId); terminalExited = true; cleanup(); } @@ -192,16 +656,12 @@ export async function sendTerminalInput( sessionId: string, data: string ): Promise { - const response = await fetch(`/api/terminal/${sessionId}/input`, { - method: 'POST', - headers: { 'Content-Type': 'text/plain' }, - body: data, - }); - - if (!response.ok) { - const error = await response.json().catch(() => ({ error: 'Failed to send input' })); - throw new Error(error.error || 'Failed to send terminal input'); + const globalState = getTerminalInputWsGlobalState(); + if (globalState.manager && await globalState.manager.sendInput(sessionId, data)) { + return; } + + await sendTerminalInputHttp(sessionId, data); } export async function resizeTerminal( @@ -222,6 +682,8 @@ export async function resizeTerminal( } export async function closeTerminal(sessionId: string): Promise { + getTerminalInputWsGlobalState().manager?.unbindSession(sessionId); + const response = await fetch(`/api/terminal/${sessionId}`, { method: 'DELETE', }); @@ -236,6 +698,8 @@ export async function restartTerminalSession( currentSessionId: string, options: { cwd: string; cols?: number; rows?: number } ): Promise { + getTerminalInputWsGlobalState().manager?.unbindSession(currentSessionId); + const response = await fetch(`/api/terminal/${currentSessionId}/restart`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, @@ -251,7 +715,9 @@ export async function restartTerminalSession( throw new Error(error.error || 'Failed to restart terminal'); } - return response.json(); + const session = await response.json() as TerminalSession; + applyTerminalInputCapability(session.capabilities?.input); + return session; } export async function forceKillTerminal(options: { @@ -268,4 +734,51 @@ export async function forceKillTerminal(options: { const error = await response.json().catch(() => ({ error: 'Failed to force kill terminal' })); throw new Error(error.error || 'Failed to force kill terminal'); } + + if (options.sessionId) { + getTerminalInputWsGlobalState().manager?.unbindSession(options.sessionId); + } +} + +export function disposeTerminalInputTransport(): void { + const globalState = getTerminalInputWsGlobalState(); + globalState.manager?.close(); + globalState.manager = null; + globalState.capability = null; +} + +export function primeTerminalInputTransport(): void { + const globalState = getTerminalInputWsGlobalState(); + if (globalState.capability && !isWsInputSupported(globalState.capability)) { + return; + } + + const wsPath = globalState.capability?.ws?.path ?? DEFAULT_TERMINAL_INPUT_WS_PATH; + const socketUrl = normalizeWebSocketPath(wsPath); + if (!socketUrl) { + return; + } + + if (!globalState.manager) { + globalState.manager = new TerminalInputWsManager(); + } + + if (globalState.manager.isConnectedOrConnecting(socketUrl)) { + return; + } + + globalState.manager.configure(socketUrl); + globalState.manager.prime(); +} + +const hotModule = (import.meta as ImportMeta & { + hot?: { + dispose: (callback: () => void) => void; + }; +}).hot; + +if (hotModule) { + hotModule.dispose(() => { + disposeTerminalInputTransport(); + }); } diff --git a/packages/web/package.json b/packages/web/package.json index b6a8dc49..0da8e070 100644 --- a/packages/web/package.json +++ b/packages/web/package.json @@ -60,6 +60,7 @@ "strip-json-comments": "^5.0.3", "tailwind-merge": "^3.3.1", "web-push": "^3.6.7", + "ws": "^8.18.3", "yaml": "^2.8.1", "zustand": "^5.0.8" }, diff --git a/packages/web/server/TERMINAL_INPUT_WS_PROTOCOL.md b/packages/web/server/TERMINAL_INPUT_WS_PROTOCOL.md new file mode 100644 index 00000000..a518ca1a --- /dev/null +++ b/packages/web/server/TERMINAL_INPUT_WS_PROTOCOL.md @@ -0,0 +1,44 @@ +# Terminal Input WS Protocol + +## Goal +Reduce terminal input latency by replacing per-keystroke HTTP requests with a persistent WebSocket input channel, while keeping SSE output and HTTP endpoints as compatibility fallback. + +## Scope +- Input path: WebSocket (`/api/terminal/input-ws`) +- Output path: SSE (`/api/terminal/:sessionId/stream`) +- HTTP input fallback remains: `POST /api/terminal/:sessionId/input` + +## Framing +- Text frame: terminal keystroke payload (hot path) + - Examples: `"\r"`, `"\u001b[A"`, `"\u0003"` +- Binary frame: control envelope + - Byte 0: tag (`0x01` = JSON control) + - Bytes 1..N: UTF-8 JSON payload + +## Control Messages +- Bind active socket to terminal session: + - client -> server: `{"t":"b","s":"","v":1}` +- Keepalive ping: + - client -> server: `{"t":"p","v":1}` + - server -> client: `{"t":"po","v":1}` +- Server control responses: + - ready: `{"t":"ok","v":1}` + - bind ok: `{"t":"bok","v":1}` + - error: `{"t":"e","c":"","f":true|false}` + +## Multiplexing Model +- Single shared socket per client runtime. +- Socket has one mutable `boundSessionId`. +- Client sends bind control when active terminal changes. +- Keystroke frames apply to currently bound session. +- Client keeps socket open and sends periodic keepalive pings so the channel stays ready for next input. +- Client primes/opens this socket when the Terminal tab is opened (not per keystroke). + +## Security +- UI auth session required when UI password is enabled. +- Origin validation enforced for cookie-authenticated browser upgrades. +- Invalid/malformed frames are rate-limited and may close socket. + +## Fallback Behavior +- On WS unavailable/error/close, client falls back to HTTP input immediately. +- Existing terminal behavior remains functional during mixed-version rollout. diff --git a/packages/web/server/index.js b/packages/web/server/index.js index 72f0f660..a4334c79 100644 --- a/packages/web/server/index.js +++ b/packages/web/server/index.js @@ -4,11 +4,22 @@ import path from 'path'; import { spawn, spawnSync } from 'child_process'; import fs from 'fs'; import http from 'http'; +import { WebSocketServer } from 'ws'; import { fileURLToPath } from 'url'; import os from 'os'; import crypto from 'crypto'; import { createUiAuth } from './lib/ui-auth.js'; import { startCloudflareTunnel, printTunnelWarning, checkCloudflaredAvailable } from './lib/cloudflare-tunnel.js'; +import { + TERMINAL_INPUT_WS_MAX_PAYLOAD_BYTES, + TERMINAL_INPUT_WS_PATH, + createTerminalInputWsControlFrame, + isRebindRateLimited, + normalizeTerminalInputWsMessageToText, + parseRequestPathname, + pruneRebindTimestamps, + readTerminalInputWsControlFrame, +} from './lib/terminal-input-ws-protocol.js'; import { createOpencodeServer } from '@opencode-ai/sdk/server'; import webPush from 'web-push'; @@ -1437,6 +1448,96 @@ const getUiSessionTokenFromRequest = (req) => { return null; }; +const TERMINAL_INPUT_WS_MAX_REBINDS_PER_WINDOW = 128; +const TERMINAL_INPUT_WS_REBIND_WINDOW_MS = 60 * 1000; +const TERMINAL_INPUT_WS_HEARTBEAT_INTERVAL_MS = 15 * 1000; + +const rejectWebSocketUpgrade = (socket, statusCode, reason) => { + if (!socket || socket.destroyed) { + return; + } + + const message = typeof reason === 'string' && reason.trim().length > 0 ? reason.trim() : 'Bad Request'; + const body = Buffer.from(message, 'utf8'); + const statusText = { + 400: 'Bad Request', + 401: 'Unauthorized', + 403: 'Forbidden', + 404: 'Not Found', + 500: 'Internal Server Error', + }[statusCode] || 'Bad Request'; + + try { + socket.write( + `HTTP/1.1 ${statusCode} ${statusText}\r\n` + + 'Connection: close\r\n' + + 'Content-Type: text/plain; charset=utf-8\r\n' + + `Content-Length: ${body.length}\r\n\r\n` + ); + socket.write(body); + } catch { + } + + try { + socket.destroy(); + } catch { + } +}; + + +const getRequestOriginCandidates = async (req) => { + const origins = new Set(); + const forwardedProto = typeof req.headers['x-forwarded-proto'] === 'string' + ? req.headers['x-forwarded-proto'].split(',')[0].trim().toLowerCase() + : ''; + const protocol = forwardedProto || (req.socket?.encrypted ? 'https' : 'http'); + + const forwardedHost = typeof req.headers['x-forwarded-host'] === 'string' + ? req.headers['x-forwarded-host'].split(',')[0].trim() + : ''; + const host = forwardedHost || (typeof req.headers.host === 'string' ? req.headers.host.trim() : ''); + + if (host) { + origins.add(`${protocol}://${host}`); + const [hostname, port] = host.split(':'); + const normalizedHost = typeof hostname === 'string' ? hostname.toLowerCase() : ''; + const portSuffix = typeof port === 'string' && port.length > 0 ? `:${port}` : ''; + if (normalizedHost === 'localhost') { + origins.add(`${protocol}://127.0.0.1${portSuffix}`); + origins.add(`${protocol}://[::1]${portSuffix}`); + } else if (normalizedHost === '127.0.0.1' || normalizedHost === '[::1]') { + origins.add(`${protocol}://localhost${portSuffix}`); + } + } + + try { + const settings = await readSettingsFromDiskMigrated(); + if (typeof settings?.publicOrigin === 'string' && settings.publicOrigin.trim().length > 0) { + origins.add(new URL(settings.publicOrigin.trim()).origin); + } + } catch { + } + + return origins; +}; + +const isRequestOriginAllowed = async (req) => { + const originHeader = typeof req.headers.origin === 'string' ? req.headers.origin.trim() : ''; + if (!originHeader) { + return false; + } + + let normalizedOrigin = ''; + try { + normalizedOrigin = new URL(originHeader).origin; + } catch { + return false; + } + + const allowedOrigins = await getRequestOriginCandidates(req); + return allowedOrigins.has(normalizedOrigin); +}; + const normalizePushSubscriptions = (record) => { if (!Array.isArray(record)) return []; return record @@ -2028,6 +2129,7 @@ let openCodeNotReadySince = 0; let exitOnShutdown = true; let uiAuthController = null; let cloudflareTunnelController = null; +let terminalInputWsServer = null; // Sync helper - call after modifying any HMR state variable const syncToHmrState = () => { @@ -3855,7 +3957,7 @@ function setupProxy(app) { return suffix; }, - ws: true, + ws: false, onError: (err, req, res) => { console.error(`Proxy error: ${err.message}`); if (!res.headersSent) { @@ -3930,6 +4032,24 @@ async function gracefulShutdown(options = {}) { clearInterval(healthCheckInterval); } + if (terminalInputWsServer) { + try { + for (const client of terminalInputWsServer.clients) { + try { + client.terminate(); + } catch { + } + } + + await new Promise((resolve) => { + terminalInputWsServer.close(() => resolve()); + }); + } catch { + } finally { + terminalInputWsServer = null; + } + } + // Only stop OpenCode if we started it ourselves (not when using external server) if (!ENV_SKIP_OPENCODE_START) { const portToKill = openCodePort; @@ -8414,6 +8534,192 @@ Context: const terminalSessions = new Map(); const MAX_TERMINAL_SESSIONS = 20; const TERMINAL_IDLE_TIMEOUT = 30 * 60 * 1000; + const terminalInputCapabilities = { + input: { + preferred: 'ws', + transports: ['http', 'ws'], + ws: { + path: TERMINAL_INPUT_WS_PATH, + v: 1, + enc: 'text+json-bin-control', + }, + }, + }; + + const sendTerminalInputWsControl = (socket, payload) => { + if (!socket || socket.readyState !== 1) { + return; + } + + try { + socket.send(createTerminalInputWsControlFrame(payload), { binary: true }); + } catch { + } + }; + + terminalInputWsServer = new WebSocketServer({ + noServer: true, + maxPayload: TERMINAL_INPUT_WS_MAX_PAYLOAD_BYTES, + }); + + terminalInputWsServer.on('connection', (socket) => { + const connectionState = { + boundSessionId: null, + invalidFrames: 0, + rebindTimestamps: [], + lastActivityAt: Date.now(), + }; + + sendTerminalInputWsControl(socket, { t: 'ok', v: 1 }); + + const heartbeatInterval = setInterval(() => { + if (socket.readyState !== 1) { + return; + } + + try { + socket.ping(); + } catch { + } + }, TERMINAL_INPUT_WS_HEARTBEAT_INTERVAL_MS); + + socket.on('pong', () => { + connectionState.lastActivityAt = Date.now(); + }); + + socket.on('message', (message, isBinary) => { + connectionState.lastActivityAt = Date.now(); + + if (isBinary) { + const controlMessage = readTerminalInputWsControlFrame(message); + if (!controlMessage || typeof controlMessage.t !== 'string') { + connectionState.invalidFrames += 1; + sendTerminalInputWsControl(socket, { + t: 'e', + c: 'BAD_FRAME', + f: connectionState.invalidFrames >= 10, + }); + if (connectionState.invalidFrames >= 10) { + socket.close(1008, 'protocol violation'); + } + return; + } + + if (controlMessage.t === 'p') { + sendTerminalInputWsControl(socket, { t: 'po', v: 1 }); + return; + } + + if (controlMessage.t !== 'b' || typeof controlMessage.s !== 'string') { + connectionState.invalidFrames += 1; + sendTerminalInputWsControl(socket, { + t: 'e', + c: 'BAD_FRAME', + f: connectionState.invalidFrames >= 10, + }); + if (connectionState.invalidFrames >= 10) { + socket.close(1008, 'protocol violation'); + } + return; + } + + const now = Date.now(); + connectionState.rebindTimestamps = pruneRebindTimestamps( + connectionState.rebindTimestamps, + now, + TERMINAL_INPUT_WS_REBIND_WINDOW_MS + ); + + if (isRebindRateLimited(connectionState.rebindTimestamps, TERMINAL_INPUT_WS_MAX_REBINDS_PER_WINDOW)) { + sendTerminalInputWsControl(socket, { t: 'e', c: 'RATE_LIMIT', f: false }); + return; + } + + const nextSessionId = controlMessage.s.trim(); + const targetSession = terminalSessions.get(nextSessionId); + if (!targetSession) { + connectionState.boundSessionId = null; + sendTerminalInputWsControl(socket, { t: 'e', c: 'SESSION_NOT_FOUND', f: false }); + return; + } + + connectionState.rebindTimestamps.push(now); + connectionState.boundSessionId = nextSessionId; + sendTerminalInputWsControl(socket, { t: 'bok', v: 1 }); + return; + } + + const payload = normalizeTerminalInputWsMessageToText(message); + if (payload.length === 0) { + return; + } + + if (!connectionState.boundSessionId) { + sendTerminalInputWsControl(socket, { t: 'e', c: 'NOT_BOUND', f: false }); + return; + } + + const session = terminalSessions.get(connectionState.boundSessionId); + if (!session) { + connectionState.boundSessionId = null; + sendTerminalInputWsControl(socket, { t: 'e', c: 'SESSION_NOT_FOUND', f: false }); + return; + } + + try { + session.ptyProcess.write(payload); + session.lastActivity = Date.now(); + } catch { + sendTerminalInputWsControl(socket, { t: 'e', c: 'WRITE_FAIL', f: false }); + } + }); + + socket.on('close', () => { + clearInterval(heartbeatInterval); + }); + + socket.on('error', (error) => { + void error; + }); + }); + + server.on('upgrade', (req, socket, head) => { + const pathname = parseRequestPathname(req.url); + if (pathname !== TERMINAL_INPUT_WS_PATH) { + return; + } + + const handleUpgrade = async () => { + try { + if (uiAuthController?.enabled) { + const sessionToken = uiAuthController?.ensureSessionToken?.(req, null); + if (!sessionToken) { + rejectWebSocketUpgrade(socket, 401, 'UI authentication required'); + return; + } + + const originAllowed = await isRequestOriginAllowed(req); + if (!originAllowed) { + rejectWebSocketUpgrade(socket, 403, 'Invalid origin'); + return; + } + } + + if (!terminalInputWsServer) { + rejectWebSocketUpgrade(socket, 500, 'Terminal WebSocket unavailable'); + return; + } + + terminalInputWsServer.handleUpgrade(req, socket, head, (ws) => { + terminalInputWsServer.emit('connection', ws, req); + }); + } catch { + rejectWebSocketUpgrade(socket, 500, 'Upgrade failed'); + } + }; + + void handleUpgrade(); + }); setInterval(() => { const now = Date.now(); @@ -8484,7 +8790,7 @@ Context: }); console.log(`Created terminal session: ${sessionId} in ${cwd}`); - res.json({ sessionId, cols: cols || 80, rows: rows || 24 }); + res.json({ sessionId, cols: cols || 80, rows: rows || 24, capabilities: terminalInputCapabilities }); } catch (error) { console.error('Failed to create terminal session:', error); res.status(500).json({ error: error.message || 'Failed to create terminal session' }); @@ -8705,7 +9011,7 @@ Context: }); console.log(`Restarted terminal session: ${sessionId} -> ${newSessionId} in ${cwd}`); - res.json({ sessionId: newSessionId, cols: cols || 80, rows: rows || 24 }); + res.json({ sessionId: newSessionId, cols: cols || 80, rows: rows || 24, capabilities: terminalInputCapabilities }); } catch (error) { console.error('Failed to restart terminal session:', error); res.status(500).json({ error: error.message || 'Failed to restart terminal session' }); diff --git a/packages/web/server/lib/terminal-input-ws-protocol.js b/packages/web/server/lib/terminal-input-ws-protocol.js new file mode 100644 index 00000000..cdf69699 --- /dev/null +++ b/packages/web/server/lib/terminal-input-ws-protocol.js @@ -0,0 +1,66 @@ +export const TERMINAL_INPUT_WS_PATH = '/api/terminal/input-ws'; +export const TERMINAL_INPUT_WS_CONTROL_TAG_JSON = 0x01; +export const TERMINAL_INPUT_WS_MAX_PAYLOAD_BYTES = 64 * 1024; + +export const parseRequestPathname = (requestUrl) => { + if (typeof requestUrl !== 'string' || requestUrl.length === 0) { + return ''; + } + + try { + return new URL(requestUrl, 'http://localhost').pathname; + } catch { + return ''; + } +}; + +export const normalizeTerminalInputWsMessageToBuffer = (rawData) => { + if (Buffer.isBuffer(rawData)) { + return rawData; + } + + if (Array.isArray(rawData)) { + return Buffer.concat(rawData.map((chunk) => (Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)))); + } + + return Buffer.from(rawData); +}; + +export const normalizeTerminalInputWsMessageToText = (rawData) => { + if (typeof rawData === 'string') { + return rawData; + } + + return normalizeTerminalInputWsMessageToBuffer(rawData).toString('utf8'); +}; + +export const readTerminalInputWsControlFrame = (rawData) => { + if (!rawData) { + return null; + } + + const buffer = normalizeTerminalInputWsMessageToBuffer(rawData); + if (buffer.length < 2 || buffer[0] !== TERMINAL_INPUT_WS_CONTROL_TAG_JSON) { + return null; + } + + try { + const parsed = JSON.parse(buffer.subarray(1).toString('utf8')); + if (!parsed || typeof parsed !== 'object') { + return null; + } + return parsed; + } catch { + return null; + } +}; + +export const createTerminalInputWsControlFrame = (payload) => { + const jsonBytes = Buffer.from(JSON.stringify(payload), 'utf8'); + return Buffer.concat([Buffer.from([TERMINAL_INPUT_WS_CONTROL_TAG_JSON]), jsonBytes]); +}; + +export const pruneRebindTimestamps = (timestamps, now, windowMs) => + timestamps.filter((timestamp) => now - timestamp < windowMs); + +export const isRebindRateLimited = (timestamps, maxPerWindow) => timestamps.length >= maxPerWindow; diff --git a/packages/web/server/lib/terminal-input-ws-protocol.test.js b/packages/web/server/lib/terminal-input-ws-protocol.test.js new file mode 100644 index 00000000..3a3d4ef6 --- /dev/null +++ b/packages/web/server/lib/terminal-input-ws-protocol.test.js @@ -0,0 +1,138 @@ +import { describe, expect, it } from 'bun:test'; + +import { + TERMINAL_INPUT_WS_CONTROL_TAG_JSON, + TERMINAL_INPUT_WS_PATH, + createTerminalInputWsControlFrame, + isRebindRateLimited, + normalizeTerminalInputWsMessageToBuffer, + normalizeTerminalInputWsMessageToText, + parseRequestPathname, + pruneRebindTimestamps, + readTerminalInputWsControlFrame, +} from './terminal-input-ws-protocol.js'; + +describe('terminal input websocket protocol', () => { + it('uses fixed websocket path', () => { + expect(TERMINAL_INPUT_WS_PATH).toBe('/api/terminal/input-ws'); + }); + + it('encodes control frames with control tag prefix', () => { + const frame = createTerminalInputWsControlFrame({ t: 'ok', v: 1 }); + expect(frame[0]).toBe(TERMINAL_INPUT_WS_CONTROL_TAG_JSON); + }); + + it('roundtrips control frame payload', () => { + const payload = { t: 'b', s: 'abc123', v: 1 }; + const frame = createTerminalInputWsControlFrame(payload); + expect(readTerminalInputWsControlFrame(frame)).toEqual(payload); + }); + + it('rejects control frame without protocol tag', () => { + const frame = Buffer.from(JSON.stringify({ t: 'b', s: 'abc123' }), 'utf8'); + expect(readTerminalInputWsControlFrame(frame)).toBeNull(); + }); + + it('rejects malformed control json', () => { + const frame = Buffer.concat([ + Buffer.from([TERMINAL_INPUT_WS_CONTROL_TAG_JSON]), + Buffer.from('{not json', 'utf8'), + ]); + expect(readTerminalInputWsControlFrame(frame)).toBeNull(); + }); + + it('rejects empty control payloads', () => { + expect(readTerminalInputWsControlFrame(null)).toBeNull(); + expect(readTerminalInputWsControlFrame(undefined)).toBeNull(); + expect(readTerminalInputWsControlFrame(Buffer.alloc(0))).toBeNull(); + }); + + it('rejects control json that is not object', () => { + const frame = Buffer.concat([ + Buffer.from([TERMINAL_INPUT_WS_CONTROL_TAG_JSON]), + Buffer.from('"str"', 'utf8'), + ]); + expect(readTerminalInputWsControlFrame(frame)).toBeNull(); + }); + + it('parses control frame from chunk arrays', () => { + const frame = createTerminalInputWsControlFrame({ t: 'bok', v: 1 }); + const chunks = [frame.subarray(0, 2), frame.subarray(2)]; + expect(readTerminalInputWsControlFrame(chunks)).toEqual({ t: 'bok', v: 1 }); + }); + + it('normalizes buffer passthrough', () => { + const raw = Buffer.from('abc', 'utf8'); + const normalized = normalizeTerminalInputWsMessageToBuffer(raw); + expect(normalized).toBe(raw); + expect(normalized.toString('utf8')).toBe('abc'); + }); + + it('normalizes uint8 arrays', () => { + const normalized = normalizeTerminalInputWsMessageToBuffer(new Uint8Array([97, 98, 99])); + expect(normalized.toString('utf8')).toBe('abc'); + }); + + it('normalizes array buffer payloads', () => { + const source = new Uint8Array([97, 98, 99]).buffer; + const normalized = normalizeTerminalInputWsMessageToBuffer(source); + expect(normalized.toString('utf8')).toBe('abc'); + }); + + it('normalizes chunk array payloads', () => { + const normalized = normalizeTerminalInputWsMessageToBuffer([ + Buffer.from('ab', 'utf8'), + Buffer.from('c', 'utf8'), + ]); + expect(normalized.toString('utf8')).toBe('abc'); + }); + + it('normalizes text payload from string', () => { + expect(normalizeTerminalInputWsMessageToText('\u001b[A')).toBe('\u001b[A'); + }); + + it('normalizes text payload from binary data', () => { + expect(normalizeTerminalInputWsMessageToText(Buffer.from('\r', 'utf8'))).toBe('\r'); + }); + + it('parses relative request pathname', () => { + expect(parseRequestPathname('/api/terminal/input-ws?x=1')).toBe('/api/terminal/input-ws'); + }); + + it('parses absolute request pathname', () => { + expect(parseRequestPathname('http://localhost:3000/api/terminal/input-ws')).toBe('/api/terminal/input-ws'); + }); + + it('returns empty pathname for non-string request url', () => { + expect(parseRequestPathname(null)).toBe(''); + }); + + it('returns empty pathname for invalid request url', () => { + expect(parseRequestPathname('http://')).toBe(''); + expect(parseRequestPathname('')).toBe(''); + }); + + it('prunes stale rebind timestamps', () => { + const now = 1_000; + const pruned = pruneRebindTimestamps([100, 200, 950, 999], now, 100); + expect(pruned).toEqual([950, 999]); + }); + + it('keeps rebind timestamps within active window', () => { + const now = 1_000; + const pruned = pruneRebindTimestamps([920, 950, 999], now, 100); + expect(pruned).toEqual([920, 950, 999]); + }); + + it('does not rate limit below threshold', () => { + expect(isRebindRateLimited([1, 2, 3], 4)).toBe(false); + }); + + it('does not rate limit empty window', () => { + expect(isRebindRateLimited([], 1)).toBe(false); + }); + + it('rate limits at threshold', () => { + expect(isRebindRateLimited([1, 2, 3, 4], 4)).toBe(true); + }); +});