fix(#838): validate complete broker reply frames

This commit is contained in:
ms-lead-reviewer
2026-07-18 01:49:47 -05:00
parent 6f0f8eecdc
commit 6ae214a4a2
2 changed files with 54 additions and 20 deletions

View File

@@ -1,7 +1,10 @@
import { createHash } from 'node:crypto';
import { createConnection, type Socket } from 'node:net';
const DEFAULT_TIMEOUT_MS = 3_000;
const MAX_REPLY_BYTES = 64 * 1024;
const MAX_DIAGNOSTIC_BYTES = 256;
const SENSITIVE_REPLY_FIELD = /"(?:promotion_token|session_id|token)"\s*:/;
export interface BrokerTestClientOptions {
timeoutMs?: number;
@@ -17,6 +20,8 @@ export class BrokerTransportError extends Error {
public readonly responseLength: number;
public readonly responseBytes: string;
public readonly responseHex: string;
public readonly responsePreview: string;
public readonly responseSha256: string;
public constructor(
public readonly kind: BrokerTransportFailureKind,
@@ -24,18 +29,43 @@ export class BrokerTransportError extends Error {
response: Buffer,
public readonly attempts = 1,
) {
super(`${description}; attempts=${attempts}; ${responseContext(response)}`);
const diagnostics = responseDiagnostics(response);
super(`${description}; attempts=${attempts}; ${responseContext(response, diagnostics)}`);
this.name = 'BrokerTransportError';
this.responseLength = response.length;
this.responseBytes = response.toString('utf8');
this.responseHex = response.toString('hex');
this.responseBytes = diagnostics.preview;
this.responseHex = diagnostics.hex;
this.responsePreview = diagnostics.preview;
this.responseSha256 = diagnostics.sha256;
}
}
function responseContext(response: Buffer): string {
const text = JSON.stringify(response.toString('utf8'));
const hex = response.length === 0 ? '<empty>' : response.toString('hex');
return `length=${response.length}; bytes=${text}; hex=${hex}`;
interface ResponseDiagnostics {
preview: string;
hex: string;
sha256: string;
}
function responseDiagnostics(response: Buffer): ResponseDiagnostics {
const fullText = response.toString('utf8');
const sensitive = SENSITIVE_REPLY_FIELD.test(fullText);
const bounded = response.subarray(0, MAX_DIAGNOSTIC_BYTES);
const suffix = response.length > MAX_DIAGNOSTIC_BYTES ? '…' : '';
return {
preview:
response.length === 0
? '<empty>'
: sensitive
? '<redacted-sensitive-reply>'
: `${bounded.toString('utf8')}${suffix}`,
hex: sensitive ? '<redacted>' : `${bounded.toString('hex')}${suffix}`,
sha256: createHash('sha256').update(response).digest('hex'),
};
}
function responseContext(response: Buffer, diagnostics = responseDiagnostics(response)): string {
const hex = diagnostics.hex.length === 0 ? '<empty>' : diagnostics.hex;
return `length=${response.length}; bytes=${JSON.stringify(diagnostics.preview)}; hex=${hex}; sha256=${diagnostics.sha256}`;
}
function malformedReply(response: Buffer, reason: string): BrokerTransportError {
@@ -86,11 +116,22 @@ function readBrokerReplyAttempt<T extends object>(
response = Buffer.concat([response, chunk]);
if (response.length > MAX_REPLY_BYTES) {
fail(malformedReply(response, `exceeds ${MAX_REPLY_BYTES} bytes`));
}
});
socket.once('end', () => {
if (settled) return;
const newline = response.indexOf(0x0a);
if (response.length === 0 || newline < 0) {
fail(
new BrokerTransportError(
'early-close',
'Broker reply socket closed before newline',
response,
),
);
return;
}
const newline = response.indexOf(0x0a);
if (newline < 0) return;
if (newline !== response.length - 1) {
fail(malformedReply(response, 'contains bytes after the newline terminator'));
return;
@@ -109,17 +150,6 @@ function readBrokerReplyAttempt<T extends object>(
);
}
});
socket.once('end', () => {
if (!settled) {
fail(
new BrokerTransportError(
'early-close',
'Broker reply socket closed before newline',
response,
),
);
}
});
socket.once('connect', () => {
try {
write(socket);