Claude skill Node.js stream security

Stream security in MCP skills is distinct from stream correctness. A skill may process normal inputs correctly while being exploitable by an attacker who sends a 10 GB payload, submits thousands of small chunks at high frequency, provides a stream implementation that never emits end, or causes a downstream error that is silently swallowed. MCP amplifies these risks because tool calls are issued by an LLM operating on external content — a prompt injection in a document the LLM processes can instruct it to pass a very large file URL, or to use a specific stream argument that the attacker controls. The skill author who tests with reasonably sized inputs will not see these failures. SkillAudit’s static analysis detects the code patterns that create this exposure without requiring adversarial test input.

SA-SKL-STREAM-001: Readable stream with no highWaterMark — unbounded buffering until OOM

Every Node.js Readable stream maintains an internal buffer between the data source and the consuming code. When the consumer is slower than the producer — because it’s processing with an LLM, writing to disk, or awaiting a database write — the stream buffers incoming data in Node.js heap memory. The highWaterMark option controls the target buffer size and determines when the stream signals the producer to pause (false from write()) or resume (drain event). The default highWaterMark is 16 KB for object-mode streams and 64 KB for byte streams. This default is not a hard cap — it is a target that the stream’s back-pressure mechanism uses to signal the producer. If the producer ignores the signal (or if there is no back-pressure mechanism), data continues to accumulate.

In practice, the default highWaterMark is not the binding limit. What limits actual memory usage is whether back-pressure is applied end-to-end through the entire pipeline. A Readable stream piped to a Transform that is piped to a slow consumer will accumulate data in the buffers of both the Readable and the Transform if the consumer’s back-pressure signal is not propagated correctly. An attacker who provides a 10 GB input file causes the skill to buffer progressively larger chunks in heap until the Node.js process runs out of memory and crashes — a denial of service against the MCP skill and all concurrent users of the same process.

// VULNERABLE: Readable stream with no highWaterMark and no size limit
import { createReadStream } from 'fs';
import { pipeline } from 'stream/promises';

server.setToolHandler('process_file', async (args) => {
  // args.filePath comes from LLM tool call
  // No highWaterMark limit — uses default 64 KB target (NOT a hard cap)
  // No total size check before starting the stream
  const src = createReadStream(args.filePath);
  //          ^^^^^ if args.filePath points to a 10 GB file, this streams all of it

  // Transform that calls an LLM for each chunk — very slow consumer
  const transform = new LLMProcessingTransform();

  // The writable destination
  const dest = createWriteStream('/var/app/output/result.txt');

  // If the LLM processing is slow, the internal buffers grow without bound
  await pipeline(src, transform, dest);
  return { content: [{ type: 'text', text: 'Done' }] };
});

// What happens with a 10 GB file:
// 1. createReadStream starts reading at 64 KB/read (default highWaterMark)
// 2. LLMProcessingTransform is slow — each chunk takes 500ms for LLM call
// 3. src continues to push data because pipeline's back-pressure
//    propagation depends on all stages correctly implementing pause/resume
// 4. Even with back-pressure, if any stage drops the signal, data accumulates
// 5. heap grows → V8 GC thrashes → OOM kill or process crash

// Also vulnerable: PassThrough with no size limit
import { PassThrough } from 'stream';
const buffer = new PassThrough();
request.pipe(buffer);  // if request is huge, PassThrough buffers it all

// Also vulnerable: async iterator without size tracking
let totalBytes = 0;
for await (const chunk of readableStream) {
  totalBytes += chunk.length;
  // No check on totalBytes — processes the entire input regardless of size
  await processChunk(chunk);
}

// SECURE: explicit highWaterMark + total bytes limit + abort on excess
import { createReadStream } from 'fs';
import { pipeline } from 'stream/promises';
import { Transform } from 'stream';
import { stat } from 'fs/promises';

const MAX_FILE_BYTES = 50 * 1024 * 1024; // 50 MB hard limit

