JS: new Quality query - Unhandled errors in .pipe() chain

javascript/ql/integration-tests/query-suite/javascript-code-quality.qls.expected

@@ -2,4 +2,5 @@ ql/javascript/ql/src/Declarations/IneffectiveParameterType.ql
 ql/javascript/ql/src/Expressions/ExprHasNoEffect.ql
 ql/javascript/ql/src/Expressions/MissingAwait.ql
 ql/javascript/ql/src/LanguageFeatures/SpuriousArguments.ql
+ql/javascript/ql/src/Quality/UnhandledStreamPipe.ql
 ql/javascript/ql/src/RegExp/RegExpAlwaysMatches.ql

javascript/ql/integration-tests/query-suite/javascript-security-and-quality.qls.expected

@@ -80,6 +80,7 @@ ql/javascript/ql/src/NodeJS/InvalidExport.ql
 ql/javascript/ql/src/NodeJS/MissingExports.ql
 ql/javascript/ql/src/Performance/PolynomialReDoS.ql
 ql/javascript/ql/src/Performance/ReDoS.ql
+ql/javascript/ql/src/Quality/UnhandledStreamPipe.ql
 ql/javascript/ql/src/React/DirectStateMutation.ql
 ql/javascript/ql/src/React/InconsistentStateUpdate.ql
 ql/javascript/ql/src/React/UnsupportedStateUpdateInLifecycleMethod.ql

javascript/ql/src/Quality/UnhandledStreamPipe.qhelp

@@ -0,0 +1,44 @@
+<!DOCTYPE qhelp PUBLIC
+  "-//Semmle//qhelp//EN"
+  "qhelp.dtd">
+<qhelp>
+
+<overview>
+<p>
+In Node.js, calling the <code>pipe()</code> method on a stream without proper error handling can lead to silent failures, where errors are dropped and not propagated downstream. This can result in unexpected behavior and make debugging difficult. It is crucial to ensure that error handling is implemented when using stream pipes to maintain the reliability of the application.
+</p>
+</overview>
+
+<recommendation>
+<p>
+Instead of using <code>pipe()</code> with manual error handling, prefer using the <code>pipeline</code> function from the Node.js stream module. The <code>pipeline</code> function automatically handles errors and ensures proper cleanup of resources. This approach is more robust and eliminates the risk of forgetting to handle errors.
+</p>
+<p>
+If you must use <code>pipe()</code>, always attach an error handler to the source stream using methods like <code>on('error', handler)</code> to ensure that any errors during the streaming process are properly handled.
+</p>
+</recommendation>
+
+<example>
+<p>
+The following code snippet demonstrates a problematic usage of the <code>pipe()</code> method without error handling:
+</p>
+
+<sample src="examples/UnhandledStreamPipe.js" />
+
+<p>
+A better approach is to use the <code>pipeline</code> function, which automatically handles errors:
+</p>
+
+<sample src="examples/UnhandledStreamPipeGood.js" />
+
+<p>
+Alternatively, if you need to use <code>pipe()</code>, make sure to add error handling:
+</p>
+
+<sample src="examples/UnhandledStreamPipeManualError.js" />
+</example>
+
+<references>
+<li>Node.js Documentation: <a href="https://nodejs.org/api/stream.html#streampipelinestreams-callback">stream.pipeline()</a>.</li>
+</references>
+</qhelp>

javascript/ql/src/Quality/UnhandledStreamPipe.ql

@@ -0,0 +1,281 @@
+/**
+ * @id js/nodejs-stream-pipe-without-error-handling
+ * @name Node.js stream pipe without error handling
+ * @description Calling `pipe()` on a stream without error handling may silently drop errors and prevent proper propagation.
+ * @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"
+  }
+}
+
+/**
+ * Models flow relationships between streams and related operations.
+ * Connects destination streams to their corresponding pipe call nodes.
+ * Connects streams to their chainable methods.
+ */
+private predicate streamFlowStep(DataFlow::Node streamNode, DataFlow::Node relatedNode) {
+  exists(PipeCall pipe |
+    streamNode = pipe.getDestinationStream() and
+    relatedNode = pipe
+  )
+  or
+  exists(DataFlow::MethodCallNode chainable |
+    chainable.getMethodName() = getChainableStreamMethodName() and
+    streamNode = chainable.getReceiver() and
+    relatedNode = 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.getALocalSource()
+  or
+  exists(DataFlow::SourceNode prev |
+    prev = destinationStreamRef(t.continue(), pipe) and
+    streamFlowStep(result.getALocalUse(), prev)
+  )
+  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(ErrorHandlerRegistration handler |
+    handler = sourceStreamRef(pipeCall).getAMethodCall(getEventHandlerMethodName())
+  )
+  or
+  hasPlumber(pipeCall)
+}
+
+/**
+ * Holds if the pipe call uses `gulp-plumber`, which automatically handles stream errors.
+ * Gulp-plumber is a Node.js module that prevents pipe breaking caused by errors from gulp plugins.
+ */
+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(ErrorHandlerRegistration handler |
+    handler.getReceiver().getALocalSource() = destinationStreamRef(pipeCall)
+  )
+}
+
+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."

javascript/ql/src/Quality/examples/UnhandledStreamPipe.js

@@ -0,0 +1,6 @@
+const fs = require('fs');
+const source = fs.createReadStream('source.txt');
+const destination = fs.createWriteStream('destination.txt');
+
+// Bad: No error handling
+source.pipe(destination);

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');
+    }
+  }
+);

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);

javascript/ql/test/query-tests/Quality/UnhandledStreamPipe/arktype.js

@@ -0,0 +1,3 @@
+import { type } from 'arktype';
+
+type.string.pipe(Number);

javascript/ql/test/query-tests/Quality/UnhandledStreamPipe/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);
+})();