Python: Add Tornado source modeling

python/change-notes/2020-12-22-tornado-source-modeling.md

@@ -0,0 +1,2 @@
+lgtm,codescanning
+* Added modeling of sources of remote user input (`RemoteFlowSource`) when using `tornado` to create HTTP servers, to the new data-flow queries.

python/ql/src/semmle/python/Frameworks.qll

@@ -12,4 +12,5 @@ private import semmle.python.frameworks.MySQLdb
 private import semmle.python.frameworks.Psycopg2
 private import semmle.python.frameworks.PyMySQL
 private import semmle.python.frameworks.Stdlib
+private import semmle.python.frameworks.Tornado
 private import semmle.python.frameworks.Yaml

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

@@ -0,0 +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" }
+// }

python/ql/test/experimental/library-tests/frameworks/tornado/README.md

@@ -0,0 +1,3 @@
+From Tornado 6.0 Python 3.5+ is [required](https://www.tornadoweb.org/en/stable/index.html#installation)
+
+https://www.tornadoweb.org/en/stable/guide/structure.html#handling-request-input

python/ql/test/experimental/library-tests/frameworks/tornado/TestTaint.expected

@@ -0,0 +1,41 @@
+| taint_test.py:6 | ok   | get | name |
+| taint_test.py:6 | ok   | get | number |
+| taint_test.py:7 | ok   | get | foo |
+| taint_test.py:11 | ok   | get | self.get_argument(..) |
+| taint_test.py:12 | ok   | get | self.get_arguments(..) |
+| taint_test.py:13 | ok   | get | self.get_arguments(..)[0] |
+| taint_test.py:15 | ok   | get | self.get_body_argument(..) |
+| taint_test.py:16 | ok   | get | self.get_body_arguments(..) |
+| taint_test.py:17 | ok   | get | self.get_body_arguments(..)[0] |
+| taint_test.py:19 | ok   | get | self.get_query_argument(..) |
+| taint_test.py:20 | ok   | get | self.get_query_arguments(..) |
+| taint_test.py:21 | ok   | get | self.get_query_arguments(..)[0] |
+| taint_test.py:23 | ok   | get | self.path_args |
+| taint_test.py:24 | ok   | get | self.path_args[0] |
+| taint_test.py:26 | ok   | get | self.path_kwargs |
+| taint_test.py:27 | ok   | get | self.path_kwargs["name"] |
+| taint_test.py:34 | ok   | get | request |
+| taint_test.py:36 | ok   | get | request.uri |
+| taint_test.py:37 | ok   | get | request.path |
+| taint_test.py:38 | ok   | get | request.query |
+| taint_test.py:39 | ok   | get | request.full_url() |
+| taint_test.py:41 | ok   | get | request.remote_ip |
+| taint_test.py:43 | ok   | get | request.body |
+| taint_test.py:45 | ok   | get | request.arguments |
+| taint_test.py:46 | ok   | get | request.arguments["name"] |
+| taint_test.py:47 | ok   | get | request.arguments["name"][0] |
+| taint_test.py:49 | ok   | get | request.query_arguments |
+| taint_test.py:50 | ok   | get | request.query_arguments["name"] |
+| taint_test.py:51 | ok   | get | request.query_arguments["name"][0] |
+| taint_test.py:53 | ok   | get | request.body_arguments |
+| taint_test.py:54 | ok   | get | request.body_arguments["name"] |
+| taint_test.py:55 | ok   | get | request.body_arguments["name"][0] |
+| taint_test.py:58 | ok   | get | request.headers |
+| taint_test.py:59 | ok   | get | request.headers["header-name"] |
+| taint_test.py:60 | fail | get | request.headers.get_list(..) |
+| taint_test.py:61 | fail | get | request.headers.get_all() |
+| taint_test.py:62 | fail | get | ListComp |
+| taint_test.py:65 | ok   | get | request.cookies |
+| taint_test.py:66 | ok   | get | request.cookies["cookie-name"] |
+| taint_test.py:67 | fail | get | request.cookies["cookie-name"].key |
+| taint_test.py:68 | fail | get | request.cookies["cookie-name"].value |

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

@@ -0,0 +1,6 @@
+import experimental.dataflow.tainttracking.TestTaintLib
+import semmle.python.dataflow.new.RemoteFlowSources
+
+class RemoteFlowTestTaintConfiguration extends TestTaintTrackingConfiguration {
+  override predicate isSource(DataFlow::Node source) { source instanceof RemoteFlowSource }
+}

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

@@ -0,0 +1,69 @@
+import tornado.web
+
+
+class BasicHandler(tornado.web.RequestHandler):
+    def get(self):  # $ requestHandler
+        self.write("BasicHandler " + self.get_argument("xss"))
+
+    def post(self):  # $ requestHandler
+        self.write("BasicHandler (POST)")
+
+
+class DeepInheritance(BasicHandler):
+    def get(self):  # $ requestHandler
+        self.write("DeepInheritance" + self.get_argument("also_xss"))
+
+
+class FormHandler(tornado.web.RequestHandler):
+    def post(self):  # $ requestHandler
+        name = self.get_body_argument("name")
+        self.write(name)
+
+
+class RedirectHandler(tornado.web.RequestHandler):
+    def get(self):  # $ requestHandler
+        req = self.request
+        h = req.headers
+        url = h["url"]
+        self.redirect(url)
+
+
+class BaseReverseInheritance(tornado.web.RequestHandler):
+    def get(self):  # $ requestHandler
+        self.write("hello from BaseReverseInheritance")
+
+
+class ReverseInheritance(BaseReverseInheritance):
+    pass
+
+
+def make_app():
+    return tornado.web.Application(
+        [
+            (r"/basic", BasicHandler),  # $ routeSetup="/basic"
+            (r"/deep", DeepInheritance),  # $ routeSetup="/deep"
+            (r"/form", FormHandler),  # $ routeSetup="/form"
+            (r"/redirect", RedirectHandler),  # $ routeSetup="/redirect"
+            (r"/reverse-inheritance", ReverseInheritance),  # $ routeSetup="/reverse-inheritance"
+        ],
+        debug=True,
+    )
+
+
+if __name__ == "__main__":
+    import tornado.ioloop
+
+    app = make_app()
+    app.listen(8888)
+    tornado.ioloop.IOLoop.current().start()
+
+    # http://localhost:8888/basic?xss=foo
+    # http://localhost:8888/deep?also_xss=foo
+
+    # curl -X POST http://localhost:8888/basic
+    # curl -X POST http://localhost:8888/deep
+
+    # curl -X POST -F "name=foo" http://localhost:8888/form
+    # curl -v -H 'url: http://example.com' http://localhost:8888/redirect
+
+    # http://localhost:8888/reverse-inheritance

python/ql/test/experimental/library-tests/frameworks/tornado/options

@@ -0,0 +1 @@
+semmle-extractor-options: --max-import-depth=1 --lang=3

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

@@ -0,0 +1,35 @@
+import tornado.web
+
+
+class ResponseWriting(tornado.web.RequestHandler):
+    def get(self, type_):  # $ requestHandler routedParameter=type_
+        if type_ == "str":
+            self.write("foo")
+        elif type_ == "bytes":
+            self.write(b"foo")
+        elif type_ == "dict":
+            # Content-type will be set to `application/json`
+            self.write({"foo": 42})
+        else:
+            raise Exception("Bad type {} {}".format(type_, type(type_)))
+
+
+def make_app():
+    return tornado.web.Application(
+        [
+            (r"/ResponseWriting/(str|bytes|dict)", ResponseWriting),  # $ routeSetup="/ResponseWriting/(str|bytes|dict)"
+        ],
+        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

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

@@ -0,0 +1,110 @@
+import tornado.web
+import tornado.routing
+
+
+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))
+
+
+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))
+
+
+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))
+
+
+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))
+
+
+class OnlyLocalhost(tornado.web.RequestHandler):
+    def get(self):  # $ requestHandler
+        self.write("OnlyLocalhost")
+
+
+class One(tornado.web.RequestHandler):
+    def get(self):  # $ requestHandler
+        self.write("One")
+
+
+class Two(tornado.web.RequestHandler):
+    def get(self):  # $ requestHandler
+        self.write("Two")
+
+
+class Three(tornado.web.RequestHandler):
+    def get(self):  # $ requestHandler
+        self.write("Three")
+
+
+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))
+
+
+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")
+
+
+def make_app():
+    # see https://www.tornadoweb.org/en/stable/routing.html for even more examples
+    app = tornado.web.Application(
+        [
+            (r"/foo/([0-9]+)/([0-9]+)?", FooHandler),  # $ routeSetup="/foo/([0-9]+)/([0-9]+)?"
+            tornado.web.URLSpec(r"/bar/([0-9]+)/([0-9]+)?", BarHandler),  # $ MISSING: routeSetup="/bar/([0-9]+)/([0-9]+)?"
+            # Very verbose way to write same as FooHandler
+            tornado.routing.Rule(tornado.routing.PathMatches(r"/baz/([0-9]+)/([0-9]+)?"), BazHandler),  # $ MISSING: routeSetup="/baz/([0-9]+)/([0-9]+)?"
+            (r"/kw-args/(?P<x>[0-9]+)/(?P<y>[0-9]+)?", KwArgs),  # $ routeSetup="/kw-args/(?P<x>[0-9]+)/(?P<y>[0-9]+)?"
+            # You can do nesting
+            (r"/(one|two|three)", [
+                (r"/one", One),  # $ routeSetup="/one"
+                (r"/two", Two),  # $ routeSetup="/two"
+                (r"/three", Three)  # $ routeSetup="/three"
+            ]),
+            # which is _one_ recommended way to ensure known host is used
+            (tornado.routing.HostMatches(r"(localhost|127\.0\.0\.1)"), [
+                ("/only-localhost", OnlyLocalhost)  # $ routeSetup="/only-localhost"
+            ]),
+
+        ],
+        debug=True,
+    )
+    app.add_handlers(r".*", [(r"/added-later/([0-9]+)/([0-9]+)?", AddedLater)])  # $ routeSetup="/added-later/([0-9]+)/([0-9]+)?"
+    return app
+
+
+if __name__ == "__main__":
+
+    import tornado.ioloop
+    app = make_app()
+    app.listen(8888)
+    tornado.ioloop.IOLoop.current().start()
+
+    # http://localhost:8888/foo/42/
+    # http://localhost:8888/foo/42/1337
+
+    # http://localhost:8888/bar/42/
+    # http://localhost:8888/bar/42/1337
+
+    # http://localhost:8888/baz/42/
+    # http://localhost:8888/baz/42/1337
+
+    # http://localhost:8888/kw-args/42/
+    # http://localhost:8888/kw-args/42/1337
+
+    # http://localhost:8888/only-localhost
+
+    # http://localhost:8888/one
+    # http://localhost:8888/two
+    # http://localhost:8888/three
+
+    # http://localhost:8888/added-later

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