server.setToolHandler('process_file', async (args) => {
  const filePath = String(args.filePath ?? '');

  // Check file size BEFORE opening the stream
  const { size } = await stat(filePath);
  if (size > MAX_FILE_BYTES) {
    throw new Error(`File too large: ${size} bytes (max ${MAX_FILE_BYTES})`);
  }

  // Set explicit highWaterMark — 1 MB read buffer
  const src = createReadStream(filePath, { highWaterMark: 1024 * 1024 });

  // Size-tracking guard transform — aborts if data exceeds limit mid-stream
  let bytesRead = 0;
  const sizeGuard = new Transform({
    highWaterMark: 1024 * 1024,
    transform(chunk, _enc, callback) {
      bytesRead += chunk.length;
      if (bytesRead > MAX_FILE_BYTES) {
        callback(new Error(`Stream exceeded size limit at ${bytesRead} bytes`));
        return;
      }
      callback(null, chunk);
    }
  });

  const transform = new LLMProcessingTransform({ highWaterMark: 1024 * 256 });
  const dest = createWriteStream('/var/app/output/result.txt');

  // pipeline propagates back-pressure and errors through all stages
  await pipeline(src, sizeGuard, transform, dest);
  return { content: [{ type: 'text', text: 'Done' }] };
});

// DETECTION (SkillAudit SA-SKL-STREAM-001):
// createReadStream(args.*) with no highWaterMark option
// OR new Readable({ }) without highWaterMark when data source is user-controlled
// OR for-await loop over user-controlled stream with no bytesRead tracking
const stream001 = searchAST(
  'createReadStream(args.*, { }) missing highWaterMark ' +
  'OR createReadStream(args.*) without options ' +
  'OR for-await over user-stream without size tracking'
);
if (stream001.length > 0) => HIGH  // SA-SKL-STREAM-001

SA-SKL-STREAM-002: Transform stream with synchronous _transform() that never calls callback — back-pressure removed

Node.js Transform streams implement back-pressure through the callback parameter of _transform(chunk, encoding, callback). The back-pressure contract is: the Transform’s writable side will not accept another chunk from the producer until callback() is called, signalling that the current chunk has been processed or queued for processing. When _transform() performs synchronous work and calls callback() immediately, back-pressure is correctly applied: the producer is limited to one chunk at a time. When _transform() defers calling callback() until some later condition is met — accumulating N chunks before flushing, waiting for an external event, or a programming error that omits the callback call in one code path — the back-pressure mechanism is removed. The writable side of the Transform continues to accept data, buffering it without limit.

This failure mode is frequently introduced by Transform implementations that batch chunks for efficiency: the transform accumulates K chunks in an internal array and processes them together when the batch is full. If the batch-full condition is only checked at the end of _transform() and callback() is only called when the batch flushes, the producer sends K chunks with no pausing and the next K chunks while the batch is being processed. For large inputs, this means the entire input is buffered before any processing begins — the antithesis of streaming.

// VULNERABLE: _transform() defers callback until batch is full — back-pressure removed
import { Transform } from 'stream';

class BatchingTransform extends Transform {
  private batch: Buffer[] = [];
  private batchSize = 10;

  _transform(chunk: Buffer, _enc: string, callback: Function) {
    this.batch.push(chunk);

    if (this.batch.length >= this.batchSize) {
      // Process the batch synchronously
      const combined = Buffer.concat(this.batch);
      this.push(combined);
      this.batch = [];
      callback();  // ← callback called ONLY when batch is full
      return;
    }

    // For the first 9 chunks, callback is NEVER called
    // The Transform's writable side continues accepting data from the producer
    // Producer sends all chunks without pausing — no back-pressure applied
    // If input has 10,000 chunks, 10,000 chunks are buffered before any flush
  }

  _flush(callback: Function) {
    if (this.batch.length > 0) {
      this.push(Buffer.concat(this.batch));
      this.batch = [];
    }
    callback();
  }
}

// Also vulnerable: async _transform() that awaits before calling callback
// without a finally block — if the await throws, callback is never called
class AsyncTransform extends Transform {
  async _transform(chunk: Buffer, _enc: string, callback: Function) {
    try {
      const result = await processChunkWithLLM(chunk);
      this.push(result);
      callback();  // ← if processChunkWithLLM throws, callback never runs
    } catch (err) {
      // Missing: callback(err) — back-pressure permanently removed on error
      console.error('Transform error:', err);
      // The stream appears to stall — no more data flows, no error emitted
    }
  }
}

// SECURE: callback() always called — once per chunk, immediately or in finally
import { Transform, TransformCallback } from 'stream';

class SafeBatchingTransform extends Transform {
  private batch: Buffer[] = [];
  private batchSize = 10;
  private totalBytes = 0;
  private readonly MAX_BYTES = 50 * 1024 * 1024; // 50 MB total

