Now flags only .pipe calls which have an error somewhere down the stream, but not on the source stream.

javascript/ql/src/Quality/UnhandledStreamPipe.ql

@@ -110,8 +110,11 @@ string getStreamMethodName() {
  * A call to register an event handler on a Node.js stream.
  * This includes methods like `on`, `once`, and `addListener`.
  */
-class StreamEventRegistration extends DataFlow::MethodCallNode {
-  StreamEventRegistration() { this.getMethodName() = getEventHandlerMethodName() }
+class ErrorHandlerRegistration extends DataFlow::MethodCallNode {
+  ErrorHandlerRegistration() {
+    this.getMethodName() = getEventHandlerMethodName() and
+    this.getArgument(0).getStringValue() = "error"
+  }
 }
 
 /**
@@ -136,7 +139,12 @@ predicate streamFlowStep(DataFlow::Node streamNode, DataFlow::Node relatedNode)
  * Tracks the result of a pipe call as it flows through the program.
  */
 private DataFlow::SourceNode pipeResultTracker(DataFlow::TypeTracker t, PipeCall pipe) {
-  t.start() and result = pipe
+  t.start() and result = pipe.getALocalSource()
+  or
+  exists(DataFlow::SourceNode prev |
+    prev = pipeResultTracker(t.continue(), pipe) and
+    streamFlowStep(result.getALocalUse(), prev)
+  )
   or
   exists(DataFlow::TypeTracker t2 | result = pipeResultTracker(t2, pipe).track(t2, t))
 }
@@ -166,7 +174,7 @@ predicate isPipeFollowedByNonStreamMethod(PipeCall pipeCall) {
 predicate isPipeFollowedByNonStreamProperty(PipeCall pipeCall) {
   exists(DataFlow::PropRef propRef |
     propRef = pipeResultRef(pipeCall).getAPropertyRead() and
-    not propRef.getPropertyName() = getStreamPropertyName()
+    not propRef.getPropertyName() = [getStreamPropertyName(), getStreamMethodName()]
   )
 }
 
@@ -206,9 +214,8 @@ private DataFlow::SourceNode streamRef(PipeCall pipeCall) {
  * Holds if the source stream of the given pipe call has an `error` handler registered.
  */
 predicate hasErrorHandlerRegistered(PipeCall pipeCall) {
-  exists(StreamEventRegistration handler |
-    handler = streamRef(pipeCall).getAMethodCall(getEventHandlerMethodName()) and
-    handler.getArgument(0).getStringValue() = "error"
+  exists(ErrorHandlerRegistration handler |
+    handler = streamRef(pipeCall).getAMethodCall(getEventHandlerMethodName())
   )
   or
   hasPlumber(pipeCall)
@@ -218,6 +225,8 @@ predicate hasErrorHandlerRegistered(PipeCall pipeCall) {
  * Holds if one of the arguments of the pipe call is a `gulp-plumber` monkey patch.
  */
 predicate hasPlumber(PipeCall pipeCall) {
+  pipeCall.getDestinationStream().getALocalSource() = API::moduleImport("gulp-plumber").getACall()
+  or
   streamRef+(pipeCall) = API::moduleImport("gulp-plumber").getACall()
 }
 
@@ -246,9 +255,19 @@ private predicate hasNonStreamSourceLikeUsage(PipeCall pipeCall) {
   )
 }
 
+/**
+ * Holds if the pipe call destination stream has an error handler registered.
+ */
+predicate hasErrorHandlerDownstream(PipeCall pipeCall) {
+  exists(ErrorHandlerRegistration handler |
+    handler.getReceiver().getALocalSource() = pipeResultRef(pipeCall)
+  )
+}
+
 from PipeCall pipeCall
 where
   not hasErrorHandlerRegistered(pipeCall) and
+  hasErrorHandlerDownstream(pipeCall) and
   not isPipeFollowedByNonStreamAccess(pipeCall) and
   not hasNonStreamSourceLikeUsage(pipeCall) and
   not hasNonNodeJsStreamSource(pipeCall)

javascript/ql/test/query-tests/Quality/UnhandledStreamPipe/test.expected

@@ -5,9 +5,16 @@
 | 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:109:26:109:37 | s.pipe(dest) | Stream pipe without error handling on the source stream. Errors won't propagate downstream and may be silently dropped. |
-| test.js:116:5:116: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:125:5:125: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:143:5:143:62 | stream. ... itable) | Stream pipe without error handling on the source stream. Errors won't propagate downstream and may be silently dropped. |
-| test.js:175:17:175:40 | notStre ... itable) | Stream pipe without error handling on the source stream. Errors won't propagate downstream and may be silently dropped. |
-| test.js:185:5:185:32 | copyStr ... nation) | 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:29:5:29:40 | wrapper ... Stream) | 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: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:73:5:73:40 | wrapper ... 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. |

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

@@ -1,7 +1,7 @@
 function test() {
   {
     const stream = getStream();
-    stream.pipe(destination); // $Alert
+    stream.pipe(destination).on("error", e); // $Alert
   }
   {
     const stream = getStream();
@@ -16,7 +16,7 @@ function test() {
   {
     const stream = getStream();
     const s2 = stream;
-    s2.pipe(dest); // $Alert
+    s2.pipe(dest).on("error", e); // $Alert
   }
   {
     const stream = getStream();
@@ -42,7 +42,7 @@ function test() {
     const stream = getStream();
     stream.on('error', handleError);
     const stream2 = stream.pipe(destination);
-    stream2.pipe(destination2); // $Alert
+    stream2.pipe(destination2).on("error", e); // $Alert
   }
   {
     const stream = getStream();
@@ -57,13 +57,13 @@ function test() {
     const stream = getStream();
     stream.on('error', handleError);
     const stream2 = stream.pipe(destination);
-    stream2.pipe(destination2); // $Alert
+    stream2.pipe(destination2).on("error", e); // $Alert
   }
   { // Error handler on destination instead of source
     const stream = getStream();
     const dest = getDest();
     dest.on('error', handler);
-    stream.pipe(dest); // $Alert
+    stream.pipe(dest).on("error", e); // $Alert
   }
   { // Multiple aliases, error handler on one
     const stream = getStream();
@@ -76,7 +76,7 @@ function test() {
     const stream = getStream();
     const s2 = stream.pipe(destination1);
     stream.on('error', handleError);
-    s2.pipe(destination2); // $Alert
+    s2.pipe(destination2).on("error", e); // $Alert
   }
   { // Handler registered via .once
     const stream = getStream();
@@ -91,7 +91,7 @@ function test() {
   { // Handler registered for unrelated event
     const stream = getStream();
     stream.on('close', handleClose);
-    stream.pipe(dest); // $Alert
+    stream.pipe(dest).on("error", e); // $Alert
   }
   { // Error handler registered after pipe, but before error
     const stream = getStream();
@@ -106,14 +106,17 @@ function test() {
   }
   { // Pipe in a function, error handler not set
     const stream = getStream();
-    function doPipe(s) { s.pipe(dest); } // $Alert
+    function doPipe(s) { 
+      f = s.pipe(dest); // $Alert
+      f.on("error", e); 
+    }
     doPipe(stream);
   }
   { // Dynamic event assignment
     const stream = getStream();
     const event = 'error';
     stream.on(event, handleError);
-    stream.pipe(dest);  // $SPURIOUS:Alert
+    stream.pipe(dest).on("error", e);  // $SPURIOUS:Alert
   }
   { // Handler assigned via variable property
     const stream = getStream();
@@ -122,12 +125,12 @@ function test() {
     stream.pipe(dest);
   }
   { // Pipe with no intermediate variable, no error handler
-    getStream().pipe(dest); // $Alert
+    getStream().pipe(dest).on("error", e); // $Alert
   }
   { // Handler set via .addListener synonym
     const stream = getStream();
     stream.addListener('error', handleError);
-    stream.pipe(dest);
+    stream.pipe(dest).on("error", e);
   }
   { // Handler set via .once after .pipe
     const stream = getStream();
@@ -140,7 +143,11 @@ function test() {
   }
   { // Long chained pipe without error handler
     const stream = getStream();
-    stream.pause().setEncoding('utf8').resume().pipe(writable); // $Alert
+    stream.pause().setEncoding('utf8').resume().pipe(writable).on("error", e); // $Alert
+  }
+  { // Long chained pipe without error handler
+    const stream = getStream();
+    stream.pause().setEncoding('utf8').on("error", e).resume().pipe(writable).on("error", e);
   }
   { // Non-stream with pipe method that returns subscribable object (Streams do not have subscribe method)
     const notStream = getNotAStream();
@@ -172,7 +179,7 @@ function test() {
   }
   { // Member access on a stream after pipe
     const notStream = getNotAStream();
-    const val = notStream.pipe(writable).readable; // $Alert
+    const val = notStream.pipe(writable).on("error", e).readable; // $Alert
   }
   { // Method access on a non-stream after pipe
     const notStream = getNotAStream();
@@ -182,7 +189,7 @@ function test() {
     const fs = require('fs');
     const stream = fs.createReadStream('file.txt');
     const copyStream = stream;
-    copyStream.pipe(destination); // $Alert
+    copyStream.pipe(destination).on("error", e); // $Alert
   }
   {
     const notStream = getNotAStream();
@@ -211,6 +218,16 @@ function test() {
     const p = plumber();
     getStream().pipe(p).pipe(dest).pipe(dest).pipe(dest);
   }
+  {
+    const plumber = require('gulp-plumber');
+    const p = plumber();
+    getStream().pipe(p);
+  }
+  {
+    const plumber = require('gulp-plumber');
+    const p = plumber();
+    getStream().pipe(p).pipe(dest);
+  }
   {
     const notStream = getNotAStream();
     notStream.pipe(getStream(),()=>{});

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

@@ -0,0 +1,113 @@
+const fs = require('fs');
+const zlib = require('zlib');
+
+function foo(){
+    const source = fs.createReadStream('input.txt');
+    const gzip = zlib.createGzip();
+    const destination = fs.createWriteStream('output.txt.gz');
+    source.pipe(gzip).pipe(destination); // $Alert    
+    gzip.on('error', e);
+}
+class StreamWrapper {
+    constructor() {
+        this.outputStream = getStream();
+    }
+}
+
+function zip() {
+    const zipStream = createWriteStream(zipPath);
+    let wrapper = new StreamWrapper();
+    let stream = wrapper.outputStream;
+    stream.on('error', e);
+    stream.pipe(zipStream);
+    zipStream.on('error', e);
+}
+
+function zip1() {
+    const zipStream = createWriteStream(zipPath);
+    let wrapper = new StreamWrapper();
+    wrapper.outputStream.pipe(zipStream); // $SPURIOUS:Alert
+    wrapper.outputStream.on('error', e);
+    zipStream.on('error', e);
+}
+
+function zip2() {
+    const zipStream = createWriteStream(zipPath);
+    let wrapper = new StreamWrapper();
+    let outStream = wrapper.outputStream.pipe(zipStream); // $Alert
+    outStream.on('error', e);
+}
+
+function zip3() {
+    const zipStream = createWriteStream(zipPath);
+    let wrapper = new StreamWrapper();
+    wrapper.outputStream.pipe(zipStream); // $Alert
+    zipStream.on('error', e);
+}
+
+function zip3() {
+    const zipStream = createWriteStream(zipPath);
+    let wrapper = new StreamWrapper();
+    let source = getStream();
+    source.pipe(wrapper.outputStream); // $MISSING:Alert
+    wrapper.outputStream.on('error', e);
+}
+
+function zip4() {
+    const zipStream = createWriteStream(zipPath);
+    let stream = getStream();
+    let output = stream.pipe(zipStream); // $Alert
+    output.on('error', e);
+}
+
+class StreamWrapper2 {
+    constructor() {
+        this.outputStream = getStream();
+        this.outputStream.on('error', e);
+    }
+
+}
+function zip5() {
+    const zipStream = createWriteStream(zipPath);
+    let wrapper = new StreamWrapper2();
+    wrapper.outputStream.pipe(zipStream); // $SPURIOUS:Alert
+    zipStream.on('error', e);
+}
+
+class StreamWrapper3 {
+    constructor() {
+        this.stream = getStream();
+    }
+    pipeIt(dest) {
+        return this.stream.pipe(dest);
+    }
+    register_error_handler(listener) {
+        return this.stream.on('error', listener);
+    }
+}
+
+function zip5() {
+    const zipStream = createWriteStream(zipPath);
+    let wrapper = new StreamWrapper3();
+    wrapper.pipeIt(zipStream); // $MISSING:Alert
+    zipStream.on('error', e);
+}
+function zip6() {
+    const zipStream = createWriteStream(zipPath);
+    let wrapper = new StreamWrapper3();
+    wrapper.pipeIt(zipStream);
+    wrapper.register_error_handler(e);
+    zipStream.on('error', e);
+}
+
+function registerErr(stream, listerner) {
+    stream.on('error', listerner);
+}
+
+function zip7() {
+    const zipStream = createWriteStream(zipPath);
+    let stream = getStream();
+    registerErr(stream, e);
+    stream.pipe(zipStream); // $SPURIOUS:Alert
+    zipStream.on('error', e);
+}