Merge pull request #7131 from RasmusWL/wsgiref.simple_server

Python: Model `wsgiref.simple_server` applications

python/ql/test/library-tests/frameworks/internal-ql-helpers/PoorMansFunctionResolutionTest.ql

@@ -0,0 +1,28 @@
+private import python
+private import semmle.python.dataflow.new.DataFlow
+private import semmle.python.frameworks.internal.PoorMansFunctionResolution
+import TestUtilities.InlineExpectationsTest
+
+class InlinePoorMansFunctionResolutionTest extends InlineExpectationsTest {
+  InlinePoorMansFunctionResolutionTest() { this = "InlinePoorMansFunctionResolutionTest" }
+
+  override string getARelevantTag() { result = "resolved" }
+
+  override predicate hasActualResult(Location location, string element, string tag, string value) {
+    exists(location.getFile().getRelativePath()) and
+    exists(Function func, DataFlow::Node ref |
+      ref = poorMansFunctionTracker(func) and
+      not ref.asExpr() instanceof FunctionExpr and
+      // exclude things like `GSSA variable func`
+      exists(ref.asExpr()) and
+      // exclude decorator calls (which with our extractor rewrites does reference the
+      // function)
+      not ref.asExpr() = func.getDefinition().(FunctionExpr).getADecoratorCall()
+    |
+      value = func.getName() and
+      tag = "resolved" and
+      element = ref.toString() and
+      location = ref.getLocation()
+    )
+  }
+}

python/ql/test/library-tests/frameworks/internal-ql-helpers/test.py

@@ -0,0 +1,42 @@
+def func():
+    print("func")
+
+func() # $ resolved=func
+
+
+class MyBase:
+    def base_method(self):
+        print("base_method", self)
+
+
+class MyClass(MyBase):
+    def method1(self):
+        print("method1", self)
+
+    @classmethod
+    def cls_method(cls):
+        print("cls_method", cls)
+
+    @staticmethod
+    def static():
+        print("static")
+
+    def method2(self):
+        print("method2", self)
+        self.method1() # $ resolved=method1
+        self.base_method()
+        self.cls_method() # $ resolved=cls_method
+        self.static() # $ resolved=static
+
+
+
+
+MyClass.cls_method() # $ resolved=cls_method
+MyClass.static() # $ resolved=static
+
+x = MyClass()
+x.base_method()
+x.method1()
+x.cls_method()
+x.static()
+x.method2()

python/ql/test/library-tests/frameworks/stdlib/wsgiref_simple_server_test.py

@@ -0,0 +1,57 @@
+# This test file demonstrates how to use an application with a wsgiref.simple_server
+# see https://docs.python.org/3/library/wsgiref.html#wsgiref.simple_server.WSGIServer
+import sys
+import wsgiref.simple_server
+
+def ignore(*arg, **kwargs): pass
+ensure_tainted = ensure_not_tainted = ignore
+
+ADDRESS = ("localhost", 8000)
+
+
+# I wanted to showcase that we handle both functions and bound-methods, so it's possible
+# to run this test-file in 2 different ways.
+
+def func(environ, start_response): # $ requestHandler
+    ensure_tainted(
+        environ, # $ tainted
+        environ["PATH_INFO"], # $ tainted
+    )
+    write = start_response("200 OK", [("Content-Type", "text/plain")])
+    write(b"hello") # $ HttpResponse responseBody=b"hello"
+    write(data=b" ") # $ HttpResponse responseBody=b" "
+
+    # function return value should be an iterable that will also be written to the
+    # response.
+    return [b"world", b"!"] # $ HttpResponse responseBody=List
+
+
+class MyServer(wsgiref.simple_server.WSGIServer):
+    def __init__(self):
+        super().__init__(ADDRESS, wsgiref.simple_server.WSGIRequestHandler)
+        self.set_app(self.my_method)
+
+    def my_method(self, _env, start_response): # $ requestHandler
+        start_response("200 OK", [])
+        return [b"my_method"] # $ HttpResponse responseBody=List
+
+
+case = sys.argv[1]
+if case == "1":
+    server = wsgiref.simple_server.WSGIServer(ADDRESS, wsgiref.simple_server.WSGIRequestHandler)
+    server.set_app(func)
+elif case == "2":
+    server = MyServer()
+elif case == "3":
+    server = MyServer()
+    def func3(_env, start_response): # $ requestHandler
+        start_response("200 OK", [])
+        return [b"foo"] # $ HttpResponse responseBody=List
+    server.set_app(func3)
+else:
+    sys.exit("wrong case")
+
+
+print(f"Running on http://{ADDRESS[0]}:{ADDRESS[1]}")
+
+server.serve_forever()