src/connect.js
Rama master · réplica de lectura del repositorio original.
/**
* Connect-RPC envelope framing and compression.
*
* Connect-RPC frame format:
* [1 byte flags] [4 bytes big-endian length] [N bytes payload]
*
* Flags:
* 0x01 = gzip compressed
* 0x02 = end-of-stream (trailer frame, JSON payload)
* 0x03 = compressed + end-of-stream
*
* IMPORTANT: Connect-RPC uses HTTP/1.1 POST, NOT HTTP/2 gRPC.
* Content-Type: application/connect+proto
*/
import { gzipSync, gunzipSync } from 'zlib';
import { isUint8Array } from 'util/types';
// Hard ceiling on any single frame — both the wire length a frame may advertise
// and the number of bytes a gzip payload may inflate to. Bounding the DECOMPRESSED
// size (via zlib's maxOutputLength) is what stops a high-ratio "zip bomb" frame from
// a hijacked/compromised upstream from inflating to gigabytes and OOM-killing the
// single-process proxy before the resulting error can be caught (audit CONN-1).
// Same value/rationale as StreamingFrameParser's frame-length guard; mirrors the
// maxOutputLength pattern already used for PDF flate streams (pdf.js:52).
export const MAX_FRAME_SIZE = 16 * 1024 * 1024;
// ─── Compression helpers ───────────────────────────────────
export function gzip(buf) { return gzipSync(buf); }
export function gunzip(buf) { return gunzipSync(buf, { maxOutputLength: MAX_FRAME_SIZE }); }
export function tryGunzip(buf) {
try { return gunzipSync(buf, { maxOutputLength: MAX_FRAME_SIZE }); }
catch { return null; }
}
// ─── Envelope wrapping ─────────────────────────────────────
/**
* Wrap protobuf bytes in a Connect-RPC envelope frame.
*/
export function wrapEnvelope(protoBuf, { compress = true } = {}) {
let payload = protoBuf;
let flags = 0;
if (compress && payload.length > 0) {
payload = gzipSync(payload);
flags |= 0x01;
}
const frame = Buffer.alloc(5 + payload.length);
frame[0] = flags;
frame.writeUInt32BE(payload.length, 1);
payload.copy(frame, 5);
return frame;
}
/**
* Wrap a request for sending (single envelope, gzipped).
*/
export function wrapRequest(protoBuf) {
return wrapEnvelope(protoBuf, { compress: true });
}
/**
* Build the end-of-stream trailer frame (JSON {}).
*/
export function endOfStreamEnvelope() {
const trailer = Buffer.from('{}');
const frame = Buffer.alloc(5 + trailer.length);
frame[0] = 0x02; // end-of-stream, not compressed
frame.writeUInt32BE(trailer.length, 1);
trailer.copy(frame, 5);
return frame;
}
// ─── Request unwrapping ────────────────────────────────────
/**
* Unwrap a Connect-RPC request body → raw protobuf bytes.
* Handles both envelope-wrapped and HTTP-level gzip.
*/
export function unwrapRequest(body, headers = {}) {
let buf = Buffer.isBuffer(body) ? body : Buffer.from(body);
// HTTP-level content-encoding gzip
const encoding = headers['content-encoding'] || headers['connect-content-encoding'] || '';
if (encoding === 'gzip') {
buf = gunzipSync(buf, { maxOutputLength: MAX_FRAME_SIZE });
}
// Check if it's envelope-wrapped (flags byte + 4-byte length)
if (buf.length >= 5) {
const flags = buf[0];
const len = buf.readUInt32BE(1);
if (len === buf.length - 5 && (flags === 0 || flags === 1)) {
let payload = buf.subarray(5);
if (flags & 0x01) payload = gunzipSync(payload, { maxOutputLength: MAX_FRAME_SIZE });
return payload;
}
}
return buf;
}
// ─── Streaming frame parser ───────────────────────────────
/**
* Stateful parser that buffers incoming data and yields complete frames.
*/
export class StreamingFrameParser {
constructor() {
this._chunks = [];
this._head = 0;
this._offset = 0;
this._length = 0;
}
push(chunk) {
if (!isUint8Array(chunk)) {
// Preserve Buffer.concat's invalid-input error without copying valid data.
Buffer.concat([Buffer.alloc(0), chunk]);
}
// Borrow transport chunks until consumed. Empty chunks must not accumulate
// queue entries or turn a five-byte header lookup into an unbounded scan.
if (chunk.length === 0) return;
this._chunks.push(chunk);
this._length += chunk.length;
}
_byteAt(relative) {
let index = this._head;
let offset = this._offset + relative;
while (offset >= this._chunks[index].length) {
offset -= this._chunks[index++].length;
}
return this._chunks[index][offset];
}
_copyPayload(length) {
const payload = Buffer.alloc(length);
if (length === 0) return payload;
let index = this._head;
let offset = this._offset + 5;
while (offset >= this._chunks[index].length) {
offset -= this._chunks[index++].length;
}
let written = 0;
while (written < length) {
const chunk = this._chunks[index++];
const take = Math.min(length - written, chunk.length - offset);
payload.set(chunk.subarray(offset, offset + take), written);
written += take;
offset = 0;
}
return payload;
}
_consume(length) {
this._length -= length;
while (length > 0) {
const available = this._chunks[this._head].length - this._offset;
if (length < available) {
this._offset += length;
break;
}
length -= available;
this._chunks[this._head++] = null;
this._offset = 0;
}
if (this._head === this._chunks.length) {
this._chunks = [];
this._head = 0;
} else if (this._head * 2 >= this._chunks.length) {
// Copy only references, only after at least as many entries were retired.
// This bounds retained slots and amortizes compaction across consumed chunks.
this._chunks = this._chunks.slice(this._head);
this._head = 0;
}
}
/** Drain all complete frames. Returns [{ flags, isEndStream, payload }]. */
drain() {
// Validate the advertised length before waiting for or allocating its body.
// Reading five header bytes never walks the accumulated payload prefix.
const frames = [];
while (this._length >= 5) {
let len = 0;
for (let i = 1; i < 5; i++) len = len * 256 + this._byteAt(i);
if (len > MAX_FRAME_SIZE) {
throw new Error(`HTTP/2 frame size ${len} exceeds ${MAX_FRAME_SIZE}`);
}
if (this._length < 5 + len) break;
const flags = this._byteAt(0);
let payload = this._copyPayload(len);
if (flags & 0x01) {
// Bound the DECOMPRESSED size too: a high-ratio gzip frame that passes the
// ≤16MB wire-length check above can still inflate to gigabytes. Without
// maxOutputLength, gunzipSync accumulates up to kMaxLength (~2GB) and on a
// memory-constrained host the OOM-killer can reap the process before the
// throw below is caught (audit CONN-1).
try { payload = gunzipSync(payload, { maxOutputLength: MAX_FRAME_SIZE }); }
catch (err) {
// Don't silently drop a frame whose body we couldn't decompress —
// upstream is sending something we don't understand and the caller
// needs to know so the stream can be cancelled rather than continued
// with missing data.
throw new Error(`Connect frame decompression failed: ${err.message}`);
}
}
frames.push({
flags,
isEndStream: !!(flags & 0x02),
payload,
});
// A decompression error must leave the failing frame queued, as before.
this._consume(5 + len);
}
return frames;
}
}
// ─── Connect-RPC headers ──────────────────────────────────
export function connectHeaders(extra = {}) {
return {
'Content-Type': 'application/connect+proto',
'Connect-Protocol-Version': '1',
'Connect-Accept-Encoding': 'gzip',
'User-Agent': 'connect-es/2.0.0',
...extra,
};
}