diff --git a/javascript/config/suites/javascript/security b/javascript/config/suites/javascript/security
index 3e87d292d7b7..959f5bacefe4 100644
--- a/javascript/config/suites/javascript/security
+++ b/javascript/config/suites/javascript/security
@@ -49,6 +49,7 @@
 + semmlecode-javascript-queries/Security/CWE-640/HostHeaderPoisoningInEmailGeneration.ql: /Security/CWE/CWE-640
 + semmlecode-javascript-queries/Security/CWE-643/XpathInjection.ql: /Security/CWE/CWE-643
 + semmlecode-javascript-queries/Security/CWE-730/RegExpInjection.ql: /Security/CWE/CWE-730
++ semmlecode-javascript-queries/Security/CWE-730/UnclosedStream.ql: /Security/CWE/CWE-730
 + semmlecode-javascript-queries/Security/CWE-754/UnvalidatedDynamicMethodCall.ql: /Security/CWE/CWE-754
 + semmlecode-javascript-queries/Security/CWE-770/MissingRateLimiting.ql: /Security/CWE/CWE-770
 + semmlecode-javascript-queries/Security/CWE-776/XmlBomb.ql: /Security/CWE/CWE-776
diff --git a/javascript/ql/src/Security/CWE-730/UnclosedStream.qhelp b/javascript/ql/src/Security/CWE-730/UnclosedStream.qhelp
new file mode 100644
index 000000000000..76b1537924d6
--- /dev/null
+++ b/javascript/ql/src/Security/CWE-730/UnclosedStream.qhelp
@@ -0,0 +1,69 @@
+<!DOCTYPE qhelp PUBLIC
+"-//Semmle//qhelp//EN"
+"qhelp.dtd">
+<qhelp>
+
+<overview>
+
+	<p>
+
+		Callback-based input streams generate calls until they are empty or
+		closed explicitly, and they will take up system resources until
+		that point in time.
+
+	</p>
+
+	<p>
+
+		Forgetting to close a stream that no longer is useful can cause
+		unnecessary exhaustion of system resources, which may cause the
+		application to be unresponsive or even crash.
+
+	</p>
+
+</overview>
+
+<recommendation>
+
+	<p>
+
+		Close input streams explicitly when there is no interest in their
+		remaining content.
+
+	</p>
+
+</recommendation>
+
+<example>
+
+	<p>
+
+In the following example, an array stores the content of a stream
+until the array becomes too big. When that happens, an error message
+is generated, and the stream processing is supposed to end.
+
+	</p>
+
+	<sample src="examples/UnclosedStream.js"/>
+
+	<p>
+
+The stream will however keep invoking the callback until it becomes
+empty, regardless of the programmers intention to not process
+additional content. This means that the array will grow indefinitely, if a malicious actor keep sending data.
+
+Close the stream to remedy this:
+
+	</p>
+
+	<sample src="examples/UnclosedStream_fixed.js"/>
+
+</example>
+
+<references>
+
+	<li>Node.js: <a href="https://nodejs.org/api/stream.html">Stream</a></li>
+
+</references>
+
+</qhelp>
diff --git a/javascript/ql/src/Security/CWE-730/UnclosedStream.ql b/javascript/ql/src/Security/CWE-730/UnclosedStream.ql
new file mode 100644
index 000000000000..a8672bdb3c60
--- /dev/null
+++ b/javascript/ql/src/Security/CWE-730/UnclosedStream.ql
@@ -0,0 +1,128 @@
+/**
+ * @name Unclosed stream
+ * @description A stream that is not closed may leak system resources.
+ * @kind problem
+ * @problem.severity error
+ * @precision high
+ * @id js/unclosed-stream
+ * @tags security
+ *       external/cwe/cwe-730
+ *       external/cwe/cwe-404
+ */
+
+import javascript
+
+/**
+ * Gets a function that `caller` invokes directly or indirectly as a callback to a library function.
+ */
+Function getACallee(Function caller) {
+  exists(DataFlow::InvokeNode invk | invk.getEnclosingFunction() = caller |
+    result = invk.getACallee()
+    or
+    not exists(invk.getACallee()) and
+    result = invk.getCallback(invk.getNumArgument()).getFunction()
+  )
+}
+
+/**
+ * A function that can be terminated meaningfully in an asynchronous context.
+ */
+abstract class AsyncTerminatableFunction extends DataFlow::FunctionNode {
+  /**
+   * Gets a node where this function terminates.
+   *
+   * It can be expected that no further expressions in this function will be evaluated after the evaluation of this node.
+   */
+  abstract DataFlow::Node getTermination();
+
+  /**
+   * Gets a brief description of this function.
+   */
+  abstract string getKind();
+}
+
+/**
+ * A promise executor as a function that can be terminated in an asynchronous context.
+ */
+class PromiseExecutor extends AsyncTerminatableFunction {
+  PromiseDefinition def;
+
+  PromiseExecutor() { this = def.getExecutor() }
+
+  override DataFlow::CallNode getTermination() {
+    result = [def.getRejectParameter(), def.getResolveParameter()].getACall()
+  }
+
+  override string getKind() { result = "promise" }
+}
+
+/**
+ * A callback-invoking function heuristic as a function that can be terminated in an asynchronous context.
+ */
+class FunctionWithCallback extends AsyncTerminatableFunction {
+  DataFlow::ParameterNode callbackParameter;
+
+  FunctionWithCallback() {
+    // the last parameter is the callback
+    callbackParameter = this.getLastParameter() and
+    // simple escape analysis
+    not exists(DataFlow::Node escape | callbackParameter.flowsTo(escape) |
+      escape = any(DataFlow::PropWrite w).getRhs() or
+      escape = any(DataFlow::CallNode c).getAnArgument()
+    ) and
+    // no return value
+    (this.getFunction() instanceof ArrowFunctionExpr or not exists(this.getAReturn())) and
+    // all callback invocations are terminal (note that this permits calls in closures)
+    forex(DataFlow::CallNode termination | termination = callbackParameter.getACall() |
+      termination.asExpr() = any(Function f).getExit().getAPredecessor()
+    ) and
+    // avoid confusion with promises
+    not this instanceof PromiseExecutor and
+    not exists(PromiseCandidate c | this.flowsTo(c.getAnArgument()))
+  }
+
+  override DataFlow::CallNode getTermination() { result = callbackParameter.getACall() }
+
+  override string getKind() { result = "asynchronous function" }
+}
+
+/**
+ * A data stream.
+ */
+class Stream extends DataFlow::SourceNode {
+  DataFlow::FunctionNode processor;
+
+  Stream() {
+    exists(DataFlow::CallNode onData |
+      this instanceof EventEmitter and
+      onData = this.getAMethodCall(EventEmitter::on()) and
+      onData.getArgument(0).mayHaveStringValue("data") and
+      processor = onData.getCallback(1)
+    )
+  }
+
+  /**
+   * Gets a call that closes thes stream.
+   */
+  DataFlow::Node getClose() { result = this.getAMethodCall("destroy") }
+
+  /**
+   * Gets a function that processes the content of the stream.
+   */
+  DataFlow::FunctionNode getProcessor() { result = processor }
+}
+
+from
+  AsyncTerminatableFunction terminatable, DataFlow::CallNode termination, Stream stream,
+  Function processor
+where
+  stream.getAstNode().getContainer() = getACallee*(terminatable.getFunction()) and
+  termination = terminatable.getTermination() and
+  processor = stream.getProcessor().getFunction() and
+  termination.asExpr().getEnclosingFunction() = getACallee*(processor) and
+  not exists(Function destroyFunction |
+    destroyFunction = getACallee*(processor) and
+    stream.getClose().asExpr().getEnclosingFunction() = destroyFunction
+  )
+select processor, "This stream processor $@ the enclosing $@ without closing the stream.",
+  termination, "terminates", terminatable, terminatable.getKind()
diff --git a/javascript/ql/src/Security/CWE-730/examples/UnclosedStream.js b/javascript/ql/src/Security/CWE-730/examples/UnclosedStream.js
new file mode 100644
index 000000000000..16028acb24ab
--- /dev/null
+++ b/javascript/ql/src/Security/CWE-730/examples/UnclosedStream.js
@@ -0,0 +1,16 @@
+var https = require("https");
+
+new Promise(function dispatchHttpRequest(resolve, reject) {
+  https.request(options, function(stream) {
+    var responseBuffer = [];
+    stream.on("data", function(chunk) {
+      responseBuffer.push(chunk);
+
+      if (responseBuffer.length > 1024) {
+        reject("Input size of 1024 exceeded."); // BAD
+      }
+    });
+
+    stream.on("end", () => resolve(responseBuffer));
+  });
+});
diff --git a/javascript/ql/src/Security/CWE-730/examples/UnclosedStream_fixed.js b/javascript/ql/src/Security/CWE-730/examples/UnclosedStream_fixed.js
new file mode 100644
index 000000000000..31b6d04ac352
--- /dev/null
+++ b/javascript/ql/src/Security/CWE-730/examples/UnclosedStream_fixed.js
@@ -0,0 +1,17 @@
+var https = require("https");
+
+new Promise(function dispatchHttpRequest(resolve, reject) {
+  https.request(options, function(stream) {
+    var responseBuffer = [];
+    stream.on("data", function(chunk) {
+      responseBuffer.push(chunk);
+
+      if (responseBuffer.length > 1024) {
+        stream.destroy();
+        reject("Input size of 1024 exceeded."); // GOOD
+      }
+    });
+
+    stream.on("end", () => resolve(responseBuffer));
+  });
+});
diff --git a/javascript/ql/src/semmle/javascript/frameworks/NodeJSLib.qll b/javascript/ql/src/semmle/javascript/frameworks/NodeJSLib.qll
index add7809eeb2c..070ef0fb1816 100644
--- a/javascript/ql/src/semmle/javascript/frameworks/NodeJSLib.qll
+++ b/javascript/ql/src/semmle/javascript/frameworks/NodeJSLib.qll
@@ -826,7 +826,7 @@ module NodeJSLib {
   /**
    * A model of a URL request in the Node.js `http` library.
    */
-  private class NodeHttpUrlRequest extends NodeJSClientRequest::Range {
+  private class NodeHttpUrlRequest extends NodeJSClientRequest::Range, NodeJSEventEmitter {
     DataFlow::Node url;
 
     NodeHttpUrlRequest() {
@@ -881,8 +881,11 @@ module NodeJSLib {
       exists(DataFlow::MethodCallNode mcn |
         clientRequest.getAMethodCall(EventEmitter::on()) = mcn and
         mcn.getArgument(0).mayHaveStringValue(handledEvent) and
-        flowsTo(mcn.getArgument(1))
+        this.flowsTo(mcn.getArgument(1))
       )
+      or
+      this.flowsTo(clientRequest.(DataFlow::CallNode).getLastArgument()) and
+      handledEvent = "connection"
     }
 
     /**
@@ -1060,6 +1063,15 @@ module NodeJSLib {
     }
   }
 
+  private class ClientRequestEventEmitter extends NodeJSEventEmitter {
+    ClientRequestEventEmitter() {
+      exists(ClientRequestHandler handler |
+        not handler.getAHandledEvent() = "error" and
+        this = handler.getAParameter()
+      )
+    }
+  }
+
   /**
    * A registration of an event handler on a NodeJS EventEmitter instance.
    */
diff --git a/javascript/ql/test/library-tests/frameworks/EventEmitter/test.expected b/javascript/ql/test/library-tests/frameworks/EventEmitter/test.expected
index 7c8466e15ecd..4918d26fa9f7 100644
--- a/javascript/ql/test/library-tests/frameworks/EventEmitter/test.expected
+++ b/javascript/ql/test/library-tests/frameworks/EventEmitter/test.expected
@@ -1,3 +1,4 @@
+taintSteps
 | customEmitter.js:5:20:5:24 | "bar" | customEmitter.js:6:19:6:22 | data |
 | customEmitter.js:12:21:12:25 | "baz" | customEmitter.js:13:23:13:26 | data |
 | customEmitter.js:12:21:12:25 | "baz" | customEmitter.js:22:14:22:18 | yData |
@@ -15,3 +16,22 @@
 | tst.js:40:20:40:27 | "yabity" | tst.js:39:19:39:19 | x |
 | tst.js:46:28:46:38 | 'FirstData' | tst.js:43:45:43:49 | first |
 | tst.js:47:29:47:40 | 'SecondData' | tst.js:44:37:44:42 | second |
+eventEmitter
+| customEmitter.js:3:1:8:1 | class M ... );\\n\\t}\\n} |
+| customEmitter.js:17:9:17:29 | new MyS ... itter() |
+| customEmitter.js:20:9:20:29 | new MyS ... itter() |
+| tst2.js:6:12:6:42 | new Con ... , opts) |
+| tst2.js:16:10:16:24 | new Connector() |
+| tst2.js:22:12:24:2 | http.re ... {});\\n}) |
+| tst2.js:22:37:22:39 | res |
+| tst2.js:25:28:25:33 | socket |
+| tst2.js:29:12:31:2 | http.re ... {});\\n}) |
+| tst2.js:29:37:29:39 | res |
+| tst2.js:32:28:32:33 | socket |
+| tst.js:3:10:3:22 | new emitter() |
+| tst.js:13:11:13:23 | new emitter() |
+| tst.js:18:11:18:23 | new emitter() |
+| tst.js:24:11:24:23 | new emitter() |
+| tst.js:32:11:32:30 | new MyEventEmitter() |
+| tst.js:38:11:38:38 | new Ext ... itter() |
+| tst.js:42:15:42:32 | require('process') |
diff --git a/javascript/ql/test/library-tests/frameworks/EventEmitter/test.ql b/javascript/ql/test/library-tests/frameworks/EventEmitter/test.ql
index e8de4ff4c53d..9b0f16161c48 100644
--- a/javascript/ql/test/library-tests/frameworks/EventEmitter/test.ql
+++ b/javascript/ql/test/library-tests/frameworks/EventEmitter/test.ql
@@ -3,3 +3,5 @@ import javascript
 query predicate taintSteps(DataFlow::Node pred, DataFlow::Node succ) {
   exists(DataFlow::AdditionalFlowStep step | step.step(pred, succ))
 }
