Python: model HTTP sink in Tornado

python/ql/src/semmle/python/frameworks/Tornado.qll

@@ -216,6 +216,17 @@ private module Tornado {
         /** Gets a reference to one of the methods `get_arguments`, `get_body_arguments`, `get_query_arguments`. */
         DataFlow::Node argumentsMethod() { result = argumentsMethod(DataFlow::TypeTracker::end()) }
 
+        /** Gets a reference to the `write` method. */
+        private DataFlow::Node writeMethod(DataFlow::TypeTracker t) {
+          t.startInAttr("write") and
+          result = instance()
+          or
+          exists(DataFlow::TypeTracker t2 | result = writeMethod(t2).track(t2, t))
+        }
+
+        /** Gets a reference to the `write` method. */
+        DataFlow::Node writeMethod() { result = writeMethod(DataFlow::TypeTracker::end()) }
+
         private class AdditionalTaintStep extends TaintTracking::AdditionalTaintStep {
           override predicate step(DataFlow::Node nodeFrom, DataFlow::Node nodeTo) {
             // Method access
@@ -540,4 +551,29 @@ private module Tornado {
       not result = this.getArg(0)
     }
   }
+
+  // ---------------------------------------------------------------------------
+  // Response modeling
+  // ---------------------------------------------------------------------------
+  /**
+   * A call to `tornado.web.RequestHandler.write` method.
+   *
+   * See https://www.tornadoweb.org/en/stable/web.html?highlight=write#tornado.web.RequestHandler.write
+   */
+  private class TornadoRequestHandlerWriteCall extends HTTP::Server::HttpResponse::Range,
+    DataFlow::CfgNode {
+    override CallNode node;
+
+    TornadoRequestHandlerWriteCall() {
+      node.getFunction() = tornado::web::RequestHandler::writeMethod().asCfgNode()
+    }
+
+    override DataFlow::Node getBody() {
+      result.asCfgNode() in [node.getArg(0), node.getArgByName("chunk")]
+    }
+
+    override string getMimetypeDefault() { result = "text/html" }
+
+    override DataFlow::Node getMimetypeOrContentTypeArg() { none() }
+  }
 }

python/ql/test/experimental/library-tests/frameworks/tornado/ConceptsTest.ql

@@ -1,12 +1,12 @@
 import python
 import experimental.meta.ConceptsTest
-//
-// class DedicatedResponseTest extends HttpServerHttpResponseTest {
-//   DedicatedResponseTest() { file.getShortName() = "response_test.py" }
-// }
-//
-// class OtherResponseTest extends HttpServerHttpResponseTest {
-//   OtherResponseTest() { not this instanceof DedicatedResponseTest }
-//
-//   override string getARelevantTag() { result = "HttpResponse" }
-// }
+
+class DedicatedResponseTest extends HttpServerHttpResponseTest {
+  DedicatedResponseTest() { file.getShortName() = "response_test.py" }
+}
+
+class OtherResponseTest extends HttpServerHttpResponseTest {
+  OtherResponseTest() { not this instanceof DedicatedResponseTest }
+
+  override string getARelevantTag() { result = "HttpResponse" }
+}

python/ql/test/experimental/library-tests/frameworks/tornado/basic.py

@@ -3,21 +3,21 @@
 
 class BasicHandler(tornado.web.RequestHandler):
     def get(self):  # $ requestHandler
-        self.write("BasicHandler " + self.get_argument("xss"))
+        self.write("BasicHandler " + self.get_argument("xss")) # $ HttpResponse
 
     def post(self):  # $ requestHandler
-        self.write("BasicHandler (POST)")
+        self.write("BasicHandler (POST)") # $ HttpResponse
 
 
 class DeepInheritance(BasicHandler):
     def get(self):  # $ requestHandler
-        self.write("DeepInheritance" + self.get_argument("also_xss"))
+        self.write("DeepInheritance" + self.get_argument("also_xss")) # $ HttpResponse
 
 
 class FormHandler(tornado.web.RequestHandler):
     def post(self):  # $ requestHandler
         name = self.get_body_argument("name")
-        self.write(name)
+        self.write(name) # $ HttpResponse
 
 
 class RedirectHandler(tornado.web.RequestHandler):
@@ -30,7 +30,7 @@ def get(self):  # $ requestHandler
 
 class BaseReverseInheritance(tornado.web.RequestHandler):
     def get(self):  # $ requestHandler
-        self.write("hello from BaseReverseInheritance")
+        self.write("hello from BaseReverseInheritance") # $ HttpResponse
 
 
 class ReverseInheritance(BaseReverseInheritance):

python/ql/test/experimental/library-tests/frameworks/tornado/response_test.py

@@ -0,0 +1,85 @@
+import tornado.web
+import tornado.httputil
+
+
+class ResponseWriting(tornado.web.RequestHandler):
+    def get(self, type_):  # $ requestHandler routedParameter=type_
+        if type_ == "str":
+            self.write("foo") # $ HttpResponse mimetype=text/html responseBody="foo"
+        elif type_ == "bytes":
+            self.write(b"foo") # $ HttpResponse mimetype=text/html responseBody=b"foo"
+        elif type_ == "dict":
+            d = {"foo": 42}
+            # Content-type will be set to `application/json`
+            self.write(d) # $ HttpResponse responseBody=d MISSING: mimetype=application/json SPURIOUS: mimetype=text/html
+        else:
+            raise Exception("Bad type {} {}".format(type_, type(type_)))
+
+
+class ExplicitContentType(tornado.web.RequestHandler):
+    def get(self):  # $ requestHandler
+        # Note: Our current modeling makes it quite hard to give a good annotation here
+        # this write is technically while the HTTP response has mimetype text/html, but
+        # the returned HTTP response will have mimetype text/plain, which is _really_
+        # what matters.
+
+        self.write("foo") # $ HttpResponse mimetype=text/html responseBody="foo"
+        self.set_header("Content-Type", "text/plain; charset=utf-8")
+
+    def post(self): # $ requestHandler
+        self.set_header("Content-Type", "text/plain; charset=utf-8")
+        self.write("foo") # $ HttpResponse responseBody="foo" MISSING: mimetype=text/plain SPURIOUS: mimetype=text/html
+
+
+class ExampleRedirect(tornado.web.RequestHandler):
+    def get(self): # $ requestHandler
+        self.redirect("http://example.com") # TODO: Model redirect
+
+
+class ExampleConnectionWrite(tornado.web.RequestHandler):
+    def get(self, stream=False): # $ requestHandler routedParameter=stream
+
+        if not stream:
+            self.request.connection.write_headers(
+                tornado.httputil.ResponseStartLine('', 200, 'OK'),
+                tornado.httputil.HTTPHeaders(),
+            )
+            self.request.connection.write(b"foo") # $ MISSING: HttpResponse responseBody=b"foo"
+            self.request.connection.finish()
+        else:
+            # Note: The documentation says that connection.detach(): "May only be called
+            # during HTTPMessageDelegate.headers_received". Doing it here actually
+            # works, but does make tornado spit out some errors... good enough to
+            # illustrate the behavior.
+            #
+            # https://www.tornadoweb.org/en/stable/http1connection.html#tornado.http1connection.HTTP1Connection.detach
+            stream = self.request.connection.detach()
+            stream.write(b"foo stream") # $ MISSING: HttpResponse responseBody=b"foo stream"
+            stream.close()
+
+def make_app():
+    return tornado.web.Application(
+        [
+            (r"/ResponseWriting/(str|bytes|dict)", ResponseWriting),  # $ routeSetup="/ResponseWriting/(str|bytes|dict)"
+            (r"/ExplicitContentType", ExplicitContentType), # $ routeSetup="/ExplicitContentType"
+            (r"/ExampleRedirect", ExampleRedirect), # $ routeSetup="/ExampleRedirect"
+            (r"/ExampleConnectionWrite", ExampleConnectionWrite), # $ routeSetup="/ExampleConnectionWrite"
+            (r"/ExampleConnectionWrite/(stream)", ExampleConnectionWrite), # $ routeSetup="/ExampleConnectionWrite/(stream)"
+        ],
+        debug=True,
+    )
+
+
+if __name__ == "__main__":
+    import tornado.ioloop
+
+    app = make_app()
+    app.listen(8888)
+    tornado.ioloop.IOLoop.current().start()
+
+    # http://localhost:8888/ResponseWriting/str
+    # http://localhost:8888/ResponseWriting/bytes
+    # http://localhost:8888/ResponseWriting/dict
+    # http://localhost:8888/ExplicitContentType
+    # http://localhost:8888/ExampleRedirect
+    # http://localhost:8888/ExampleConnectionWrite

python/ql/test/experimental/library-tests/frameworks/tornado/routing_test.py

@@ -4,55 +4,55 @@
 
 class FooHandler(tornado.web.RequestHandler):
     def get(self, x, y=None, not_used=None):  # $ requestHandler routedParameter=x routedParameter=y
-        self.write("FooHandler {} {}".format(x, y))
+        self.write("FooHandler {} {}".format(x, y)) # $ HttpResponse
 
 
 class BarHandler(tornado.web.RequestHandler):
     def get(self, x, y=None, not_used=None):  # $ requestHandler routedParameter=x routedParameter=y SPURIOUS: routedParameter=not_used
-        self.write("BarHandler {} {}".format(x, y))
+        self.write("BarHandler {} {}".format(x, y)) # $ HttpResponse
 
 
 class BazHandler(tornado.web.RequestHandler):
     def get(self, x, y=None, not_used=None):  # $ requestHandler routedParameter=x routedParameter=y SPURIOUS: routedParameter=not_used
-        self.write("BazHandler {} {}".format(x, y))
+        self.write("BazHandler {} {}".format(x, y)) # $ HttpResponse
 
 
 class KwArgs(tornado.web.RequestHandler):
     def get(self, *, x, y=None, not_used=None):  # $ requestHandler routedParameter=x routedParameter=y
-        self.write("KwArgs {} {}".format(x, y))
+        self.write("KwArgs {} {}".format(x, y)) # $ HttpResponse
 
 
 class OnlyLocalhost(tornado.web.RequestHandler):
     def get(self):  # $ requestHandler
-        self.write("OnlyLocalhost")
+        self.write("OnlyLocalhost") # $ HttpResponse
 
 
 class One(tornado.web.RequestHandler):
     def get(self):  # $ requestHandler
-        self.write("One")
+        self.write("One") # $ HttpResponse
 
 
 class Two(tornado.web.RequestHandler):
     def get(self):  # $ requestHandler
-        self.write("Two")
+        self.write("Two") # $ HttpResponse
 
 
 class Three(tornado.web.RequestHandler):
     def get(self):  # $ requestHandler
-        self.write("Three")
+        self.write("Three") # $ HttpResponse
 
 
 class AddedLater(tornado.web.RequestHandler):
     def get(self, x, y=None, not_used=None):  # $ requestHandler routedParameter=x routedParameter=y
-        self.write("AddedLater {} {}".format(x, y))
+        self.write("AddedLater {} {}".format(x, y)) # $ HttpResponse
 
 
 class PossiblyNotRouted(tornado.web.RequestHandler):
     # Even if our analysis can't find a route-setup for this class, we should still
     # consider it to be a handle incoming HTTP requests
 
     def get(self):  # $ requestHandler
-        self.write("NotRouted")
+        self.write("NotRouted") # $ HttpResponse
 
 
 def make_app():