  _transform(chunk: Buffer, _enc: string, callback: TransformCallback) {
    this.totalBytes += chunk.length;
    if (this.totalBytes > this.MAX_BYTES) {
      callback(new Error(`Input exceeded size limit: ${this.totalBytes} bytes`));
      return;
    }

    this.batch.push(chunk);

    if (this.batch.length >= this.batchSize) {
      const combined = Buffer.concat(this.batch);
      this.batch = [];
      this.push(combined);
    }

    // callback() called EVERY time — back-pressure correctly applied
    // Producer may only send the next chunk after this call
    callback();
  }

  _flush(callback: TransformCallback) {
    if (this.batch.length > 0) {
      this.push(Buffer.concat(this.batch));
      this.batch = [];
    }
    callback();
  }
}

// Safe async _transform: always call callback() in finally block
class SafeAsyncTransform extends Transform {
  _transform(chunk: Buffer, _enc: string, callback: TransformCallback) {
    // Use the non-async form with a self-invoking async to ensure finally runs
    (async () => {
      const result = await processChunkWithLLM(chunk);
      this.push(result);
    })()
      .then(() => callback())     // success path
      .catch((err) => callback(err));  // error path — always calls callback
  }
}

// DETECTION (SkillAudit SA-SKL-STREAM-002):
// _transform() method body with conditional callback() — not all paths call callback
// OR async _transform() without .catch(callback) or finally { callback() }
const stream002 = searchAST(
  '_transform with conditional callback() — not called on every code path ' +
  'OR async _transform without catch(callback) or finally callback'
);
if (stream002.length > 0) => HIGH  // SA-SKL-STREAM-002

SA-SKL-STREAM-003: pipeline() without error handling — file handles and connections leaked

Node.js’s stream.pipeline() was introduced specifically to address the resource leak problem with the original stream.pipe() API. When pipe() encounters an error in any stream, it does not automatically destroy the other streams in the pipeline — each stream must have its own error event handler, and the handler must explicitly call destroy() on the other streams. Missing error handlers mean that a failing read stream leaves the write stream open; a failing transform leaves both the upstream readable and the downstream writable open. Open file descriptors, open TCP sockets, and open HTTP response streams all leak, accumulating until the operating system’s per-process file descriptor limit is reached, at which point the MCP skill process can no longer open new files or network connections.

stream.pipeline() fixes this by automatically destroying all streams when one encounters an error. However, it only does so when the error is properly propagated through the pipeline’s callback or Promise rejection. Skills that call pipeline() but do not await the returned Promise, do not attach a .catch() handler, or silently swallow errors in the callback still leak the underlying resources because the error is not propagated to Node.js’s unhandled rejection handling, and the cleanup logic that pipeline() would perform on proper error propagation is never triggered for the other streams.

// VULNERABLE: pipeline() without error handling — resources leaked on stream error
import { pipeline } from 'stream';
import { createReadStream, createWriteStream } from 'fs';
import { promisify } from 'util';

// Pattern 1: pipeline() with callback that ignores errors
server.setToolHandler('process_file', async (args) => {
  const src = createReadStream(args.filePath);
  const transform = new DataTransform();
  const dest = createWriteStream('/var/app/output/result.txt');

  // Callback form — error ignored, resources not cleaned up
  pipeline(src, transform, dest, (err) => {
    if (err) {
      console.error('Pipeline error:', err);
      // ← No destroy() calls — src, transform, dest remain open
      // ← File descriptor for src and dest remain open
      // ← dest may be partially written, file handle leaked
    }
    // ← No resolve/reject — caller awaits nothing, error is swallowed
  });

  // Caller immediately returns — pipeline runs in background
  // If pipeline errors, the skill's tool response is never sent
  // The skill hangs from the caller's perspective, and FDs leak
  return { content: [{ type: 'text', text: 'Processing started' }] };
});

// Pattern 2: promises pipeline() not awaited — errors become unhandled rejections
import { pipeline as pipelineAsync } from 'stream/promises';

server.setToolHandler('process_file_2', async (args) => {
  const src = createReadStream(args.filePath);
  const dest = createWriteStream('/var/app/output/result.txt');

  // pipeline() returns a Promise but it is NOT awaited
  // An error in src or dest becomes an unhandled Promise rejection
  // Node.js logs it but does not destroy the other stream
  pipelineAsync(src, dest);  // ← missing await

  return { content: [{ type: 'text', text: 'Done' }] };
});

// Pattern 3: old pipe() API with no error listeners — most dangerous form
server.setToolHandler('process_file_3', async (args) => {
  const src = createReadStream(args.filePath);
  const dest = createWriteStream('/tmp/output');

  // pipe() does NOT auto-destroy on error — each stream needs error handler
  src.pipe(dest);  // ← no error handlers on src or dest
  // If src emits 'error', dest stays open; if dest emits 'error', src stays open
  // Each failed request leaks one or two file descriptors
  // After ~1000 failed requests, process hits ulimit and cannot open new files
});

