Add models for responses

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

@@ -4,7 +4,6 @@
  */
 
 private import python
-private import semmle.python.dataflow.new.DataFlow
 private import semmle.python.dataflow.new.RemoteFlowSources
 private import semmle.python.dataflow.new.TaintTracking
 private import semmle.python.Concepts
@@ -141,4 +140,95 @@ module Pyramid {
       }
     }
   }
+
+  module Response {
+    private class PyramidReturnResponse extends Http::Server::HttpResponse::Range {
+      PyramidReturnResponse() {
+        this.asCfgNode() = any(View::ViewCallable vc).getAReturnValueFlowNode() and
+        not this = instance()
+      }
+
+      override DataFlow::Node getBody() { result = this }
+
+      override DataFlow::Node getMimetypeOrContentTypeArg() { none() }
+
+      override string getMimetypeDefault() { result = "text/html" }
+    }
+
+    private API::Node classRef() {
+      result = API::moduleImport("pyramid").getMember("response").getMember("Response")
+    }
+
+    abstract class InstanceSource extends DataFlow::LocalSourceNode,
+      Http::Server::HttpResponse::Range
+    { }
+
+    /** Gets a reference to an instance of `pyramid.request.Request`. */
+    private DataFlow::TypeTrackingNode instance(DataFlow::TypeTracker t) {
+      t.start() and
+      result instanceof InstanceSource
+      or
+      exists(DataFlow::TypeTracker t2 | result = instance(t2).track(t2, t))
+    }
+
+    /** Gets a reference to an instance of `pyramid.request.Request`. */
+    DataFlow::Node instance() { instance(DataFlow::TypeTracker::end()).flowsTo(result) }
+
+    private class ClassInstantiation extends InstanceSource, DataFlow::CallCfgNode {
+      ClassInstantiation() { this = classRef().getACall() }
+
+      override DataFlow::Node getBody() { result = [this.getArg(0), this.getArgByName("body")] }
+
+      override DataFlow::Node getMimetypeOrContentTypeArg() {
+        result = [this.getArg(4), this.getArgByName("content_type")]
+      }
+
+      override string getMimetypeDefault() { result = "text/html" }
+    }
+
+    private class ResponseBodySet extends Http::Server::HttpResponse::Range instanceof DataFlow::AttrWrite
+    {
+      string attrName;
+
+      ResponseBodySet() {
+        this.getObject() = instance() and
+        this.getAttributeName() = attrName and
+        attrName in ["body", "body_file", "json", "json_body", "text", "ubody", "unicode_body"]
+      }
+
+      override DataFlow::Node getBody() { result = this.(DataFlow::AttrWrite).getValue() }
+
+      override DataFlow::Node getMimetypeOrContentTypeArg() { none() }
+
+      override string getMimetypeDefault() {
+        if attrName in ["json", "json_body"]
+        then result = "application/json"
+        else result = "text/html"
+      }
+    }
+
+    private class RequestResponseAttr extends InstanceSource instanceof DataFlow::AttrRead {
+      RequestResponseAttr() {
+        this.getObject() = Request::instance() and this.getAttributeName() = "response"
+      }
+
+      override DataFlow::Node getBody() { none() }
+
+      override DataFlow::Node getMimetypeOrContentTypeArg() { none() }
+
+      override string getMimetypeDefault() { result = "text/html" }
+    }
+
+    private class SetCookieCall extends Http::Server::CookieWrite::Range, DataFlow::MethodCallNode {
+      SetCookieCall() { this.calls(instance(), "set_cookie") }
+
+      override DataFlow::Node getHeaderArg() { none() }
+
+      override DataFlow::Node getNameArg() { result = [this.getArg(0), this.getArgByName("name")] }
+
+      override DataFlow::Node getValueArg() {
+        result = [this.getArg(1), this.getArgByName("value")]
+      }
+    }
+  }
 }

python/ql/test/library-tests/frameworks/pyramid/ConceptsTest.expected

@@ -0,0 +1,2 @@
+testFailures
+failures

python/ql/test/library-tests/frameworks/pyramid/ConceptsTest.ql

@@ -0,0 +1,2 @@
+import python
+import experimental.meta.ConceptsTest

python/ql/test/library-tests/frameworks/pyramid/pyramid_test.py

@@ -1,8 +1,13 @@
 from pyramid.view import view_config
 from pyramid.config import Configurator
+from pyramid.response import Response
+from wsgiref.simple_server import make_server
+
+def ignore(*args, **kwargs): pass
+ensure_tainted = ensure_not_tainted = ignore
 
 @view_config(route_name="test1")
-def test1(request):
+def test1(request): # $ requestHandler routedParameter=request
     ensure_tainted(
         request,                   # $ tainted
 
@@ -65,13 +70,29 @@ def test1(request):
         request.copy_get().body     # $ tainted
         ) 
 
-def test2(request):
+    return Response("Ok") # $ HttpResponse responseBody="Ok" mimetype=text/html
+
+def test2(request): # $ requestHandler routedParameter=request
     ensure_tainted(request) # $ tainted
 
-@view_config(route_name="test1")
-def test3(context, request):
+    resp = Response("Ok", content_type="text/plain") # $ HttpResponse responseBody="Ok" mimetype=text/plain
+    resp.body = "Ok2" # $ HttpResponse responseBody="Ok2" SPURIOUS: mimetype=text/html
+    return resp
+
+@view_config(route_name="test3", renderer="string")
+def test3(context, request): # $ requestHandler routedParameter=request
     ensure_tainted(request) # $ tainted
+    resp = request.response # $ HttpResponse mimetype=text/html
+    resp.set_cookie("hi", "there") # $ CookieWrite CookieName="hi" CookieValue="there"
+    resp.set_cookie(value="there", name="hi") # $ CookieWrite CookieName="hi" CookieValue="there"
+    return "Ok" # $ HttpResponse responseBody="Ok" mimetype=text/html
 
 if __name__ == "__main__":
     with Configurator() as config:
-        config.add_view(test2, route_name="test2")
+        for i in range(1,4):
+            config.add_route(f"test{i}", f"/test{i}")
+        config.add_view(test2, route_name="test2")
+        config.scan()
+        server = make_server('127.0.0.1', 8000, config.make_wsgi_app())
+        print("serving")
+        server.serve_forever()