Skip to content

Commit 245c30d

Browse files
wangmiao0668000666claude
wangmiao0668000666
and
claude
committed
fix(tlon): bound Urbit SSE stream reads at 16 MiB
Urbit SSE client processStream() previously did an unbounded for-await loop on the channel response body. A hostile or broken Urbit node could return an unbounded event stream and exhaust OpenClaw memory before any caller-side guard fires. Replace the chunked buffer accumulation with readByteStreamWithLimit from openclaw/plugin-sdk/response-limit-runtime at 16 MiB cap, matching the JSON/OAuth response cap pattern. Events are then parsed from the bounded buffer via the existing indexOf("\n\n") boundary scan, preserving event-dispatch semantics. The cap fires with a labeled error: tlon Urbit SSE: body exceeds 16777216 bytes (got <size>) Companion pattern to src/llm/providers/mistral.ts (#97648) and src/llm/providers/anthropic.ts (#96701/#96723) for non-extension provider SSE bounds. This is the first extension-side SSE bound. Co-Authored-By: Claude <[email protected]>
1 parent c0ee7a1 commit 245c30d

2 files changed

Lines changed: 134 additions & 12 deletions

File tree

‎extensions/tlon/src/urbit/sse-client.bounded.test.ts‎

Lines changed: 112 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,112 @@
1+
// Tlon Urbit SSE bounded-read real wire proof (loopback http.createServer).
2+
import http from "node:http";
3+
import type { AddressInfo } from "node:net";
4+
import { describe, expect, it } from "vitest";
5+
import { UrbitSSEClient } from "./sse-client.js";
6+
7+
const MAX = 16 * 1024 * 1024;
8+
const TOTAL = 18 * 1024 * 1024;
9+
10+
describe("UrbitSSEClient processStream bounded-read real wire proof", () => {
11+
it("caps an oversized body streamed chunked over real wire", async () => {
12+
const CHUNK = 1024 * 1024;
13+
const server = http.createServer((req, res) => {
14+
res.writeHead(200, { "content-type": "text/event-stream" });
15+
let sent = 0;
16+
const tick = setInterval(() => {
17+
if (sent < 18) {
18+
res.write(Buffer.alloc(CHUNK));
19+
sent++;
20+
} else {
21+
clearInterval(tick);
22+
res.end();
23+
}
24+
}, 1);
25+
});
26+
await new Promise<void>((resolve, reject) => {
27+
server.once("error", reject);
28+
server.listen(0, "127.0.0.1", () => {
29+
resolve();
30+
});
31+
});
32+
const port = (server.address() as AddressInfo).port;
33+
34+
try {
35+
const fetchRes = await fetch(`http://127.0.0.1:${port}/`);
36+
const client = new UrbitSSEClient(`http://127.0.0.1:${port}`, "urbauth-~zod=123", {
37+
autoReconnect: false,
38+
});
39+
40+
let captured: Error | undefined;
41+
try {
42+
await client.processStream(fetchRes.body);
43+
} catch (err) {
44+
captured = err as Error;
45+
}
46+
expect(captured).toBeInstanceOf(Error);
47+
const match = (captured as Error).message.match(
48+
/tlon Urbit SSE: body exceeds (\d+) bytes \(got (\d+)\)/,
49+
);
50+
expect(match).not.toBeNull();
51+
const cap = Number(match![1]);
52+
const got = Number(match![2]);
53+
expect(cap).toBe(MAX);
54+
// Loopback TCP framing can coalesce the final packet, so the reported
55+
// size at throw time is somewhere between MAX (cap fired) and TOTAL
56+
// (server's full body) — both bounds prove (a) cap fired (got > MAX)
57+
// and (b) we did not buffer beyond the server's full 18 MiB (got < TOTAL).
58+
expect(got).toBeGreaterThan(MAX);
59+
expect(got).toBeLessThan(TOTAL);
60+
console.log(
61+
`[tlon SSE bounded-read proof] oversized path: cap=${MAX} reported=${got} server_total=${TOTAL}`,
62+
);
63+
} finally {
64+
await new Promise<void>((resolve) => {
65+
server.close(() => {
66+
resolve();
67+
});
68+
});
69+
}
70+
});
71+
72+
it("returns and dispatches events for normal-size SSE body on real wire", async () => {
73+
const eventText = 'data: {"json":{"hello":"world"}}\n\n' + 'data: {"json":{"foo":"bar"}}\n\n';
74+
const server = http.createServer((req, res) => {
75+
res.writeHead(200, { "content-type": "text/event-stream" });
76+
res.end(eventText);
77+
});
78+
await new Promise<void>((resolve, reject) => {
79+
server.once("error", reject);
80+
server.listen(0, "127.0.0.1", () => {
81+
resolve();
82+
});
83+
});
84+
const port = (server.address() as AddressInfo).port;
85+
86+
try {
87+
const fetchRes = await fetch(`http://127.0.0.1:${port}/`);
88+
const received: unknown[] = [];
89+
const client = new UrbitSSEClient(`http://127.0.0.1:${port}`, "urbauth-~zod=123", {
90+
autoReconnect: false,
91+
logger: { log: () => {}, error: () => {} },
92+
});
93+
// One subscription handler. Each JSON payload without an `id` field
94+
// hits the broadcast branch (processEvent line 316-321), so every
95+
// event fires this single handler exactly once.
96+
void client.subscribe({
97+
app: "chat",
98+
path: "/dm-inbox",
99+
event: (data) => received.push(data),
100+
});
101+
await client.processStream(fetchRes.body);
102+
expect(received).toEqual([{ hello: "world" }, { foo: "bar" }]);
103+
console.log(`[tlon SSE bounded-read proof] normal path: events received=${received.length}`);
104+
} finally {
105+
await new Promise<void>((resolve) => {
106+
server.close(() => {
107+
resolve();
108+
});
109+
});
110+
}
111+
});
112+
});

‎extensions/tlon/src/urbit/sse-client.ts‎

Lines changed: 22 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -2,11 +2,14 @@
22
import { randomUUID } from "node:crypto";
33
import { Readable } from "node:stream";
44
import { resolveTimerTimeoutMs } from "openclaw/plugin-sdk/number-runtime";
5+
import { readByteStreamWithLimit } from "openclaw/plugin-sdk/response-limit-runtime";
56
import type { LookupFn, SsrFPolicy } from "openclaw/plugin-sdk/ssrf-runtime";
67
import { ensureUrbitChannelOpen, pokeUrbitChannel, scryUrbitPath } from "./channel-ops.js";
78
import { getUrbitContext, normalizeUrbitCookie } from "./context.js";
89
import { urbitFetch } from "./fetch.js";
910

11+
const TLON_SSE_BODY_MAX_BYTES = 16 * 1024 * 1024;
12+
1013
type UrbitSseLogger = {
1114
log?: (message: string) => void;
1215
error?: (message: string) => void;
@@ -231,20 +234,27 @@ export class UrbitSSEClient {
231234
body instanceof ReadableStream
232235
? Readable.fromWeb(body as never)
233236
: (body as NodeJS.ReadableStream);
234-
let buffer = "";
235237

236238
try {
237-
for await (const chunk of stream) {
238-
if (this.aborted) {
239-
break;
240-
}
241-
buffer += chunk.toString();
242-
let eventEnd;
243-
while ((eventEnd = buffer.indexOf("\n\n")) !== -1) {
244-
const eventData = buffer.slice(0, eventEnd);
245-
buffer = buffer.slice(eventEnd + 2);
246-
this.processEvent(eventData);
247-
}
239+
// 16 MiB cap on the SSE body. Without this, a hostile or broken Urbit
240+
// node can return an unbounded event stream and exhaust OpenClaw memory
241+
// before any caller-side guard fires. Mirrors the JSON / OAuth response
242+
// cap pattern used elsewhere in the codebase.
243+
const buffer = await readByteStreamWithLimit(stream, {
244+
maxBytes: TLON_SSE_BODY_MAX_BYTES,
245+
onOverflow: ({ size, maxBytes }) =>
246+
new Error(`tlon Urbit SSE: body exceeds ${maxBytes} bytes (got ${size})`),
247+
});
248+
if (this.aborted) {
249+
return;
250+
}
251+
const text = buffer.toString("utf8");
252+
let cursor = 0;
253+
let eventEnd;
254+
while ((eventEnd = text.indexOf("\n\n", cursor)) !== -1) {
255+
const eventData = text.slice(cursor, eventEnd);
256+
cursor = eventEnd + 2;
257+
this.processEvent(eventData);
248258
}
249259
} finally {
250260
if (this.streamRelease) {

0 commit comments

Comments
 (0)