+
+query predicate eventEmitter(EventEmitter e) { any() }
diff --git a/javascript/ql/test/library-tests/frameworks/EventEmitter/tst2.js b/javascript/ql/test/library-tests/frameworks/EventEmitter/tst2.js
index 23cc0ffb5ba5..617c5be480e2 100644
--- a/javascript/ql/test/library-tests/frameworks/EventEmitter/tst2.js
+++ b/javascript/ql/test/library-tests/frameworks/EventEmitter/tst2.js
@@ -1,5 +1,5 @@
-var util = require('util');
-var EventEmitter = require('events').EventEmitter;
+var util = require("util");
+var EventEmitter = require("events").EventEmitter;
 
 var Connector = function() {
   if (!(this instanceof Connector)) {
@@ -16,3 +16,19 @@ Connector.prototype.foo = function() {};
 var em = new Connector();
 em.on("foo", bar => {});
 em.emit("foo", "bar");
+
+var http = require("http");
+
+let req1 = http.request(x, function(res) {
+  res.on("data", function(data) {});
+});
+req1.on("socket", function(socket) {
+  socket.on("data", function(data) {});
+});
+
+let req2 = http.request(x, function(res) {
+  res.on("error", function(error) {});
+});
+req2.on("socket", function(socket) {
+  socket.on("error", function(error) {});
+});
diff --git a/javascript/ql/test/library-tests/frameworks/NodeJSLib/http.js b/javascript/ql/test/library-tests/frameworks/NodeJSLib/http.js
new file mode 100644
index 000000000000..af26259997d4
--- /dev/null
+++ b/javascript/ql/test/library-tests/frameworks/NodeJSLib/http.js
@@ -0,0 +1,2 @@
+var http = require("http");
+http.request(x, data => data.on("data", d => undefined));
diff --git a/javascript/ql/test/library-tests/frameworks/NodeJSLib/tests.expected b/javascript/ql/test/library-tests/frameworks/NodeJSLib/tests.expected
index 2e048dc43b89..c80b0059cb6a 100644
--- a/javascript/ql/test/library-tests/frameworks/NodeJSLib/tests.expected
+++ b/javascript/ql/test/library-tests/frameworks/NodeJSLib/tests.expected
@@ -96,6 +96,7 @@ test_RouteSetup_getServer
 | src/indirect2.js:18:14:18:35 | http.cr ... er(get) | src/indirect2.js:18:14:18:35 | http.cr ... er(get) |
 | src/indirect.js:34:14:34:58 | http.cr ... dler()) | src/indirect.js:34:14:34:58 | http.cr ... dler()) |
 test_ClientRequest
