+
+
+
+In Node.js, calling the pipe() method on a stream without proper error handling can lead to unexplained failures, where errors are dropped and not propagated downstream. This can result in unwanted behavior and make debugging difficult. To reliably handle all errors, every stream in the pipeline must have an error handler registered.
+
+
+
+
+
+Instead of using pipe() with manual error handling, prefer using the pipeline function from the Node.js stream module. The pipeline function automatically handles errors and ensures proper cleanup of resources. This approach is more robust and eliminates the risk of forgetting to handle errors.
+
+
+If you must use pipe(), always attach an error handler to the source stream using methods like on('error', handler) to ensure that any errors emitted by the input stream are properly handled. When multiple pipe() calls are chained, an error handler should be attached before each step of the pipeline.
+
+
+
+
+
+The following code snippet demonstrates a problematic usage of the pipe() method without error handling:
+
+
+
+
+
+A better approach is to use the pipeline function, which automatically handles errors:
+
+
+
+
+
+Alternatively, if you need to use pipe(), make sure to add error handling:
+
+
+
+
+
+
+Node.js Documentation: stream.pipeline().
+
+
diff --git a/javascript/ql/src/Quality/UnhandledErrorInStreamPipeline.ql b/javascript/ql/src/Quality/UnhandledErrorInStreamPipeline.ql
new file mode 100644
index 000000000000..a6142a2e6e73
--- /dev/null
+++ b/javascript/ql/src/Quality/UnhandledErrorInStreamPipeline.ql
@@ -0,0 +1,303 @@
+/**
+ * @id js/unhandled-error-in-stream-pipeline
+ * @name Unhandled error in stream pipeline
+ * @description Calling `pipe()` on a stream without error handling will drop errors coming from the input stream
+ * @kind problem
+ * @problem.severity warning
+ * @precision high
+ * @tags quality
+ * maintainability
+ * error-handling
+ * frameworks/nodejs
+ */
+
+import javascript
+import semmle.javascript.filters.ClassifyFiles
+
+/**
+ * A call to the `pipe` method on a Node.js stream.
+ */
+class PipeCall extends DataFlow::MethodCallNode {
+ PipeCall() {
+ this.getMethodName() = "pipe" and
+ this.getNumArgument() = [1, 2] and
+ not this.getArgument([0, 1]).asExpr() instanceof Function and
+ not this.getArgument(0).asExpr() instanceof ObjectExpr and
+ not this.getArgument(0).getALocalSource() = getNonNodeJsStreamType()
+ }
+
+ /** Gets the source stream (receiver of the pipe call). */
+ DataFlow::Node getSourceStream() { result = this.getReceiver() }
+
+ /** Gets the destination stream (argument of the pipe call). */
+ DataFlow::Node getDestinationStream() { result = this.getArgument(0) }
+}
+
+/**
+ * Gets a reference to a value that is known to not be a Node.js stream.
+ * This is used to exclude pipe calls on non-stream objects from analysis.
+ */
+private DataFlow::Node getNonNodeJsStreamType() {
+ result = getNonStreamApi().getAValueReachableFromSource()
+}
+
+/**
+ * Gets API nodes from modules that are known to not provide Node.js streams.
+ * This includes reactive programming libraries, frontend frameworks, and other non-stream APIs.
+ */
+private API::Node getNonStreamApi() {
+ exists(string moduleName |
+ moduleName
+ .regexpMatch([
+ "rxjs(|/.*)", "@strapi(|/.*)", "highland(|/.*)", "execa(|/.*)", "arktype(|/.*)",
+ "@ngrx(|/.*)", "@datorama(|/.*)", "@angular(|/.*)", "react.*", "@langchain(|/.*)",
+ ]) and
+ result = API::moduleImport(moduleName)
+ )
+ or
+ result = getNonStreamApi().getAMember()
+ or
+ result = getNonStreamApi().getAParameter().getAParameter()
+ or
+ result = getNonStreamApi().getReturn()
+ or
+ result = getNonStreamApi().getPromised()
+}
+
+/**
+ * Gets the method names used to register event handlers on Node.js streams.
+ * These methods are used to attach handlers for events like `error`.
+ */
+private string getEventHandlerMethodName() { result = ["on", "once", "addListener"] }
+
+/**
+ * Gets the method names that are chainable on Node.js streams.
+ */
+private string getChainableStreamMethodName() {
+ result =
+ [
+ "setEncoding", "pause", "resume", "unpipe", "destroy", "cork", "uncork", "setDefaultEncoding",
+ "off", "removeListener", getEventHandlerMethodName()
+ ]
+}
+
+/**
+ * Gets the method names that are not chainable on Node.js streams.
+ */
+private string getNonchainableStreamMethodName() {
+ result = ["read", "write", "end", "pipe", "unshift", "push", "isPaused", "wrap", "emit"]
+}
+
+/**
+ * Gets the property names commonly found on Node.js streams.
+ */
+private string getStreamPropertyName() {
+ result =
+ [
+ "readable", "writable", "destroyed", "closed", "readableHighWaterMark", "readableLength",
+ "readableObjectMode", "readableEncoding", "readableFlowing", "readableEnded", "flowing",
+ "writableHighWaterMark", "writableLength", "writableObjectMode", "writableFinished",
+ "writableCorked", "writableEnded", "defaultEncoding", "allowHalfOpen", "objectMode",
+ "errored", "pending", "autoDestroy", "encoding", "path", "fd", "bytesRead", "bytesWritten",
+ "_readableState", "_writableState"
+ ]
+}
+
+/**
+ * Gets all method names commonly found on Node.js streams.
+ */
+private string getStreamMethodName() {
+ result = [getChainableStreamMethodName(), getNonchainableStreamMethodName()]
+}
+
+/**
+ * A call to register an event handler on a Node.js stream.
+ * This includes methods like `on`, `once`, and `addListener`.
+ */
+class ErrorHandlerRegistration extends DataFlow::MethodCallNode {
+ ErrorHandlerRegistration() {
+ this.getMethodName() = getEventHandlerMethodName() and
+ this.getArgument(0).getStringValue() = "error"
+ }
+}
+
+/**
+ * Holds if the stream in `node1` will propagate to `node2`.
+ */
+private predicate streamFlowStep(DataFlow::Node node1, DataFlow::Node node2) {
+ exists(PipeCall pipe |
+ node1 = pipe.getDestinationStream() and
+ node2 = pipe
+ )
+ or
+ exists(DataFlow::MethodCallNode chainable |
+ chainable.getMethodName() = getChainableStreamMethodName() and
+ node1 = chainable.getReceiver() and
+ node2 = chainable
+ )
+}
+
+/**
+ * Tracks the result of a pipe call as it flows through the program.
+ */
+private DataFlow::SourceNode destinationStreamRef(DataFlow::TypeTracker t, PipeCall pipe) {
+ t.start() and
+ (result = pipe or result = pipe.getDestinationStream().getALocalSource())
+ or
+ exists(DataFlow::SourceNode prev |
+ prev = destinationStreamRef(t.continue(), pipe) and
+ streamFlowStep(prev, result)
+ )
+ or
+ exists(DataFlow::TypeTracker t2 | result = destinationStreamRef(t2, pipe).track(t2, t))
+}
+
+/**
+ * Gets a reference to the result of a pipe call.
+ */
+private DataFlow::SourceNode destinationStreamRef(PipeCall pipe) {
+ result = destinationStreamRef(DataFlow::TypeTracker::end(), pipe)
+}
+
+/**
+ * Holds if the pipe call result is used to call a non-stream method.
+ * Since pipe() returns the destination stream, this finds cases where
+ * the destination stream is used with methods not typical of streams.
+ */
+private predicate isPipeFollowedByNonStreamMethod(PipeCall pipeCall) {
+ exists(DataFlow::MethodCallNode call |
+ call = destinationStreamRef(pipeCall).getAMethodCall() and
+ not call.getMethodName() = getStreamMethodName()
+ )
+}
+
+/**
+ * Holds if the pipe call result is used to access a property that is not typical of streams.
+ */
+private predicate isPipeFollowedByNonStreamProperty(PipeCall pipeCall) {
+ exists(DataFlow::PropRef propRef |
+ propRef = destinationStreamRef(pipeCall).getAPropertyRead() and
+ not propRef.getPropertyName() = [getStreamPropertyName(), getStreamMethodName()]
+ )
+}
+
+/**
+ * Holds if the pipe call result is used in a non-stream-like way,
+ * either by calling non-stream methods or accessing non-stream properties.
+ */
+private predicate isPipeFollowedByNonStreamAccess(PipeCall pipeCall) {
+ isPipeFollowedByNonStreamMethod(pipeCall) or
+ isPipeFollowedByNonStreamProperty(pipeCall)
+}
+
+/**
+ * Gets a reference to a stream that may be the source of the given pipe call.
+ * Uses type back-tracking to trace stream references in the data flow.
+ */
+private DataFlow::SourceNode sourceStreamRef(DataFlow::TypeBackTracker t, PipeCall pipeCall) {
+ t.start() and
+ result = pipeCall.getSourceStream().getALocalSource()
+ or
+ exists(DataFlow::SourceNode prev |
+ prev = sourceStreamRef(t.continue(), pipeCall) and
+ streamFlowStep(result.getALocalUse(), prev)
+ )
+ or
+ exists(DataFlow::TypeBackTracker t2 | result = sourceStreamRef(t2, pipeCall).backtrack(t2, t))
+}
+
+/**
+ * Gets a reference to a stream that may be the source of the given pipe call.
+ */
+private DataFlow::SourceNode sourceStreamRef(PipeCall pipeCall) {
+ result = sourceStreamRef(DataFlow::TypeBackTracker::end(), pipeCall)
+}
+
+/**
+ * Holds if the source stream of the given pipe call has an `error` handler registered.
+ */
+private predicate hasErrorHandlerRegistered(PipeCall pipeCall) {
+ exists(DataFlow::Node stream |
+ stream = sourceStreamRef(pipeCall).getALocalUse() and
+ (
+ stream.(DataFlow::SourceNode).getAMethodCall(_) instanceof ErrorHandlerRegistration
+ or
+ exists(DataFlow::SourceNode base, string propName |
+ stream = base.getAPropertyRead(propName) and
+ base.getAPropertyRead(propName).getAMethodCall(_) instanceof ErrorHandlerRegistration
+ )
+ or
+ exists(DataFlow::PropWrite propWrite, DataFlow::SourceNode instance |
+ propWrite.getRhs().getALocalSource() = stream and
+ instance = propWrite.getBase().getALocalSource() and
+ instance.getAPropertyRead(propWrite.getPropertyName()).getAMethodCall(_) instanceof
+ ErrorHandlerRegistration
+ )
+ )
+ )
+ or
+ hasPlumber(pipeCall)
+}
+
+/**
+ * Holds if the pipe call uses `gulp-plumber`, which automatically handles stream errors.
+ * `gulp-plumber` returns a stream that uses monkey-patching to ensure all subsequent streams in the pipeline propagate their errors.
+ */
+private predicate hasPlumber(PipeCall pipeCall) {
+ pipeCall.getDestinationStream().getALocalSource() = API::moduleImport("gulp-plumber").getACall()
+ or
+ sourceStreamRef+(pipeCall) = API::moduleImport("gulp-plumber").getACall()
+}
+
+/**
+ * Holds if the source or destination of the given pipe call is identified as a non-Node.js stream.
+ */
+private predicate hasNonNodeJsStreamSource(PipeCall pipeCall) {
+ sourceStreamRef(pipeCall) = getNonNodeJsStreamType() or
+ destinationStreamRef(pipeCall) = getNonNodeJsStreamType()
+}
+
+/**
+ * Holds if the source stream of the given pipe call is used in a non-stream-like way.
+ */
+private predicate hasNonStreamSourceLikeUsage(PipeCall pipeCall) {
+ exists(DataFlow::MethodCallNode call, string name |
+ call.getReceiver().getALocalSource() = sourceStreamRef(pipeCall) and
+ name = call.getMethodName() and
+ not name = getStreamMethodName()
+ )
+ or
+ exists(DataFlow::PropRef propRef, string propName |
+ propRef.getBase().getALocalSource() = sourceStreamRef(pipeCall) and
+ propName = propRef.getPropertyName() and
+ not propName = [getStreamPropertyName(), getStreamMethodName()]
+ )
+}
+
+/**
+ * Holds if the pipe call destination stream has an error handler registered.
+ */
+private predicate hasErrorHandlerDownstream(PipeCall pipeCall) {
+ exists(DataFlow::SourceNode stream |
+ stream = destinationStreamRef(pipeCall) and
+ (
+ exists(ErrorHandlerRegistration handler | handler.getReceiver().getALocalSource() = stream)
+ or
+ exists(DataFlow::SourceNode base, string propName |
+ stream = base.getAPropertyRead(propName) and
+ base.getAPropertyRead(propName).getAMethodCall(_) instanceof ErrorHandlerRegistration
+ )
+ )
+ )
+}
+
+from PipeCall pipeCall
+where
+ not hasErrorHandlerRegistered(pipeCall) and
+ hasErrorHandlerDownstream(pipeCall) and
+ not isPipeFollowedByNonStreamAccess(pipeCall) and
+ not hasNonStreamSourceLikeUsage(pipeCall) and
+ not hasNonNodeJsStreamSource(pipeCall) and
+ not isTestFile(pipeCall.getFile())
+select pipeCall,
+ "Stream pipe without error handling on the source stream. Errors won't propagate downstream and may be silently dropped."
diff --git a/javascript/ql/src/Quality/examples/UnhandledStreamPipe.js b/javascript/ql/src/Quality/examples/UnhandledStreamPipe.js
new file mode 100644
index 000000000000..95c1661a8b9f
--- /dev/null
+++ b/javascript/ql/src/Quality/examples/UnhandledStreamPipe.js
@@ -0,0 +1,8 @@
+const fs = require('fs');
+const source = fs.createReadStream('source.txt');
+const destination = fs.createWriteStream('destination.txt');
+
+// Bad: Only destination has error handling, source errors are unhandled
+source.pipe(destination).on('error', (err) => {
+ console.error('Destination error:', err);
+});
diff --git a/javascript/ql/src/Quality/examples/UnhandledStreamPipeGood.js b/javascript/ql/src/Quality/examples/UnhandledStreamPipeGood.js
new file mode 100644
index 000000000000..08b9b2a1aab5
--- /dev/null
+++ b/javascript/ql/src/Quality/examples/UnhandledStreamPipeGood.js
@@ -0,0 +1,17 @@
+const { pipeline } = require('stream');
+const fs = require('fs');
+const source = fs.createReadStream('source.txt');
+const destination = fs.createWriteStream('destination.txt');
+
+// Good: Using pipeline for automatic error handling
+pipeline(
+ source,
+ destination,
+ (err) => {
+ if (err) {
+ console.error('Pipeline failed:', err);
+ } else {
+ console.log('Pipeline succeeded');
+ }
+ }
+);
diff --git a/javascript/ql/src/Quality/examples/UnhandledStreamPipeManualError.js b/javascript/ql/src/Quality/examples/UnhandledStreamPipeManualError.js
new file mode 100644
index 000000000000..113bc8117746
--- /dev/null
+++ b/javascript/ql/src/Quality/examples/UnhandledStreamPipeManualError.js
@@ -0,0 +1,16 @@
+const fs = require('fs');
+const source = fs.createReadStream('source.txt');
+const destination = fs.createWriteStream('destination.txt');
+
+// Alternative Good: Manual error handling with pipe()
+source.on('error', (err) => {
+ console.error('Source stream error:', err);
+ destination.destroy(err);
+});
+
+destination.on('error', (err) => {
+ console.error('Destination stream error:', err);
+ source.destroy(err);
+});
+
+source.pipe(destination);
diff --git a/javascript/ql/test/query-tests/Quality/UnhandledErrorInStreamPipeline/UnhandledErrorInStreamPipeline.expected b/javascript/ql/test/query-tests/Quality/UnhandledErrorInStreamPipeline/UnhandledErrorInStreamPipeline.expected
new file mode 100644
index 000000000000..1b01574112d5
--- /dev/null
+++ b/javascript/ql/test/query-tests/Quality/UnhandledErrorInStreamPipeline/UnhandledErrorInStreamPipeline.expected
@@ -0,0 +1,19 @@
+| test.js:4:5:4:28 | stream. ... nation) | Stream pipe without error handling on the source stream. Errors won't propagate downstream and may be silently dropped. |
+| test.js:19:5:19:17 | s2.pipe(dest) | Stream pipe without error handling on the source stream. Errors won't propagate downstream and may be silently dropped. |
+| test.js:45:5:45:30 | stream2 ... ation2) | Stream pipe without error handling on the source stream. Errors won't propagate downstream and may be silently dropped. |
+| test.js:60:5:60:30 | stream2 ... ation2) | Stream pipe without error handling on the source stream. Errors won't propagate downstream and may be silently dropped. |
+| test.js:66:5:66:21 | stream.pipe(dest) | Stream pipe without error handling on the source stream. Errors won't propagate downstream and may be silently dropped. |
+| test.js:79:5:79:25 | s2.pipe ... ation2) | Stream pipe without error handling on the source stream. Errors won't propagate downstream and may be silently dropped. |
+| test.js:94:5:94:21 | stream.pipe(dest) | Stream pipe without error handling on the source stream. Errors won't propagate downstream and may be silently dropped. |
+| test.js:110:11:110:22 | s.pipe(dest) | Stream pipe without error handling on the source stream. Errors won't propagate downstream and may be silently dropped. |
+| test.js:119:5:119:21 | stream.pipe(dest) | Stream pipe without error handling on the source stream. Errors won't propagate downstream and may be silently dropped. |
+| test.js:128:5:128:26 | getStre ... e(dest) | Stream pipe without error handling on the source stream. Errors won't propagate downstream and may be silently dropped. |
+| test.js:146:5:146:62 | stream. ... itable) | Stream pipe without error handling on the source stream. Errors won't propagate downstream and may be silently dropped. |
+| test.js:182:17:182:40 | notStre ... itable) | Stream pipe without error handling on the source stream. Errors won't propagate downstream and may be silently dropped. |
+| test.js:192:5:192:32 | copyStr ... nation) | Stream pipe without error handling on the source stream. Errors won't propagate downstream and may be silently dropped. |
+| tst.js:8:5:8:21 | source.pipe(gzip) | Stream pipe without error handling on the source stream. Errors won't propagate downstream and may be silently dropped. |
+| tst.js:37:21:37:56 | wrapper ... Stream) | Stream pipe without error handling on the source stream. Errors won't propagate downstream and may be silently dropped. |
+| tst.js:44:5:44:40 | wrapper ... Stream) | Stream pipe without error handling on the source stream. Errors won't propagate downstream and may be silently dropped. |
+| tst.js:52:5:52:37 | source. ... Stream) | Stream pipe without error handling on the source stream. Errors won't propagate downstream and may be silently dropped. |
+| tst.js:59:18:59:39 | stream. ... Stream) | Stream pipe without error handling on the source stream. Errors won't propagate downstream and may be silently dropped. |
+| tst.js:111:5:111:26 | stream. ... Stream) | Stream pipe without error handling on the source stream. Errors won't propagate downstream and may be silently dropped. |
diff --git a/javascript/ql/test/query-tests/Quality/UnhandledErrorInStreamPipeline/UnhandledErrorInStreamPipeline.qlref b/javascript/ql/test/query-tests/Quality/UnhandledErrorInStreamPipeline/UnhandledErrorInStreamPipeline.qlref
new file mode 100644
index 000000000000..0da7b1900f69
--- /dev/null
+++ b/javascript/ql/test/query-tests/Quality/UnhandledErrorInStreamPipeline/UnhandledErrorInStreamPipeline.qlref
@@ -0,0 +1,2 @@
+query: Quality/UnhandledErrorInStreamPipeline.ql
+postprocess: utils/test/InlineExpectationsTestQuery.ql
diff --git a/javascript/ql/test/query-tests/Quality/UnhandledErrorInStreamPipeline/arktype.js b/javascript/ql/test/query-tests/Quality/UnhandledErrorInStreamPipeline/arktype.js
new file mode 100644
index 000000000000..cac5e57511d4
--- /dev/null
+++ b/javascript/ql/test/query-tests/Quality/UnhandledErrorInStreamPipeline/arktype.js
@@ -0,0 +1,3 @@
+import { type } from 'arktype';
+
+type.string.pipe(Number);
diff --git a/javascript/ql/test/query-tests/Quality/UnhandledErrorInStreamPipeline/execa.js b/javascript/ql/test/query-tests/Quality/UnhandledErrorInStreamPipeline/execa.js
new file mode 100644
index 000000000000..052875e849bc
--- /dev/null
+++ b/javascript/ql/test/query-tests/Quality/UnhandledErrorInStreamPipeline/execa.js
@@ -0,0 +1,11 @@
+const execa = require('execa');
+
+(async () => {
+ const first = execa('node', ['empty.js']);
+ const second = execa('node', ['stdin.js']);
+
+ first.stdout.pipe(second.stdin);
+
+ const {stdout} = await second;
+ console.log(stdout);
+})();
diff --git a/javascript/ql/test/query-tests/Quality/UnhandledErrorInStreamPipeline/fizz-pipe.js b/javascript/ql/test/query-tests/Quality/UnhandledErrorInStreamPipeline/fizz-pipe.js
new file mode 100644
index 000000000000..94906ee46b68
--- /dev/null
+++ b/javascript/ql/test/query-tests/Quality/UnhandledErrorInStreamPipeline/fizz-pipe.js
@@ -0,0 +1,29 @@
+import React, { Suspense } from "react";
+import { renderToPipeableStream } from "react-dom/server";
+import { PassThrough } from "stream";
+import { act } from "react-dom/test-utils";
+
+
+const writable = new PassThrough();
+let output = "";
+writable.on("data", chunk => { output += chunk.toString(); });
+writable.on("end", () => { /* stream ended */ });
+
+let errors = [];
+let shellErrors = [];
+
+await act(async () => {
+ renderToPipeableStream(
+