Python: Model RouteSetup for flask

python/ql/src/experimental/semmle/python/frameworks/Flask.qll

@@ -1,5 +1,6 @@
 /**
- * Provides classes modeling security-relevant aspects of the `flask` package.
+ * Provides classes modeling security-relevant aspects of the `flask` PyPI package.
+ * See https://flask.palletsprojects.com/en/1.1.x/.
  */
 
 private import python
@@ -11,6 +12,10 @@ private import experimental.semmle.python.frameworks.Werkzeug
 
 // for old improved impl see
 // https://github.com/github/codeql/blob/9f95212e103c68d0c1dfa4b6f30fb5d53954ccef/python/ql/src/semmle/python/web/flask/Request.qll
+/**
+ * Provides models for the `flask` PyPI package.
+ * See https://flask.palletsprojects.com/en/1.1.x/.
+ */
 private module Flask {
   /** Gets a reference to the `flask` module. */
   DataFlow::Node flask(DataFlow::TypeTracker t) {
@@ -23,6 +28,7 @@ private module Flask {
   /** Gets a reference to the `flask` module. */
   DataFlow::Node flask() { result = flask(DataFlow::TypeTracker::end()) }
 
+  /** Provides models for the `flask` module. */
   module flask {
     /** Gets a reference to the `flask.request` object. */
     DataFlow::Node request(DataFlow::TypeTracker t) {
@@ -32,13 +38,171 @@ private module Flask {
       t.startInAttr("request") and
       result = flask()
       or
-      exists(DataFlow::TypeTracker t2 | result = flask::request(t2).track(t2, t))
+      exists(DataFlow::TypeTracker t2 | result = request(t2).track(t2, t))
     }
 
     /** Gets a reference to the `flask.request` object. */
-    DataFlow::Node request() { result = flask::request(DataFlow::TypeTracker::end()) }
+    DataFlow::Node request() { result = request(DataFlow::TypeTracker::end()) }
+
+    /** Gets a reference to the `flask.Flask` class. */
+    private DataFlow::Node classFlask(DataFlow::TypeTracker t) {
+      t.start() and
+      result = DataFlow::importMember("flask", "Flask")
+      or
+      t.startInAttr("Flask") and
+      result = flask()
+      or
+      exists(DataFlow::TypeTracker t2 | result = classFlask(t2).track(t2, t))
+    }
+
+    /** Gets a reference to the `flask.Flask` class. */
+    DataFlow::Node classFlask() { result = classFlask(DataFlow::TypeTracker::end()) }
+
+    /** Gets a reference to an instance of `flask.Flask` (a Flask application). */
+    private DataFlow::Node app(DataFlow::TypeTracker t) {
+      t.start() and
+      result.asCfgNode().(CallNode).getFunction() = flask::classFlask().asCfgNode()
+      or
+      exists(DataFlow::TypeTracker t2 | result = app(t2).track(t2, t))
+    }
+
+    /** Gets a reference to an instance of `flask.Flask` (a flask application). */
+    DataFlow::Node app() { result = app(DataFlow::TypeTracker::end()) }
   }
 
+  // ---------------------------------------------------------------------------
+  // routing modeling
+  // ---------------------------------------------------------------------------
+  /**
+   * Gets a reference to the attribute `attr_name` of a flask application.
+   * WARNING: Only holds for a few predefined attributes.
+   */
+  private DataFlow::Node app_attr(DataFlow::TypeTracker t, string attr_name) {
+    attr_name in ["route", "add_url_rule"] and
+    t.startInAttr(attr_name) and
+    result = flask::app()
+    or
+    // Due to bad performance when using normal setup with `app_attr(t2, attr_name).track(t2, t)`
+    // we have inlined that code and forced a join
+    exists(DataFlow::TypeTracker t2 |
+      exists(DataFlow::StepSummary summary |
+        app_attr_first_join(t2, attr_name, result, summary) and
+        t = t2.append(summary)
+      )
+    )
+  }
+
+  pragma[nomagic]
+  private predicate app_attr_first_join(
+    DataFlow::TypeTracker t2, string attr_name, DataFlow::Node res, DataFlow::StepSummary summary
+  ) {
+    DataFlow::StepSummary::step(app_attr(t2, attr_name), res, summary)
+  }
+
+  /**
+   * Gets a reference to the attribute `attr_name` of a flask application.
+   * WARNING: Only holds for a few predefined attributes.
+   */
+  private DataFlow::Node app_attr(string attr_name) {
+    result = app_attr(DataFlow::TypeTracker::end(), attr_name)
+  }
+
+  private string werkzeug_rule_re() {
+    // since flask uses werkzeug internally, we are using its routing rules from
+    // https://github.com/pallets/werkzeug/blob/4dc8d6ab840d4b78cbd5789cef91b01e3bde01d5/src/werkzeug/routing.py#L138-L151
+    result =
+      "(?<static>[^<]*)<(?:(?<converter>[a-zA-Z_][a-zA-Z0-9_]*)(?:\\((?<args>.*?)\\))?\\:)?(?<variable>[a-zA-Z_][a-zA-Z0-9_]*)>"
+  }
+
+  /** A route setup made by flask (sharing handling of URL patterns). */
+  abstract private class FlaskRouteSetup extends HTTP::Server::RouteSetup::Range {
+    override Parameter getARoutedParameter() {
+      exists(string name |
+        result = this.getARouteHandler().getArgByName(name) and
+        exists(string match |
+          match = this.getUrlPattern().regexpFind(werkzeug_rule_re(), _, _) and
+          name = match.regexpCapture(werkzeug_rule_re(), 4)
+        )
+      )
+    }
+
+    /** Gets the argument used to pass in the URL pattern. */
+    abstract DataFlow::Node getUrlPatternArg();
+
+    override string getUrlPattern() {
+      exists(StrConst str |
+        DataFlow::localFlow(DataFlow::exprNode(str), this.getUrlPatternArg()) and
+        result = str.getText()
+      )
+    }
+  }
+
+  /**
+   * A call to `flask.Flask.route`.
+   *
+   * See https://flask.palletsprojects.com/en/1.1.x/api/#flask.Flask.route
+   */
+  private class FlaskAppRouteCall extends FlaskRouteSetup {
+    CallNode call;
+
+    FlaskAppRouteCall() {
+      call.getFunction() = app_attr("route").asCfgNode() and
+      this.asCfgNode() = call
+    }
+
+    override DataFlow::Node getUrlPatternArg() {
+      exists(ControlFlowNode pattern_arg |
+        (
+          pattern_arg = call.getArg(0)
+          or
+          pattern_arg = call.getArgByName("rule")
+        ) and
+        result.asCfgNode() = pattern_arg
+      )
+    }
+
+    override Function getARouteHandler() { result.getADecorator() = call.getNode() }
+  }
+
+  /**
+   * A call to `flask.Flask.add_url_rule`.
+   *
+   * See https://flask.palletsprojects.com/en/1.1.x/api/#flask.Flask.add_url_rule
+   */
+  private class FlaskAppAddUrlRule extends FlaskRouteSetup {
+    CallNode call;
+
+    FlaskAppAddUrlRule() {
+      call.getFunction() = app_attr("add_url_rule").asCfgNode() and
+      this.asCfgNode() = call
+    }
+
+    override DataFlow::Node getUrlPatternArg() {
+      exists(ControlFlowNode pattern_arg |
+        (
+          pattern_arg = call.getArg(0)
+          or
+          pattern_arg = call.getArgByName("rule")
+        ) and
+        result.asCfgNode() = pattern_arg
+      )
+    }
+
+    override Function getARouteHandler() {
+      exists(ControlFlowNode view_func_arg, DataFlow::Node func_src |
+        view_func_arg = call.getArg(2)
+        or
+        view_func_arg = call.getArgByName("view_func")
+      |
+        DataFlow::localFlow(func_src, any(DataFlow::Node dest | dest.asCfgNode() = view_func_arg)) and
+        func_src.asExpr().(CallableExpr) = result.getDefinition()
+      )
+    }
+  }
+
+  // ---------------------------------------------------------------------------
+  // flask.Request taint modeling
+  // ---------------------------------------------------------------------------
   // TODO: Do we even need this class? :|
   /**
    * A source of remote flow from a flask request.

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

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

python/ql/test/experimental/library-tests/frameworks/flask/old_test.py

@@ -3,15 +3,15 @@ import flask
 from flask import Flask, request, make_response
 app = Flask(__name__)
 
-@app.route("/")
-def hello_world():
+@app.route("/")  # $routeSetup="/"
+def hello_world():  # $routeHandler
     return "Hello World!"
 
 from flask.views import MethodView
 
 class MyView(MethodView):
 
-    def get(self, user_id):
+    def get(self, user_id): # $f-:routeHandler
         if user_id is None:
             # return a list of users
             pass
@@ -21,46 +21,46 @@ class MyView(MethodView):
 
 the_view = MyView.as_view('my_view')
 
-app.add_url_rule('/the/', defaults={'user_id': None},
+app.add_url_rule('/the/', defaults={'user_id': None},  # $routeSetup="/the/"
                  view_func=the_view, methods=['GET',])
 
-@app.route("/dangerous")
-def dangerous():
+@app.route("/dangerous")  # $routeSetup="/dangerous"
+def dangerous():  # $routeHandler
     return request.args.get('payload')
 
-@app.route("/dangerous-with-cfg-split")
-def dangerous2():
+@app.route("/dangerous-with-cfg-split")  # $routeSetup="/dangerous-with-cfg-split"
+def dangerous2():  # $routeHandler
     x = request.form['param0']
     if request.method == "POST":
         return request.form['param1']
     return None
 
-@app.route('/unsafe')
-def unsafe():
+@app.route("/unsafe")  # $routeSetup="/unsafe"
+def unsafe():  # $routeHandler
     first_name = request.args.get('name', '')
     return make_response("Your name is " + first_name)
 
-@app.route('/safe')
-def safe():
+@app.route("/safe")  # $routeSetup="/safe"
+def safe():  # $routeHandler
     first_name = request.args.get('name', '')
     return make_response("Your name is " + escape(first_name))
 
-@app.route('/hello/<name>')
-def hello(name):
+@app.route("/hello/<name>")  # $routeSetup="/hello/<name>"
+def hello(name):  # $routeHandler $routedParameter=name
     return make_response("Your name is " + name)
 
-@app.route('/foo/<path:subpath>')
-def foo(subpath):
+@app.route("/foo/<path:subpath>")  # $routeSetup="/foo/<path:subpath>"
+def foo(subpath):  # $routeHandler $routedParameter=subpath
     return make_response("The subpath is " + subpath)
 
-@app.route('/multiple/') # TODO: not recognized as route
-@app.route('/multiple/foo/<foo>') # TODO: not recognized as route
-@app.route('/multiple/bar/<bar>')
-def multiple(foo=None, bar=None):
+@app.route("/multiple/")  # $routeSetup="/multiple/"
+@app.route("/multiple/foo/<foo>")  # $routeSetup="/multiple/foo/<foo>"
+@app.route("/multiple/bar/<bar>")  # $routeSetup="/multiple/bar/<bar>"
+def multiple(foo=None, bar=None):  # $routeHandler $routedParameter=foo $routedParameter=bar
     return make_response("foo={!r} bar={!r}".format(foo, bar))
 
-@app.route('/complex/<string(length=2):lang_code>')
-def complex(lang_code):
+@app.route("/complex/<string(length=2):lang_code>")  # $routeSetup="/complex/<string(length=2):lang_code>"
+def complex(lang_code):  # $routeHandler $routedParameter=lang_code
     return make_response("lang_code {}".format(lang_code))
 
 if __name__ == "__main__":

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

@@ -0,0 +1,27 @@
+import flask
+
+from flask import Flask, make_response
+app = Flask(__name__)
+
+
+SOME_ROUTE = "/some/route"
+@app.route(SOME_ROUTE) # $routeSetup="/some/route"
+def some_route():  # $routeHandler
+    return make_response("some_route")
+
+
+# TODO: We should be able to handle this one
+def index():  # $routeHandler
+    return make_response("index")
+app.add_url_rule('/index', 'index', index)  # $routeSetup="/index"
+
+
+# We don't support this yet, and I think that's OK
+def later_set():  # $f-:routeHandler
+    return make_response("later_set")
+app.add_url_rule('/later-set', 'later_set', view_func=None)  # $routeSetup="/later-set"
+app.view_functions['later_set'] = later_set
+
+
+if __name__ == "__main__":
+    app.run(debug=True)

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

@@ -1,8 +1,8 @@
 from flask import Flask, request
 app = Flask(__name__)
 
-@app.route('/test_taint/<name>/<int:number>')
-def test_taint(name = "World!", number="0", foo="foo"):
+@app.route("/test_taint/<name>/<int:number>")  # $routeSetup="/test_taint/<name>/<int:number>"
+def test_taint(name = "World!", number="0", foo="foo"):  # $routeHandler $routedParameter=name $routedParameter=number
     ensure_tainted(name, number)
     ensure_not_tainted(foo)
 
@@ -191,8 +191,8 @@ def test_taint(name = "World!", number="0", foo="foo"):
 
 
 
-@app.route('/debug/<foo>/<bar>', methods=['GET'])
-def debug(foo, bar):
+@app.route("/debug/<foo>/<bar>", methods=['GET']) # $routeSetup="/debug/<foo>/<bar>"
+def debug(foo, bar):  # $routeHandler $routedParameter=foo $routedParameter=bar
     print("request.view_args", request.view_args)
 
     print("request.headers {!r}".format(request.headers))
@@ -202,8 +202,8 @@ def debug(foo, bar):
 
     return 'ok'
 
-@app.route('/stream', methods=['POST'])
-def stream():
+@app.route("/stream", methods=['POST'])  # $routeSetup="/stream"
+def stream():  # $routeHandler
     print(request.path)
     s = request.stream
     print(s)
@@ -212,8 +212,8 @@ def stream():
 
     return 'ok'
 
-@app.route('/input_stream', methods=['POST'])
-def input_stream():
+@app.route("/input_stream", methods=['POST'])  # $routeSetup="/input_stream"
+def input_stream():  # $routeHandler
     print(request.path)
     s = request.input_stream
     print(s)
@@ -223,15 +223,15 @@ def input_stream():
 
     return 'ok'
 
-@app.route('/form', methods=['POST'])
-def form():
+@app.route("/form", methods=['POST'])  # $routeSetup="/form"
+def form():  # $routeHandler
     print(request.path)
     print("request.form", request.form)
 
     return 'ok'
 
-@app.route('/cache_control', methods=['POST'])
-def cache_control():
+@app.route("/cache_control", methods=['POST'])  # $routeSetup="/cache_control"
+def cache_control():  # $routeHandler
     print(request.path)
     print("request.cache_control.max_age", request.cache_control.max_age, type(request.cache_control.max_age))
     print("request.cache_control.max_stale", request.cache_control.max_stale, type(request.cache_control.max_stale))
@@ -239,16 +239,16 @@ def cache_control():
 
     return 'ok'
 
-@app.route('/file_upload', methods=['POST'])
-def file_upload():
+@app.route("/file_upload", methods=['POST'])  # $routeSetup="/file_upload"
+def file_upload():  # $routeHandler
     print(request.path)
     for k,v in request.files.items():
         print(k, v, v.name, v.filename, v.stream)
 
     return 'ok'
 
-@app.route('/args', methods=['GET'])
-def args():
+@app.route("/args", methods=['GET'])  # $routeSetup="/args"
+def args():  # $routeHandler
     print(request.path)
     print("request.args", request.args)