diff --git a/docs/scratchpads/838-broker-acceptance-flake.md b/docs/scratchpads/838-broker-acceptance-flake.md index 1a9fa41..6934048 100644 --- a/docs/scratchpads/838-broker-acceptance-flake.md +++ b/docs/scratchpads/838-broker-acceptance-flake.md @@ -36,3 +36,7 @@ Eliminate high-contention uncaught JSON parse failures in lease-broker and mutat - Residual/tripwire: an unavoidable peer disconnect or send failure can still lose acknowledgement after a valid transaction commits. The affected adapter call fails closed. A valid `promote_lease` may nevertheless remain VERIFIED after its acknowledgement is lost; that is authority-observability divergence requiring WI-3/Opus security review rather than expansion of #838. #838 does not add retries or attempt a protocol redesign. - Final package evidence: recursive Mosaic dependency build passed; 73 Vitest files / 1,392 tests passed; deadline unit tests 2/2, real runtime tool tests 15/15, launch guard tests 12/12, inventory 14/14, and shell regressions passed. - Final coverage: `broker-test-client.ts` 99.24% statements/lines, 86.11% branches, 100% functions under per-file >=85% thresholds. Repository typecheck 42/42, lint 23/23, and format check passed. +- Independent Codex exact-head review requested one framing fix: a valid frame followed by trailing bytes in a later socket chunk could resolve before the garbage arrived. Security review also flagged complete token-bearing reply bodies in diagnostic properties/logs. +- Review RED evidence: delayed cross-chunk garbage resolved `{ ok: true }` instead of rejecting, and a truncated `promotion_token` remained in the typed error. The tests use separate timed writes to prevent kernel/event-loop coalescing. +- Review remediation: the shared client now accumulates through EOF, requires exactly one terminal newline, then parses inside a rejecting error boundary. Diagnostic bodies are capped at 256 bytes, sensitive broker fields are fully redacted, and a SHA-256 digest preserves correlation without credential disclosure. +- Post-review coverage: `broker-test-client.ts` 99.35% statements/lines, 86.66% branches, 100% functions. Final full suite is 73 files / 1,394 tests plus all Python/shell gates; recursive build, typecheck, lint, and format are green. diff --git a/packages/mosaic/src/lease-broker/broker-test-client.ts b/packages/mosaic/src/lease-broker/broker-test-client.ts index f26ed88..b3aeddb 100644 --- a/packages/mosaic/src/lease-broker/broker-test-client.ts +++ b/packages/mosaic/src/lease-broker/broker-test-client.ts @@ -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 ? '' : 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 + ? '' + : sensitive + ? '' + : `${bounded.toString('utf8')}${suffix}`, + hex: sensitive ? '' : `${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 ? '' : 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( 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( ); } }); - socket.once('end', () => { - if (!settled) { - fail( - new BrokerTransportError( - 'early-close', - 'Broker reply socket closed before newline', - response, - ), - ); - } - }); socket.once('connect', () => { try { write(socket);