@@ -0,0 +1,90 @@
+import tornado.web
+
+
+class TaintTest(tornado.web.RequestHandler):
+    def get(self, name = "World!", number="0", foo="foo"):  # $ requestHandler routedParameter=name routedParameter=number
+        ensure_tainted(name, number)
+        ensure_not_tainted(foo)
+
+        ensure_tainted(
+            # see https://www.tornadoweb.org/en/stable/web.html#input
+            self.get_argument("name"),
+            self.get_arguments("name"),
+            self.get_arguments("name")[0],
+
+            self.get_body_argument("name"),
+            self.get_body_arguments("name"),
+            self.get_body_arguments("name")[0],
+
+            self.get_query_argument("name"),
+            self.get_query_arguments("name"),
+            self.get_query_arguments("name")[0],
+
+            self.path_args,
+            self.path_args[0],
+
+            self.path_kwargs,
+            self.path_kwargs["name"],
+        )
+
+        request = self.request
+
+        ensure_tainted(
+            # see https://www.tornadoweb.org/en/stable/httputil.html#tornado.httputil.HTTPServerRequest
+            request,
+
+            request.uri,
+            request.path,
+            request.query,
+            request.full_url(),
+
+            request.remote_ip,
+
+            request.body,
+
+            request.arguments,
+            request.arguments["name"],
+            request.arguments["name"][0],
+
+            request.query_arguments,
+            request.query_arguments["name"],
+            request.query_arguments["name"][0],
+
+            request.body_arguments,
+            request.body_arguments["name"],
+            request.body_arguments["name"][0],
+
+            # dict-like, see https://www.tornadoweb.org/en/stable/httputil.html#tornado.httputil.HTTPHeaders
+            request.headers,
+            request.headers["header-name"],
+            request.headers.get_list("header-name"),
+            request.headers.get_all(),
+            [(k, v) for (k, v) in request.headers.get_all()],
+
+            # Dict[str, http.cookies.Morsel]
+            request.cookies,
+            request.cookies["cookie-name"],
+            request.cookies["cookie-name"].key,
+            request.cookies["cookie-name"].value,
+        )
+
+
+def make_app():
+    return tornado.web.Application(
+        [
+            (r"/test_taint/([^/]+)/([0-9]+)", TaintTest),  # $ routeSetup="/test_taint/([^/]+)/([0-9]+)"
+        ],
+        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