Skip to content

Commit 4e0078e

Browse files
minggangwminggangw
committed
Address comments
1 parent 14cbe8e commit 4e0078e

3 files changed

Lines changed: 184 additions & 23 deletions

File tree

‎lib/runtime/transports/http.js‎

Lines changed: 66 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -63,6 +63,7 @@ class HttpRequestConnection extends Connection {
6363
this._capability = capability;
6464
this._payload = payload;
6565
this._replied = false;
66+
this._closed = false;
6667
// Synthesised frame id so the dispatcher can correlate even though
6768
// the HTTP path doesn't carry one on the wire.
6869
this._id = '__http__';
@@ -84,7 +85,9 @@ class HttpRequestConnection extends Connection {
8485
/**
8586
* Receive a reply from the dispatcher. Translates `{ok, payload,
8687
* error, code}` into the appropriate HTTP response. Subsequent calls
87-
* are no-ops — HTTP is one-shot.
88+
* are no-ops — HTTP is one-shot. Always emits `'close'` exactly once
89+
* after the reply is written so the dispatcher's per-connection
90+
* cleanup runs and lazy-created Publishers/Clients are released.
8891
*/
8992
send(frame) {
9093
if (this._replied) return;
@@ -98,7 +101,7 @@ class HttpRequestConnection extends Connection {
98101
'streaming_unsupported',
99102
'streaming events are not supported over HTTP'
100103
);
101-
return;
104+
return this._emitCloseOnce();
102105
}
103106

104107
if (frame.ok === true) {
@@ -113,13 +116,14 @@ class HttpRequestConnection extends Connection {
113116
});
114117
this.res.end(body);
115118
}
116-
return;
119+
return this._emitCloseOnce();
117120
}
118121

119122
// Failure path.
120123
const code = frame.code || 'internal_error';
121124
const status = _STATUS_BY_CODE[code] || 500;
122125
this._writeError(status, code, frame.error || 'request failed');
126+
this._emitCloseOnce();
123127
}
124128

125129
/** Force-close the response. The dispatcher's `cleanup` will follow. */
@@ -128,6 +132,16 @@ class HttpRequestConnection extends Connection {
128132
if (!this._replied) {
129133
this._writeError(500, 'internal_error', 'connection closed');
130134
}
135+
this._emitCloseOnce();
136+
}
137+
138+
/**
139+
* Idempotent `'close'` emitter — guarantees the dispatcher's
140+
* cleanup runs exactly once even if `send()` and `close()` race.
141+
*/
142+
_emitCloseOnce() {
143+
if (this._closed) return;
144+
this._closed = true;
131145
this.emit('close');
132146
}
133147

