Add taint test

Author: joefarebrother

python/ql/lib/semmle/python/frameworks/Socketio.qll

@@ -30,6 +30,15 @@ module SocketIO {
   private class EventHandler extends Http::Server::RequestHandler::Range {
     EventHandler() {
       serverEventAnnotation().getAValueReachableFromSource().asExpr() = this.getADecorator()
+      or
+      exists(DataFlow::CallCfgNode c, DataFlow::Node arg | c = server().getMember("on").getACall() |
+        (
+          arg = c.getArg(1)
+          or
+          arg = c.getArgByName("handler")
+        ) and
+        poorMansFunctionTracker(this) = arg
+      )
     }
 
     override Parameter getARoutedParameter() {
@@ -44,20 +53,13 @@ module SocketIO {
       exists(DataFlow::CallCfgNode c | c = server().getMember(["emit", "send"]).getACall() |
         this = c.getArgByName("callback")
       )
-      or
-      exists(DataFlow::CallCfgNode c | c = server().getMember("on").getACall() |
-        this = c.getArg(1) or
-        this = c.getArgByName("handler")
-      )
     }
   }
 
   private class CallbackHandler extends Http::Server::RequestHandler::Range {
     CallbackHandler() { any(CallbackArgument ca) = poorMansFunctionTracker(this) }
 
-    override Parameter getARoutedParameter() {
-      result = this.getAnArg() and not result = this.getArg(0)
-    }
+    override Parameter getARoutedParameter() { result = this.getAnArg() }
 
     override string getFramework() { result = "socketio" }
   }

python/ql/test/library-tests/frameworks/socketio/InlineTaintTest.expected

@@ -0,0 +1,3 @@
+argumentToEnsureNotTaintedNotMarkedAsSpurious
+untaintedArgumentToEnsureTaintedNotMarkedAsMissing
+testFailures

python/ql/test/library-tests/frameworks/socketio/InlineTaintTest.ql

@@ -0,0 +1,2 @@
+import experimental.meta.InlineTaintTest
+import MakeInlineTaintTest<TestTaintTrackingConfig>

python/ql/test/library-tests/frameworks/socketio/taint_test.py

@@ -0,0 +1,48 @@
+import sys
+import socketio 
+import sys
+
+def ensure_tainted(*args):
+    print("tainted", args)
+
+def ensure_not_tainted(*args):
+    print("not tainted", args)
+
+sio = socketio.Server()
+
+@sio.event
+def connect(sid, environ, auth): # $ requestHandler routedParameter=sid routedParameter=environ routedParameter=auth
+    ensure_not_tainted(sid)
+    ensure_tainted(environ,  # $ tainted
+                   auth)     # $ tainted
+
+@sio.event
+def event1(sid, data): # $ requestHandler routedParameter=sid routedParameter=data
+    ensure_not_tainted(sid)
+    ensure_tainted(data)  # $ tainted
+    res = sio.call("e1", sid=sid)
+    ensure_tainted(res)  # $ tainted
+    sio.emit("e2", "hi", to=sid, callback=lambda x: ensure_tainted(x))   # $ tainted
+    sio.send("hi", to=sid, callback=lambda x: ensure_tainted(x))   # $ tainted
+
+asio = socketio.AsyncServer(async_mode='asgi')
+
+@asio.event
+async def event2(sid, data): # $ requestHandler routedParameter=sid routedParameter=data
+    ensure_not_tainted(sid)
+    ensure_tainted(data)  # $ tainted
+    res = await asio.call("e2", sid=sid)
+    ensure_tainted(res)  # $ tainted
+    await asio.emit("e3", "hi", to=sid, callback=lambda x: ensure_tainted(x))   # $ tainted
+    await asio.send("hi", to=sid, callback=lambda x: ensure_tainted(x))   # $ tainted
+
+if __name__ == "__main__":
+    
+    if "--async" in sys.argv:
+        import uvicorn
+        app = socketio.ASGIApp(asio)
+        uvicorn.run(app, host='127.0.0.1', port=8000)
+    else:
+        import eventlet
+        app = socketio.WSGIApp(sio)
+        eventlet.wsgi.server(eventlet.listen(('', 8000)), app)

python/ql/test/library-tests/frameworks/socketio/test.py

@@ -3,23 +3,23 @@
 sio = socketio.Server()
 
 @sio.on("connect")
-def connect(sid, environ, auth): # $ requestHandler routedParameter=environ routedParameter=auth
+def connect(sid, environ, auth): # $ requestHandler routedParameter=sid routedParameter=environ routedParameter=auth
     print("connect", sid, environ, auth)
 
 @sio.on("event1")
-def handle(sid, data): # $ requestHandler routedParameter=data
+def handle(sid, data): # $ requestHandler routedParameter=sid routedParameter=data
     print("e1", sid, data)
 
 @sio.event
-def event2(sid, data): # $ requestHandler routedParameter=data
+def event2(sid, data): # $ requestHandler routedParameter=sid routedParameter=data
     print("e2", sid, data)
 
-def event3(sid, data): # $ requestHandler routedParameter=data
+def event3(sid, data): # $ requestHandler routedParameter=sid routedParameter=data
     print("e3", sid, data)
 
 sio.on("event3", handler=event3)
 
-sio.on("event4", lambda sid,data: print("e4", sid, data)) # $ requestHandler routedParameter=data
+sio.on("event4", lambda sid,data: print("e4", sid, data)) # $ requestHandler routedParameter=sid routedParameter=data