+| http.js:2:1:2:56 | http.re ... fined)) |
 | src/http.js:18:1:18:30 | http.re ... uth" }) |
 | src/http.js:21:15:26:6 | http.re ... \\n    }) |
 | src/http.js:27:16:27:73 | http.re ... POST'}) |
diff --git a/javascript/ql/test/query-tests/Security/CWE-730/UnclosedStream.expected b/javascript/ql/test/query-tests/Security/CWE-730/UnclosedStream.expected
new file mode 100644
index 000000000000..be598667e8e1
--- /dev/null
+++ b/javascript/ql/test/query-tests/Security/CWE-730/UnclosedStream.expected
@@ -0,0 +1,5 @@
+| unclosed-stream.js:9:23:9:50 | d => re ... silly") | This stream processor $@ the enclosing $@ without closing the stream. | unclosed-stream.js:9:28:9:50 | resolve ... silly") | terminates | unclosed-stream.js:3:13:46:1 | functio ...   });\\n} | promise |
+| unclosed-stream.js:11:23:11:49 | d => re ... silly") | This stream processor $@ the enclosing $@ without closing the stream. | unclosed-stream.js:11:28:11:49 | rejectP ... silly") | terminates | unclosed-stream.js:3:13:46:1 | functio ...   });\\n} | promise |
+| unclosed-stream.js:13:23:13:50 | d => re ... silly") | This stream processor $@ the enclosing $@ without closing the stream. | unclosed-stream.js:5:5:5:24 | rejectPromise(value) | terminates | unclosed-stream.js:3:13:46:1 | functio ...   });\\n} | promise |
+| unclosed-stream.js:35:23:35:50 | d => re ... silly") | This stream processor $@ the enclosing $@ without closing the stream. | unclosed-stream.js:35:28:35:50 | resolve ... silly") | terminates | unclosed-stream.js:3:13:46:1 | functio ...   });\\n} | promise |
+| unclosed-stream.js:43:23:43:50 | d => re ... silly") | This stream processor $@ the enclosing $@ without closing the stream. | unclosed-stream.js:43:28:43:50 | resolve ... silly") | terminates | unclosed-stream.js:3:13:46:1 | functio ...   });\\n} | promise |
diff --git a/javascript/ql/test/query-tests/Security/CWE-730/UnclosedStream.qlref b/javascript/ql/test/query-tests/Security/CWE-730/UnclosedStream.qlref
new file mode 100644
index 000000000000..f9580df7182a
--- /dev/null
+++ b/javascript/ql/test/query-tests/Security/CWE-730/UnclosedStream.qlref
@@ -0,0 +1 @@
+Security/CWE-730/UnclosedStream.ql
diff --git a/javascript/ql/test/query-tests/Security/CWE-730/unclosed-stream.js b/javascript/ql/test/query-tests/Security/CWE-730/unclosed-stream.js
new file mode 100644
index 000000000000..5ae77da5a617
--- /dev/null
+++ b/javascript/ql/test/query-tests/Security/CWE-730/unclosed-stream.js
@@ -0,0 +1,46 @@
+var http = require("http");
+
+new Promise(function(resolvePromise, rejectPromise) {
+  function rejectIndirect(value) {
+    rejectPromise(value);
+  }
+
+  http.request({}, function(stream) {
+    stream.on("data", d => resolvePromise("silly")); // NOT OK
+
+    stream.on("data", d => rejectPromise("silly")); // NOT OK
+
+    stream.on("data", d => rejectIndirect("silly")); // NOT OK
+
+    stream.on(
+      "error",
+      e => rejectIndirect(e) // OK
+    );
+
+    stream.on(
+      "end",
+      e => resolvePromise(e) // OK
+    );
+  });
+
+  http.request({}, function(stream) {
+    stream.on("data", d => {
+      // OK
+      stream.destroy();
+      resolvePromise("silly");
+    });
+  });
+
+  http.request({}, function(stream) {
+    stream.on("data", d => resolvePromise("silly")); // OK [INCONSISTENCY]: the other data handler closes the stream
+    stream.on("data", d => {
+      // OK
+      stream.destroy();
+      resolvePromise("silly");
+    });
+  });
+  http.request({}, function(stream) {
+    stream.on("data", d => resolvePromise("silly")); // OK [INCONSISTENCY]: the stream is closed automatically
+    setTimeout(100, () => stream.destroy());
+  });
+});