@@ -222,12 +236,25 @@ class HttpTransport extends TransportAdapter {
222236
// ---------- internals ----------
223237

224238
_route(req, res) {
225-
if (this.verifyRequest && this.verifyRequest(req) === false) {
226-
return _writeJson(res, 401, {
227-
ok: false,
228-
error: 'unauthorized',
229-
code: 'unauthorized',
230-
});
239+
if (this.verifyRequest) {
240+
let allowed;
241+
try {
242+
allowed = this.verifyRequest(req);
243+
} catch (e) {
244+
debug('verifyRequest threw: %s', (e && e.stack) || e);
245+
return _writeJson(res, 500, {
246+
ok: false,
247+
error: 'verifyRequest hook failed',
248+
code: 'internal_error',
249+
});
250+
}
251+
if (allowed === false) {
252+
return _writeJson(res, 401, {
253+
ok: false,
254+
error: 'unauthorized',
255+
code: 'unauthorized',
256+
});
257+
}
231258
}
232259

233260
let pathname;
@@ -261,7 +288,16 @@ class HttpTransport extends TransportAdapter {
261288
});
262289
}
263290
const kind = tail.slice(0, slash);
264-
const name = '/' + decodeURIComponent(tail.slice(slash + 1));
291+
let name;
292+
try {
293+
name = '/' + decodeURIComponent(tail.slice(slash + 1));
294+
} catch {
295+
return _writeJson(res, 400, {
296+
ok: false,
297+
error: `invalid percent-encoding in capability name: ${tail.slice(slash + 1)}`,
298+
code: 'invalid_url',
299+
});
300+
}
265301

266302
if (kind !== 'call' && kind !== 'publish') {
267303
return _writeJson(res, 404, {
@@ -308,6 +344,16 @@ function _writeJson(res, status, body) {
308344
}
309345

310346
function _readJsonBody(req, maxBytes, cb) {
347+
// Guard against double-callback: the body may exceed maxBytes (we call
348+
// cb(err) and req.destroy()), and the destroy itself can synchronously
349+
// emit 'error' which would otherwise reach cb(err) a second time.
350+
let done = false;
351+
const finish = (err, value) => {
352+
if (done) return;
353+
done = true;
354+
cb(err, value);
355+
};
356+
311357
const ctype = (req.headers['content-type'] || '').toLowerCase();
312358
if (
313359
req.method === 'POST' &&
@@ -322,35 +368,38 @@ function _readJsonBody(req, maxBytes, cb) {
322368
`unsupported content-type: ${ctype} (expected application/json)`
323369
);
324370
e.code = 'invalid_content_type';
325-
return cb(e);
371+
return finish(e);
326372
}
327373
}
328374
let total = 0;
329375
const chunks = [];
330376
req.on('data', (chunk) => {
377+
if (done) return;
331378
total += chunk.length;
332379
if (total > maxBytes) {
333-
req.destroy();
334380
const e = new Error(`request body exceeds ${maxBytes} bytes`);
335381
e.code = 'payload_too_large';
336-
return cb(e);
382+
finish(e);
383+
req.destroy();
384+
return;
337385
}
338386
chunks.push(chunk);
339387
});
340388
req.on('end', () => {
389+
if (done) return;
341390
const raw = Buffer.concat(chunks).toString('utf8');
342-
if (!raw) return cb(null, {});
391+
if (!raw) return finish(null, {});
343392
try {
344-
cb(null, JSON.parse(raw));
393+
finish(null, JSON.parse(raw));
345394
} catch (e) {
346395
const err = new Error(`invalid JSON: ${e.message}`);
347396
err.code = 'invalid_json';
348-
cb(err);
397+
finish(err);
349398
}
350399
});
351400
req.on('error', (e) => {
352401
e.code = e.code || 'request_error';
353-
cb(e);
402+
finish(e);
354403
});
355404
}
356405

‎test/test-web-http.js‎

Lines changed: 90 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,8 +17,9 @@ const {
1717
} = require('../lib/runtime');
1818

1919
let connect;
20+
let RosClient;
2021
before(async function () {
21-
({ connect } = await import('../web/index.js'));
22+
({ connect, RosClient } = await import('../web/index.js'));
2223
});
2324

2425
describe('rclnodejs/web — HTTP transport (call + publish)', function () {
@@ -238,6 +239,94 @@ describe('rclnodejs/web — HTTP transport (call + publish)', function () {
238239
}
239240
});
240241
});
242+
243+
// -----------------------------------------------------------------
244+
// Defensive paths — regressions for the resource-leak, malformed-URL,
245+
// throwing-verifyRequest, and bad-endpoint cases.
246+
// -----------------------------------------------------------------
247+
describe('defensive paths', function () {
248+
it('runs dispatcher cleanup on every HTTP request (no Publisher leak)', async function () {
249+
// Each `publish()` lazily creates one Publisher per (connection,
250+
// capability). Without an emitted 'close' on HttpRequestConnection,
251+
// those Publishers would accumulate forever. We can't directly
252+
// observe the Map, but we can publish many times and assert the
253+
// node's publisher list doesn't grow proportionally.
254+
const before = node.countPublishers('/http_chatter');
255+
for (let i = 0; i < 5; i++) {
256+
const res = await fetch(httpBase + '/capability/publish/http_chatter', {
257+
method: 'POST',
258+
headers: { 'content-type': 'application/json' },
259+
body: JSON.stringify({ data: 'leak-check-' + i }),
260+
});
261+
assert.strictEqual(res.status, 204);
262+
}
263+
// Discovery is async; allow a tick.
264+
await new Promise((r) => setTimeout(r, 50));
265+
const after = node.countPublishers('/http_chatter');
266+
assert.ok(
267+
after - before <= 1,
268+
`expected at most 1 leftover publisher after 5 requests, got ${after - before}`
269+
);
270+
});
271+
272+
it('returns 400 invalid_url for malformed percent-encoding', async function () {
273+
const res = await fetch(httpBase + '/capability/call/%E0%A4%A');
274+
assert.strictEqual(res.status, 400);
275+
const body = await res.json();
276+
assert.strictEqual(body.code, 'invalid_url');
277+
});
278+
279+
it('returns 500 internal_error if verifyRequest throws', async function () {
280+
const failingRuntime = createRuntime({
281+
node,
282+
transport: new HttpTransport({
283+
port: 0,
284+
host: '127.0.0.1',
285+
verifyRequest: () => {
286+
throw new Error('boom');
287+
},
288+
}),
289+
});
290+
failingRuntime.expose({
291+
call: { '/http_add': 'example_interfaces/srv/AddTwoInts' },
292+
});
293+
await failingRuntime.start();
294+
const port = failingRuntime.transports[0].port;
295+
try {
296+
const res = await fetch(
297+
`http://127.0.0.1:${port}/capability/call/http_add`,
298+
{
299+
method: 'POST',
300+
headers: { 'content-type': 'application/json' },
301+
body: '{}',
302+
}
303+
);
304+
assert.strictEqual(res.status, 500);
305+
const body = await res.json();
306+
assert.strictEqual(body.code, 'internal_error');
307+
} finally {
308+
await failingRuntime.stop();
309+
}
310+
});
311+
312+
it('rejects connect({}) with at-least-one-of error', function () {
313+
assert.throws(() => new RosClient({}), /at least one of http or ws/);
314+
});
315+
316+
it('rejects connect({http: 42}) with non-string error', function () {
317+
assert.throws(
318+
() => new RosClient({ http: 42 }),
319+
/http must be a non-empty string/
320+
);
321+
});
322+
323+
it('rejects connect({ws: "http://x"}) with wrong-scheme error', function () {
324+
assert.throws(
325+
() => new RosClient({ ws: 'http://x' }),
326+
/ws must start with ws:\/\/ or wss:\/\//
327+
);
328+
});
329+
});
241330
});
242331

