diff --git a/packages/client/src/daemon-client-relay-e2ee-transport.ts b/packages/client/src/daemon-client-relay-e2ee-transport.ts index 1bc175bf1b..e25779a7bb 100644 --- a/packages/client/src/daemon-client-relay-e2ee-transport.ts +++ b/packages/client/src/daemon-client-relay-e2ee-transport.ts @@ -1,3 +1,4 @@ +import { MAX_RELAY_WEBSOCKET_MESSAGE_BYTES } from "@getpaseo/protocol/websocket-message-limit"; import { createClientChannel, type EncryptedChannel, @@ -8,7 +9,11 @@ import type { DaemonTransportFactory, TransportLogger, } from "./daemon-client-transport-types.js"; -import { extractRelayMessage, normalizeTransportPayload } from "./daemon-client-transport-utils.js"; +import { + extractRelayMessage, + normalizeTransportPayload, + messageTooLargeError, +} from "./daemon-client-transport-utils.js"; type OpenHandler = () => void; type CloseHandler = (event?: unknown) => void; @@ -137,7 +142,13 @@ export function createEncryptedTransport( if (!channel) { throw new Error("Encrypted channel not ready"); } - void channel.send(normalizeTransportPayload(data)).catch((error) => { + const payload = normalizeTransportPayload(data); + // Reject synchronously so the requesting operation fails without closing the transport. + const wireBytes = channel.outboundWireByteLength(payload); + if (wireBytes > MAX_RELAY_WEBSOCKET_MESSAGE_BYTES) { + throw messageTooLargeError(wireBytes, MAX_RELAY_WEBSOCKET_MESSAGE_BYTES); + } + void channel.send(payload).catch((error) => { emitError(error); }); }, diff --git a/packages/client/src/daemon-client-transport-utils.ts b/packages/client/src/daemon-client-transport-utils.ts index 3b39cf5114..cb767c1d87 100644 --- a/packages/client/src/daemon-client-transport-utils.ts +++ b/packages/client/src/daemon-client-transport-utils.ts @@ -133,3 +133,10 @@ export function encodeUtf8String(value: string): Uint8Array { } return out; } + +export function messageTooLargeError(size: number, limit: number): Error { + const toMiB = (bytes: number) => (bytes / (1024 * 1024)).toFixed(1); + return new Error( + `Message is too large to send (${toMiB(size)} MiB). The limit is ${toMiB(limit)} MiB, so send fewer or smaller attachments.`, + ); +} diff --git a/packages/client/src/daemon-client.test.ts b/packages/client/src/daemon-client.test.ts index 95921a96a0..f62f46b733 100644 --- a/packages/client/src/daemon-client.test.ts +++ b/packages/client/src/daemon-client.test.ts @@ -7527,3 +7527,29 @@ test.each([false, true])( await send; }, ); + +test("measures an outbound message in UTF-8 bytes against the daemon's message limit", async () => { + const mock = createMockTransport(); + const client = new DaemonClient({ + url: "ws://test", + clientId: "message_limit_unit_test", + transportFactory: () => mock.transport, + reconnect: { enabled: false }, + }); + clients.push(client); + const connectPromise = client.connect(); + mock.triggerOpen(); + await connectPromise; + const mib = 1024 * 1024; + + // "é" is one UTF-16 unit but two UTF-8 bytes, so this text is ~98 MiB on the wire. + void client.sendMessage("agent-1", "é".repeat(49 * mib)).catch(() => undefined); + expect(mock.sent).toHaveLength(1); + + // ~60M characters is under the 100 MiB limit; ~120 MiB of UTF-8 is over it. + await expect(client.sendMessage("agent-1", "é".repeat(60 * mib))).rejects.toThrow( + "Message is too large to send (120.0 MiB). The limit is 100.0 MiB", + ); + expect(mock.sent).toHaveLength(1); + expect(client.getConnectionState().status).toBe("connected"); +}); diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index f900fe41e5..2bd19f2974 100644 --- a/packages/client/src/daemon-client.ts +++ b/packages/client/src/daemon-client.ts @@ -1,3 +1,4 @@ +import { messageTooLargeError } from "./daemon-client-transport-utils.js"; import { legacyUsageIcon } from "./legacy-usage-icons.js"; import { subscribeTimeline, type TimelineMessage } from "./timeline-subscription/index.js"; import { ProviderSnapshotUpdates } from "./provider-snapshots/index.js"; @@ -13,6 +14,7 @@ import type { z } from "zod"; import type { SessionEventSubscription } from "@getpaseo/protocol/messages"; import type { ClientCapability } from "@getpaseo/protocol/client-capabilities"; import type { AgentAttentionNotificationPayload } from "@getpaseo/protocol/agent-attention-notification"; +import { MAX_WEBSOCKET_MESSAGE_BYTES } from "@getpaseo/protocol/websocket-message-limit"; import { AgentCreateFailedStatusPayloadSchema, AgentCreatedStatusPayloadSchema, @@ -1132,6 +1134,19 @@ function getTransportFrameSize(frame: string | Uint8Array | ArrayBuffer): number return frame.byteLength; } +function assertTransportFrameFits(frame: string | Uint8Array | ArrayBuffer): void { + if (typeof frame !== "string") { + if (frame.byteLength > MAX_WEBSOCKET_MESSAGE_BYTES) + throw messageTooLargeError(frame.byteLength, MAX_WEBSOCKET_MESSAGE_BYTES); + return; + } + // UTF-8 needs at most three bytes per UTF-16 unit, so only a string near the limit is measured. + if (frame.length * 3 <= MAX_WEBSOCKET_MESSAGE_BYTES) return; + const size = new TextEncoder().encode(frame).byteLength; + if (size > MAX_WEBSOCKET_MESSAGE_BYTES) + throw messageTooLargeError(size, MAX_WEBSOCKET_MESSAGE_BYTES); +} + function describeInboundTransportFrame( frame: unknown, rawBytes: Uint8Array | null, @@ -1767,6 +1782,8 @@ export class DaemonClient { if (!this.transport) { throw new DaemonConnectionError("Transport not connected"); } + // The daemon closes the connection on an oversized message, and creations resubmit on reconnect. + assertTransportFrameFits(frame); const isOpen = this.beginTraceSection("paseo.ws.frame.outbound", { kind: typeof frame === "string" ? "text" : "binary", size: String(getTransportFrameSize(frame)), diff --git a/packages/protocol/src/websocket-message-limit.ts b/packages/protocol/src/websocket-message-limit.ts new file mode 100644 index 0000000000..9e2bb48e21 --- /dev/null +++ b/packages/protocol/src/websocket-message-limit.ts @@ -0,0 +1,5 @@ +/** Largest WebSocket message the daemon accepts. A bigger one closes the connection with 1009. */ +export const MAX_WEBSOCKET_MESSAGE_BYTES = 100 * 1024 * 1024; + +/** Largest encrypted frame the relay accepts. */ +export const MAX_RELAY_WEBSOCKET_MESSAGE_BYTES = 32 * 1024 * 1024; diff --git a/packages/server/package.json b/packages/server/package.json index b80941d56f..541cf5eba1 100644 --- a/packages/server/package.json +++ b/packages/server/package.json @@ -89,7 +89,7 @@ "speech:transcribe:local": "tsx scripts/transcribe-local-wav.ts", "test": "npm run test:unit && npm run test:integration", "test:unit": "vitest run --fileParallelism --exclude \"**/*.e2e.test.ts\"", - "test:integration": "vitest run --maxWorkers=1 src/server/daemon-e2e/filesystem.e2e.test.ts src/server/daemon-e2e/file-download.e2e.test.ts src/server/daemon-e2e/cli-logs-follow.e2e.test.ts src/server/creation/creation.e2e.test.ts src/server/daemon-e2e/terminal-workspace-sdk.e2e.test.ts src/server/daemon-e2e/models.e2e.test.ts src/server/daemon-e2e/live-preferences.e2e.test.ts src/server/agent/model-catalog.e2e.test.ts src/server/plugins/plugin-paseo-api.e2e.test.ts && vitest run --maxWorkers=1 src/server/daemon-e2e/git-operations.e2e.test.ts -t \"merges into\" && vitest run --maxWorkers=1 src/server/session.create-agent-worktree-autoarchive.e2e.test.ts -t \"creates a worktree and auto-archives both\" && vitest run --maxWorkers=1 src/server/agent/providers/opencode-bridge.local.e2e.test.ts -t \"versioned runtime\"", + "test:integration": "vitest run --maxWorkers=1 src/server/daemon-e2e/filesystem.e2e.test.ts src/server/daemon-e2e/file-download.e2e.test.ts src/server/daemon-e2e/cli-logs-follow.e2e.test.ts src/server/creation/creation.e2e.test.ts src/server/daemon-e2e/terminal-workspace-sdk.e2e.test.ts src/server/daemon-e2e/models.e2e.test.ts src/server/daemon-e2e/live-preferences.e2e.test.ts src/server/daemon-e2e/oversized-message.e2e.test.ts src/server/agent/model-catalog.e2e.test.ts src/server/plugins/plugin-paseo-api.e2e.test.ts && vitest run --maxWorkers=1 src/server/daemon-e2e/git-operations.e2e.test.ts -t \"merges into\" && vitest run --maxWorkers=1 src/server/session.create-agent-worktree-autoarchive.e2e.test.ts -t \"creates a worktree and auto-archives both\" && vitest run --maxWorkers=1 src/server/agent/providers/opencode-bridge.local.e2e.test.ts -t \"versioned runtime\" && vitest run --maxWorkers=1 src/server/daemon-e2e/relay-transport.e2e.test.ts -t \"oversized attachments\"", "test:hub-cli-contract": "vitest run --maxWorkers=1 src/server/hub/hub-cli-contract.test.ts", "test:integration:all": "npm run test:e2e", "test:integration:real": "vitest run real.e2e.test.ts", diff --git a/packages/server/src/server/daemon-e2e/oversized-message.e2e.test.ts b/packages/server/src/server/daemon-e2e/oversized-message.e2e.test.ts new file mode 100644 index 0000000000..fc68fe44f4 --- /dev/null +++ b/packages/server/src/server/daemon-e2e/oversized-message.e2e.test.ts @@ -0,0 +1,58 @@ +import { describe, test, expect, beforeEach, afterEach } from "vitest"; +import { mkdtempSync, rmSync } from "fs"; +import { tmpdir } from "os"; +import path from "path"; +import { createDaemonTestContext, type DaemonTestContext } from "../test-utils/index.js"; + +let ctx: DaemonTestContext; +let cwd: string; + +beforeEach(async () => { + ctx = await createDaemonTestContext(); + cwd = mkdtempSync(path.join(tmpdir(), "daemon-e2e-")); +}); + +afterEach(async () => { + await ctx.cleanup(); + rmSync(cwd, { recursive: true, force: true }); +}, 60000); + +describe("a message larger than the daemon accepts", () => { + const oversizedImage = { data: "A".repeat(101 * 1024 * 1024), mimeType: "image/png" }; + + test("fails the creation and keeps the connection up", async () => { + const statuses: string[] = []; + const unsubscribe = ctx.client.subscribeConnectionStatus((state) => { + statuses.push(state.status); + }); + + await expect( + ctx.client.createAgent({ + provider: "claude", + cwd, + title: "Oversized first message", + initialPrompt: "Describe these screens.", + images: [oversizedImage], + }), + ).rejects.toThrow(/too large/); + unsubscribe(); + + expect(statuses).toEqual(["connected"]); + expect((await ctx.client.fetchAgents()).entries).toEqual([]); + }, 30000); + + test("fails the send and keeps the connection up", async () => { + const agent = await ctx.client.createAgent({ provider: "claude", cwd, title: "Agent" }); + const statuses: string[] = []; + const unsubscribe = ctx.client.subscribeConnectionStatus((state) => { + statuses.push(state.status); + }); + + await expect( + ctx.client.sendMessage(agent.id, "Describe these screens.", { images: [oversizedImage] }), + ).rejects.toThrow(/too large/); + unsubscribe(); + + expect(statuses).toEqual(["connected"]); + }, 30000); +}); diff --git a/packages/server/src/server/daemon-e2e/relay-transport.e2e.test.ts b/packages/server/src/server/daemon-e2e/relay-transport.e2e.test.ts index 3edede83d3..d204842e0c 100644 --- a/packages/server/src/server/daemon-e2e/relay-transport.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/relay-transport.e2e.test.ts @@ -1,4 +1,6 @@ import { describe, expect, test } from "vitest"; +import { withTimeout } from "../../utils/promise-timeout.js"; +import { DaemonClient } from "../test-utils/daemon-client.js"; import { WebSocket } from "ws"; import pino from "pino"; import { Writable } from "node:stream"; @@ -202,12 +204,20 @@ async function waitForCapturedLog( if (options.useLocalRelay) { relayArgs.push("--var", "PASEO_RELAY_UPSTREAM:"); } - relayProcess = spawn("npx", relayArgs, { - cwd: relayDir, - env: { ...process.env }, - stdio: ["ignore", "pipe", "pipe"], - detached: false, - }); + relayProcess = spawn( + process.execPath, + [ + "--experimental-vm-modules", + path.resolve(relayDir, "../../node_modules/wrangler/wrangler-dist/cli.js"), + ...relayArgs.slice(1), + ], + { + cwd: relayDir, + env: { ...process.env }, + stdio: ["ignore", "pipe", "pipe"], + detached: false, + }, + ); relayProcess.stdout?.on("data", (data: Buffer) => { const lines = data @@ -241,6 +251,73 @@ async function waitForCapturedLog( relayProcess = null; }; + test("oversized attachments fail without disconnecting or resubmitting over the relay", async () => { + await startRelay({ useLocalRelay: true }); + const daemon = await createTestPaseoDaemon({ + relayEnabled: true, + relayEndpoint: `127.0.0.1:${relayPort}`, + relayUseTls: false, + relayPublicUseTls: false, + }); + let client: DaemonClient | null = null; + try { + const offerUrl = await getPairingOfferUrl({ + paseoHome: daemon.paseoHome, + relayEnabled: true, + relayEndpoint: daemon.config.relayEndpoint, + }); + const { serverId, daemonPublicKeyB64 } = decodeOfferFromFragmentUrl(offerUrl); + client = new DaemonClient({ + url: buildRelayWebSocketUrl({ + endpoint: `127.0.0.1:${relayPort}`, + useTls: false, + serverId, + role: "client", + }), + e2ee: { enabled: true, daemonPublicKeyB64 }, + reconnect: { baseDelayMs: 100, maxDelayMs: 100 }, + }); + await client.connect(); + const statuses: string[] = []; + client.subscribeConnectionStatus((state) => statuses.push(state.status)); + for (const imageMiB of [30, 61]) { + // Attachments are base64 in the JSON, then E2EE base64 expands the frame again. + const images = [ + { data: "A".repeat(4 * Math.ceil((imageMiB * 1024 * 1024) / 3)), mimeType: "image/png" }, + ]; + await expect( + withTimeout( + client.createWorkspace({ + source: { kind: "directory", path: daemon.staticDir }, + agent: { + provider: "claude", + cwd: daemon.staticDir, + initialPrompt: "Describe these screens.", + images, + }, + }), + 10000, + "Workspace creation did not settle", + ), + ).rejects.toThrow(/too large.*32\.0 MiB/); + expect((await client.fetchAgents()).entries).toEqual([]); + } + const agent = await client.createAgent({ provider: "claude", cwd: daemon.staticDir }); + await expect( + client.sendMessage(agent.id, "Describe these screens.", { + images: [{ data: "A".repeat(40 * 1024 * 1024), mimeType: "image/png" }], + }), + ).rejects.toThrow(/too large.*32\.0 MiB/); + await client.sendMessage(agent.id, "A small follow-up still works."); + expect((await client.fetchAgents()).entries).toHaveLength(1); + expect(statuses).toEqual(["connected"]); + } finally { + await client?.close(); + await daemon.close(); + await stopRelay(); + } + }, 60000); + test("daemon connects to relay and client ping/pong works through relay", async () => { process.env.PASEO_PRIMARY_LAN_IP = "192.168.1.12"; diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index 219e04b0c2..5e93681a66 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -33,6 +33,7 @@ import { } from "./messages.js"; import { asUint8Array, decodeBinaryFrame } from "@getpaseo/protocol/binary-frames/index"; import type { TerminalActivity } from "@getpaseo/protocol/terminal-activity"; +import { MAX_WEBSOCKET_MESSAGE_BYTES } from "@getpaseo/protocol/websocket-message-limit"; import type { HostnamesConfig } from "./hostnames.js"; import { isHostnameAllowed } from "./hostnames.js"; import { @@ -822,6 +823,7 @@ export class VoiceAssistantWebSocketServer { const wss = new WebSocketServer({ server, path: "/ws", + maxPayload: MAX_WEBSOCKET_MESSAGE_BYTES, handleProtocols: (protocols) => selectWebSocketProtocol(protocols, password), verifyClient: ({ req }, callback) => { this.verifyWsUpgrade( diff --git a/packages/server/src/utils/checkout-git.ts b/packages/server/src/utils/checkout-git.ts index caa7e65cb0..7a53c753b9 100644 --- a/packages/server/src/utils/checkout-git.ts +++ b/packages/server/src/utils/checkout-git.ts @@ -1,3 +1,4 @@ +import { MAX_RELAY_WEBSOCKET_MESSAGE_BYTES } from "@getpaseo/protocol/websocket-message-limit"; import { resolve, dirname, basename } from "path"; import { existsSync, realpathSync } from "fs"; import { open as openFile, readFile, stat as statFile } from "fs/promises"; @@ -2152,13 +2153,13 @@ export async function getCheckoutSnapshotFacts( const PER_FILE_DIFF_MAX_BYTES = 1024 * 1024; // 1MB const TOTAL_DIFF_MAX_BYTES = 2 * 1024 * 1024; // 2MB -const RELAY_MAX_FRAME_BYTES = 32 * 1024 * 1024; const CHECKOUT_DIFF_FRAME_HEADROOM_BYTES = 1024 * 1024; // Temporary until diffs load lazily per file. The Paseo relay's 32 MiB frame limit is // binding: string frames are encrypted and base64-encoded. Reserve 1 MiB plaintext for // the surrounding WebSocket JSON envelope after inverting that exact wire expansion. export const CHECKOUT_DIFF_MAX_STRUCTURED_BYTES = - maxBase64EncryptedPlaintextByteLength(RELAY_MAX_FRAME_BYTES) - CHECKOUT_DIFF_FRAME_HEADROOM_BYTES; + maxBase64EncryptedPlaintextByteLength(MAX_RELAY_WEBSOCKET_MESSAGE_BYTES) - + CHECKOUT_DIFF_FRAME_HEADROOM_BYTES; interface StructuredDiffAccumulator { files: ParsedDiffFile[];