// SECURE: await pipeline() in try/catch; or use stream.finished() for monitoring
import { pipeline } from 'stream/promises';
import { createReadStream, createWriteStream } from 'fs';
import { finished } from 'stream/promises';

server.setToolHandler('process_file', async (args) => {
  const filePath = resolveAndValidatePath(args.filePath);
  const src = createReadStream(filePath, { highWaterMark: 1024 * 1024 });
  const transform = new DataTransform({ highWaterMark: 1024 * 256 });
  const dest = createWriteStream('/var/app/output/result.txt');

  try {
    // await pipeline() propagates errors AND destroys all streams on error
    // Cleanup happens automatically — no file descriptor leaks
    await pipeline(src, transform, dest);
  } catch (err) {
    // All three streams have been destroyed by pipeline() at this point
    // Log the error with context for observability
    throw new Error(`File processing failed: ${(err as Error).message}`);
  }

  return { content: [{ type: 'text', text: 'Processed successfully' }] };
});

// Using stream.finished() to detect premature close and ensure cleanup
import { finished } from 'stream/promises';
async function safeStreamProcess(src: Readable, dest: Writable): Promise<void> {
  src.pipe(dest);
  try {
    await finished(dest);  // resolves when dest closes normally
  } catch (err) {
    // Premature close or error — destroy both streams
    src.destroy();
    dest.destroy();
    throw err;
  }
}

// DETECTION (SkillAudit SA-SKL-STREAM-003):
// pipeline(src, dest, callback) where callback does not call destroy on error
// OR pipelineAsync(src, dest) without await or .catch()
// OR src.pipe(dest) without error event listeners on both streams
const stream003 = searchAST(
  'pipeline(*) without await or .catch() ' +
  'OR pipeline(*) with callback that does not call destroy() on error ' +
  'OR .pipe() without .on("error", ...) on both streams'
);
if (stream003.length > 0) => HIGH  // SA-SKL-STREAM-003

SA-SKL-STREAM-004: Duplex stream passed as tool argument — attacker-controlled stream stalls pipeline

Node.js Duplex streams implement both the Readable and Writable interfaces. They are legitimate objects for representing bidirectional I/O channels like TCP sockets or TLS streams. Some MCP skills accept a stream object as a tool argument — for example, a skill that wraps an external streaming API might accept a stream argument representing the data source. When the skill then pipes this stream through an internal transform pipeline, the attacker has full control over the stream’s behaviour because the stream object they provide can implement any logic they choose. A malicious Duplex implementation that never emits the end event keeps the pipeline open forever: the downstream transform waits for more data that never arrives, the request handler awaiting the pipeline never resolves, the connection to the MCP client times out or is closed by the client, and the skill’s process holds an open pipeline, an open output stream, and all associated memory indefinitely.

This vulnerability is a class of resource exhaustion rather than information disclosure or code execution, but in the MCP context it is severe: a single attacker-controlled stream that never ends occupies one pipeline slot, one or more file descriptors, and the associated memory. An attacker who issues many concurrent tool calls each with a never-ending stream can exhaust the MCP skill’s connection pool, gorge heap with buffered data, and deny service to all legitimate users. The fix is categorical: MCP skills must never accept stream objects from tool arguments. Tool arguments are JSON-serialisable values; streams are not. Any skill that accepts a stream from arguments is either incorrectly architectured (it should accept a URL or a bounded Buffer) or exposes this vulnerability.

// VULNERABLE: accepting a stream object from tool arguments
import { pipeline } from 'stream/promises';

server.setToolHandler('process_stream', async (args) => {
  // args.dataStream is a Duplex stream provided by the caller
  // The caller (attacker) controls the stream implementation entirely
  const dataStream = args.dataStream as Duplex;

  const transform = new DataTransform();
  const dest = createWriteStream('/var/app/output/result.txt');

  // pipeline awaits the stream to emit 'end' (readable side) or 'finish' (writable)
  // Attacker's Duplex never emits 'end' — pipeline awaits forever
  await pipeline(dataStream, transform, dest);
  // ↑ This Promise never resolves — the tool handler is permanently suspended

  return { content: [{ type: 'text', text: 'Done' }] };
  // ↑ This return is never reached
});

