From e93d78b46c1dc074086ff958bc1de3378701cbc7 Mon Sep 17 00:00:00 2001 From: Aron Prins Date: Sat, 20 Jun 2026 20:32:49 +0200 Subject: [PATCH] fix(server): harden live events upgrade sockets (#8383) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - The server exposes live event updates to board and agent clients over WebSocket upgrade requests > - WebSocket upgrade authorization can be asynchronous, leaving the raw HTTP upgrade socket in server-owned code before `ws` takes over > - If the client disconnects during that authorization window, the server can still try to reject or upgrade a closed socket > - A raw socket write after peer disconnect can emit `EPIPE` / `ECONNRESET`, and without a listener that can become process-fatal > - This pull request hardens the pre-`ws` upgrade socket path and adds regression coverage for disconnect/error races > - The benefit is that live event reconnect churn degrades gracefully instead of risking a server crash ## Linked Issues or Issue Description External public context: Refs https://github.com/aronprins/paperclip-desktop/issues/14 No matching open upstream issue or PR was found when searching `paperclipai/paperclip` for `EPIPE`, `rejectUpgrade`, `live-events-ws`, `websocket upgrade`, and related socket-write terms. ### Bug report **What happened?** The live events WebSocket upgrade handler could write a rejection response to the raw upgrade socket after the peer had already disconnected during async authorization. A transport-level `EPIPE` or similar socket error on that raw socket can terminate the server process if it is emitted without an error listener. **Expected behavior** The server should not write to destroyed/non-writable upgrade sockets, should tolerate raw socket errors while authorization is pending, and should not call `handleUpgrade()` after the socket is no longer writable. **Steps to reproduce** 1. Start a Paperclip server with live events enabled. 2. Open a raw WebSocket upgrade request to `/api/companies/:companyId/events/ws`. 3. Disconnect the client before async authorization resolves. 4. Let the server take the reject or upgrade path. 5. Observe that the pre-fix path can still write to or upgrade a closed socket. **Paperclip version or commit** Reproduced by static inspection on `master` before this PR at `950484d20`. **Deployment mode** Any mode using live event WebSocket upgrades. The report was originally observed from local Desktop embedding, but the vulnerable code is in the server package. ## What Changed - Added a writable-state guard for raw upgrade sockets before rejecting or completing an upgrade. - Changed rejection responses from raw `write()` + `destroy()` to guarded `end()` with warning logging if rejection fails synchronously. - Attached a temporary raw socket error listener during the async upgrade authorization window and cleaned it up on socket close or successful `ws.handleUpgrade()`. - Added regression tests for rejecting after an early socket close and for handling raw socket `error` events while authorization is pending. ## Verification - `pnpm exec vitest run server/src/__tests__/live-events-ws.test.ts` - `pnpm --filter @paperclipai/server typecheck` - `git diff --check` ## Risks - Low risk: the change is scoped to the live-events WebSocket upgrade path before `ws` takes ownership of the socket. - Rejected upgrade responses now use graceful `socket.end(...)`; clients should still receive the same HTTP status text when the socket is writable. - The temporary error listener is removed before successful `handleUpgrade()` so normal WebSocket client error handling remains owned by the existing `connection` path. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used OpenAI Codex, GPT-5 coding agent with repository-aware tool use, shell command execution, GitHub connector access, and local code editing. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] If this change affects the UI, I have included before/after screenshots - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- server/src/__tests__/live-events-ws.test.ts | 123 ++++++++++++++++++++ server/src/realtime/live-events-ws.ts | 44 ++++++- 2 files changed, 164 insertions(+), 3 deletions(-) create mode 100644 server/src/__tests__/live-events-ws.test.ts diff --git a/server/src/__tests__/live-events-ws.test.ts b/server/src/__tests__/live-events-ws.test.ts new file mode 100644 index 00000000..b52c1106 --- /dev/null +++ b/server/src/__tests__/live-events-ws.test.ts @@ -0,0 +1,123 @@ +import { EventEmitter } from "node:events"; +import type { IncomingMessage } from "node:http"; +import type { Duplex } from "node:stream"; +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { setupLiveEventsWebSocketServer } from "../realtime/live-events-ws.js"; +import { logger } from "../middleware/logger.js"; + +vi.mock("../middleware/logger.js", () => ({ + logger: { + warn: vi.fn(), + error: vi.fn(), + }, +})); + +class FakeUpgradeSocket extends EventEmitter { + destroyed = false; + writable = true; + writableEnded = false; + writableDestroyed = false; + endedChunks: string[] = []; + destroyCalls = 0; + + end(chunk?: string) { + if (chunk) this.endedChunks.push(chunk); + this.writableEnded = true; + this.writable = false; + setImmediate(() => { + if (this.destroyed) return; + this.emit("finish"); + if (!this.destroyed) { + this.emit("close"); + } + }); + return this; + } + + destroy() { + this.destroyCalls += 1; + this.destroyed = true; + this.writable = false; + this.writableDestroyed = true; + this.emit("close"); + return this; + } + + emitSocketError(err: Error) { + this.writable = false; + this.writableDestroyed = true; + this.emit("error", err); + } +} + +function createUpgradeRequest(overrides: Partial = {}) { + return { + url: "/api/companies/company-1/events/ws", + headers: {}, + ...overrides, + } as IncomingMessage; +} + +async function flushPromises() { + await new Promise((resolve) => setImmediate(resolve)); +} + +describe("setupLiveEventsWebSocketServer", () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it("does not write a rejection response after the raw upgrade socket is already closed", async () => { + const server = new EventEmitter(); + setupLiveEventsWebSocketServer(server as never, {} as never, { deploymentMode: "authenticated" }); + const socket = new FakeUpgradeSocket(); + + server.emit("upgrade", createUpgradeRequest(), socket as unknown as Duplex, Buffer.alloc(0)); + socket.destroy(); + await flushPromises(); + + expect(socket.endedChunks).toEqual([]); + expect(socket.destroyCalls).toBe(1); + }); + + it("handles raw upgrade socket errors during async authorization", async () => { + const server = new EventEmitter(); + let resolveSession: (value: null) => void = () => undefined; + setupLiveEventsWebSocketServer(server as never, {} as never, { + deploymentMode: "authenticated", + resolveSessionFromHeaders: () => + new Promise((resolve) => { + resolveSession = resolve; + }), + }); + const socket = new FakeUpgradeSocket(); + + server.emit("upgrade", createUpgradeRequest(), socket as unknown as Duplex, Buffer.alloc(0)); + expect(() => socket.emitSocketError(new Error("write EPIPE"))).not.toThrow(); + resolveSession(null); + await flushPromises(); + + expect(logger.warn).toHaveBeenCalledWith( + expect.objectContaining({ err: expect.any(Error), path: "/api/companies/company-1/events/ws" }), + "live websocket upgrade socket error", + ); + expect(socket.endedChunks).toEqual([]); + expect(socket.destroyed).toBe(true); + }); + + it("destroys and cleans up listeners after flushing a rejection response", async () => { + const server = new EventEmitter(); + setupLiveEventsWebSocketServer(server as never, {} as never, { deploymentMode: "authenticated" }); + const socket = new FakeUpgradeSocket(); + + server.emit("upgrade", createUpgradeRequest(), socket as unknown as Duplex, Buffer.alloc(0)); + await flushPromises(); + await flushPromises(); + + expect(socket.endedChunks[0]).toContain("403 Forbidden"); + expect(socket.destroyed).toBe(true); + expect(socket.listenerCount("error")).toBe(0); + expect(socket.listenerCount("close")).toBe(0); + expect(socket.listenerCount("finish")).toBe(0); + }); +}); diff --git a/server/src/realtime/live-events-ws.ts b/server/src/realtime/live-events-ws.ts index d18e2930..07f4b05f 100644 --- a/server/src/realtime/live-events-ws.ts +++ b/server/src/realtime/live-events-ws.ts @@ -54,10 +54,31 @@ function hashToken(token: string) { return createHash("sha256").update(token).digest("hex"); } +function isWritableUpgradeSocket(socket: Duplex) { + const maybeWritableState = socket as Duplex & { writable?: boolean; writableEnded?: boolean; writableDestroyed?: boolean }; + return !socket.destroyed && maybeWritableState.writable !== false && !maybeWritableState.writableEnded && !maybeWritableState.writableDestroyed; +} + +function closeUpgradeSocket(socket: Duplex) { + if (!socket.destroyed) { + socket.destroy(); + } +} + function rejectUpgrade(socket: Duplex, statusLine: string, message: string) { const safe = message.replace(/[\r\n]+/g, " ").trim(); - socket.write(`HTTP/1.1 ${statusLine}\r\nConnection: close\r\nContent-Type: text/plain\r\n\r\n${safe}`); - socket.destroy(); + if (!isWritableUpgradeSocket(socket)) { + closeUpgradeSocket(socket); + return; + } + + try { + socket.once("finish", () => closeUpgradeSocket(socket)); + socket.end(`HTTP/1.1 ${statusLine}\r\nConnection: close\r\nContent-Type: text/plain\r\n\r\n${safe}`); + } catch (err) { + logger.warn({ err }, "failed to reject live websocket upgrade"); + closeUpgradeSocket(socket); + } } function parseCompanyId(pathname: string) { @@ -234,6 +255,17 @@ export function setupLiveEventsWebSocketServer( }); server.on("upgrade", (req, socket, head) => { + const onRawSocketError = (err: Error) => { + logger.warn({ err, path: req.url }, "live websocket upgrade socket error"); + }; + const cleanupRawSocketListeners = () => { + socket.off("error", onRawSocketError); + socket.off("close", cleanupRawSocketListeners); + }; + + socket.on("error", onRawSocketError); + socket.once("close", cleanupRawSocketListeners); + if (!req.url) { rejectUpgrade(socket, "400 Bad Request", "missing url"); return; @@ -242,7 +274,7 @@ export function setupLiveEventsWebSocketServer( const url = new URL(req.url, "http://localhost"); const companyId = parseCompanyId(url.pathname); if (!companyId) { - socket.destroy(); + closeUpgradeSocket(socket); return; } @@ -256,9 +288,15 @@ export function setupLiveEventsWebSocketServer( return; } + if (!isWritableUpgradeSocket(socket)) { + cleanupRawSocketListeners(); + return; + } + const reqWithContext = req as IncomingMessageWithContext; reqWithContext.paperclipUpgradeContext = context; + cleanupRawSocketListeners(); wss.handleUpgrade(req, socket, head, (ws: WsSocket) => { wss.emit("connection", ws, reqWithContext); });