243332
function waitFor(predicate, timeoutMs) {

‎web/client.js‎

Lines changed: 28 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -459,11 +459,14 @@ export class RosClient {
459459
*/
460460
function _resolveUrls(url) {
461461
if (url && typeof url === 'object' && !Array.isArray(url)) {
462-
return {
463-
httpUrl: url.http || null,
464-
wsUrl: url.ws || null,
465-
wsExplicit: !!url.ws,
466-
};
462+
const httpUrl = _validateEndpoint(url.http, 'http', /^https?:\/\//i);
463+
const wsUrl = _validateEndpoint(url.ws, 'ws', /^wss?:\/\//i);
464+
if (!httpUrl && !wsUrl) {
465+
throw new TypeError(
466+
'connect({http, ws}): at least one of http or ws must be provided'
467+
);
468+
}
469+
return { httpUrl, wsUrl, wsExplicit: !!wsUrl };
467470
}
468471
if (typeof url !== 'string' || !url) {
469472
throw new TypeError(
@@ -493,6 +496,26 @@ function _resolveUrls(url) {
493496
);
494497
}
495498

499+
/**
500+
* Validate one endpoint of an `{http, ws}` pair. Returns the trimmed
501+
* URL on success, `null` if the field is absent, or throws a clear
502+
* TypeError if the field is present but not a usable string.
503+
*/
504+
function _validateEndpoint(value, field, schemeRe) {
505+
if (value === undefined || value === null) return null;
506+
if (typeof value !== 'string' || !value) {
507+
throw new TypeError(
508+
`connect({${field}}): ${field} must be a non-empty string, got ${typeof value}`
509+
);
510+
}
511+
if (!schemeRe.test(value)) {
512+
throw new TypeError(
513+
`connect({${field}}): ${field} must start with ${field === 'http' ? 'http:// or https://' : 'ws:// or wss://'} (got ${value})`
514+
);
515+
}
516+
return value;
517+
}
518+
496519
/**
497520
* Open a connection to a Web Runtime capability endpoint.
498521
*

0 commit comments

Comments
 (0)