// How the attacker's never-ending Duplex is constructed:
class MaliciousDuplex extends Duplex {
  _read(_size: number) {
    // Emit some data to keep the pipeline alive
    this.push(Buffer.alloc(1024, 0x41));
    // Never call this.push(null) — 'end' is never emitted
    // The readable side remains open forever
  }
  _write(chunk: any, _enc: string, callback: Function) {
    callback();  // Accept writes normally — appears benign
  }
}

// The attacker can also:
// 1. Emit data very slowly — one byte per hour — to prevent timeout detection
// 2. Emit enough data to fill the transform's buffer, stalling the pipeline
//    from the consumer side rather than the producer side
// 3. Emit 'error' at an unexpected time to trigger the leak pattern from STREAM-003
//    if error handling is also absent

// Even with a timeout, the leak window allows significant resource consumption:
const TIMEOUT = 30_000;
const timeoutHandle = setTimeout(() => dataStream.destroy(), TIMEOUT);
await pipeline(dataStream, transform, dest);
clearTimeout(timeoutHandle);
// → with 100 concurrent requests and 30s timeout: 100 × 30s of open FDs and heap

// SECURE: never accept stream objects; accept Buffer/string with size limit,
// or a URL the skill fetches internally with its own size limit
import { pipeline } from 'stream/promises';
import { Readable } from 'stream';

const MAX_INPUT_BYTES = 10 * 1024 * 1024; // 10 MB

server.setToolHandler('process_data', async (args) => {
  // Accept raw bytes as base64 string, not a stream object
  const inputBase64 = String(args.inputData ?? '');

  // Validate size before decoding — base64 is ~4/3 the binary size
  if (inputBase64.length > Math.ceil(MAX_INPUT_BYTES * 1.34)) {
    throw new Error('Input data too large');
  }

  const inputBuffer = Buffer.from(inputBase64, 'base64');
  if (inputBuffer.length > MAX_INPUT_BYTES) {
    throw new Error('Decoded input exceeds size limit');
  }

  // Wrap the validated buffer in a Readable — we control the stream, not the caller
  const src = Readable.from(inputBuffer, { highWaterMark: 1024 * 256 });
  const transform = new DataTransform({ highWaterMark: 1024 * 256 });
  const dest = createWriteStream('/var/app/output/result.txt');

  // We control both ends of the pipeline — no attacker stream objects
  await pipeline(src, transform, dest);
  return { content: [{ type: 'text', text: 'Processed successfully' }] };
});

// Alternative: accept a URL, fetch it internally with a size limit
server.setToolHandler('process_url', async (args) => {
  const url = validateUrl(String(args.url ?? ''));  // allowlist or strict parse

  // Skill fetches the URL itself — attacker does not control the stream object
  const response = await fetch(url);
  if (!response.ok) throw new Error(`Fetch failed: ${response.status}`);

  const contentLength = Number(response.headers.get('content-length') ?? 0);
  if (contentLength > MAX_INPUT_BYTES) {
    throw new Error('Content-Length exceeds limit');
  }

  // Read with enforced size limit — Readable.fromWeb with byte counter
  const buffer = await readWithLimit(response.body, MAX_INPUT_BYTES);
  const src = Readable.from(buffer, { highWaterMark: 1024 * 256 });
  const transform = new DataTransform({ highWaterMark: 1024 * 256 });
  const dest = createWriteStream('/var/app/output/result.txt');

  await pipeline(src, transform, dest);
  return { content: [{ type: 'text', text: 'Processed successfully' }] };
});

// DETECTION (SkillAudit SA-SKL-STREAM-004):
// args.* used as Readable/Duplex/stream in pipeline() or .pipe()
// OR args.* cast to Readable/Duplex type and passed to stream consumers
const stream004 = searchAST(
  'pipeline(args.*, ...) OR args.*.pipe(*) ' +
  'OR (args.* as Readable) OR (args.* as Duplex) ' +
  'OR args.* passed to pipeline() positional argument'
);
if (stream004.length > 0) => HIGH  // SA-SKL-STREAM-004

Remediation summary

SkillAudit detects all four SA-SKL-STREAM-001 through STREAM-004 Node.js stream security patterns through static analysis of stream constructor calls, Transform method implementations, pipeline invocation patterns, and argument types passed to stream consumers. Related resource exhaustion patterns: SA-SKL-RCL connection pool exhaustion covers how unbounded connections and missing pool limits interact with stream-heavy workloads. See the complete 15-family SA-SKL attack surface reference for the full taxonomy of MCP skill security findings.