fix(server): harden live events upgrade sockets (#8383)
## 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 <noreply@paperclip.ing>
This commit is contained in:
Aron Prins
@@ -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<IncomingMessage> = {}) {
|
||||
return {
|
||||
url: "/api/companies/company-1/events/ws",
|
||||
headers: {},
|
||||
...overrides,
|
||||
} as IncomingMessage;
|
||||
}
|
||||
|
||||
async function flushPromises() {
|
||||
await new Promise<void>((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);
|
||||
});
|
||||
});
|
||||
@@ -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);
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user