add twisted SSH client as secondary server command injection sinks, add proper test cases

python/ql/test/experimental/query-tests/Security/CWE-074-SecondaryServerCmdInjection/Twisted.py

@@ -0,0 +1,30 @@
+#!/usr/bin/env python
+
+from fastapi import FastAPI
+from twisted.conch.endpoints import SSHCommandClientEndpoint
+from twisted.internet.protocol import Factory
+from twisted.internet import reactor
+
+
+app = FastAPI()
+
+
+@app.get("/bad1")
+async def bad1(cmd: bytes):
+    endpoint = SSHCommandClientEndpoint.newConnection(
+    reactor,
+    cmd, # $ result=BAD getSecondaryCommand=cmd  
+    b"username",
+    b"ssh.example.com",
+    22,
+    password=b"password") 
+
+    SSHCommandClientEndpoint.existingConnection(
+        endpoint,
+        cmd) # $ result=BAD getSecondaryCommand=cmd
+    
+    factory = Factory()
+    d = endpoint.connect(factory)
+    d.addCallback(lambda protocol: protocol.finished)
+
+    return {"success": "Dangerous"}

python/ql/test/experimental/query-tests/Security/CWE-074-SecondaryServerCmdInjection/paramiko.py

@@ -14,9 +14,9 @@ app = FastAPI()
 @app.get("/bad1")
 async def bad1(cmd: str):
     stdin, stdout, stderr = paramiko_ssh_client.exec_command(cmd)  # $ result=BAD getSecondaryCommand=cmd
-    return {"success": stdout}
+    return {"success": "Dangerous"}
 
 @app.get("/bad2")
 async def bad2(cmd: str):
     stdin, stdout, stderr = paramiko_ssh_client.exec_command(command=cmd)  # $ result=BAD getSecondaryCommand=cmd
-    return {"success": "OK"}
+    return {"success": "Dangerous"}