feat(python): add iast source tests for python (#1353)

* feat(python): add iast source tests for python

* fix codestyle

* feat: update iast version

* feat: remove prints

* feat: update iast version

tests/appsec/iast/source/test_body.py

@@ -11,7 +11,8 @@
 
 
 @coverage.basic
-@released(dotnet="?", golang="?", php_appsec="?", python="?", ruby="?")
+@released(dotnet="?", golang="?", php_appsec="?", ruby="?")
+@released(python={"flask-poc": "?", "uwsgi-poc": "?", "django-poc": "?", "uds-flask": "?"})
 @released(
     java={
         "jersey-grizzly2": "?",

tests/appsec/iast/source/test_cookie_name.py

@@ -11,7 +11,7 @@
 
 
 @coverage.basic
-@released(dotnet="?", golang="?", nodejs="?", php_appsec="?", python="?", ruby="?")
+@released(dotnet="?", golang="?", nodejs="?", php_appsec="?", python="1.18.0", ruby="?")
 @released(
     java={
         "resteasy-netty3": "?",

tests/appsec/iast/source/test_cookie_value.py

@@ -12,7 +12,7 @@
 
 
 @coverage.basic
-@released(dotnet="?", golang="?", nodejs="?", php_appsec="?", python="?", ruby="?")
+@released(dotnet="?", golang="?", nodejs="?", php_appsec="?", python="1.18.0", ruby="?")
 @released(
     java={
         "resteasy-netty3": "1.11.0",

tests/appsec/iast/source/test_header_name.py

@@ -11,7 +11,7 @@
 
 
 @coverage.basic
-@released(dotnet="?", golang="?", nodejs="?", php_appsec="?", python="?", ruby="?")
+@released(dotnet="?", golang="?", nodejs="?", php_appsec="?", python="1.18.0", ruby="?")
 @released(
     java={
         "jersey-grizzly2": "1.15.0",
@@ -27,12 +27,16 @@
 class TestHeaderName:
     """Verify that request headers name are tainted"""
 
+    source_name = "user"
+    if context.library.library == "python":
+        source_name = "User"
+
     source_fixture = SourceFixture(
         http_method="GET",
         endpoint="/iast/source/headername/test",
         request_kwargs={"headers": {"user": "unused"}},
         source_type="http.request.header.name",
-        source_name="user",
+        source_name=source_name,
         source_value=None,
     )
 

tests/appsec/iast/source/test_header_value.py

@@ -12,7 +12,7 @@
 
 
 @coverage.basic
-@released(dotnet="?", golang="?", nodejs="?", php_appsec="?", python="?", ruby="?")
+@released(dotnet="?", golang="?", nodejs="?", php_appsec="?", python="1.18.0", ruby="?")
 @released(
     java={
         "resteasy-netty3": "1.11.0",
@@ -28,12 +28,16 @@
 class TestHeaderValue:
     """Verify that request headers are tainted"""
 
+    source_name = "table"
+    if context.library.library == "python" and context.weblog_variant == "django-poc":
+        source_name = "HTTP_TABLE"
+
     source_fixture = SourceFixture(
         http_method="GET",
         endpoint="/iast/source/header/test",
         request_kwargs={"headers": {"table": "user"}},
         source_type="http.request.header",
-        source_name="table",
+        source_name=source_name,
         source_value="user",
     )
 
@@ -47,8 +51,10 @@ def test_source_reported(self):
     def setup_telemetry_metric_instrumented_source(self):
         self.source_fixture.setup_telemetry_metric_instrumented_source()
 
-    @missing_feature(context.library < "java@1.13.0", reason="Not implemented")
-    @missing_feature(not context.weblog_variant.startswith("spring-boot"), reason="Not implemented")
+    @missing_feature(
+        context.library < "java@1.13.0" or not context.weblog_variant.startswith("spring-boot"),
+        reason="Not implemented",
+    )
     def test_telemetry_metric_instrumented_source(self):
         self.source_fixture.test_telemetry_metric_instrumented_source()
 

tests/appsec/iast/source/test_parameter_name.py

@@ -11,7 +11,7 @@
 
 
 @coverage.basic
-@released(dotnet="?", golang="?", nodejs="?", php_appsec="?", python="?", ruby="?")
+@released(dotnet="?", golang="?", nodejs="?", php_appsec="?", ruby="?")
 @released(
     java={
         "jersey-grizzly2": "1.15.0",
@@ -22,6 +22,7 @@
         "*": "1.5.0",
     }
 )
+@released(python={"flask-poc": "?", "uwsgi-poc": "?", "django-poc": "1.18.0", "uds-flask": "?"})
 @missing_feature(weblog_variant="spring-boot-3-native", reason="GraalVM. Tracing support only")
 class TestParameterName:
     """Verify that request parameters are tainted"""
@@ -40,6 +41,7 @@ def setup_source_post_reported(self):
 
     @missing_feature(weblog_variant="express4", reason="Tainted as request body")
     @bug(weblog_variant="resteasy-netty3", reason="Not reported")
+    @bug(library="python", reason="Python frameworks need a header, if not, 415 status code")
     def test_source_post_reported(self):
         self.source_post_fixture.test()
 

tests/appsec/iast/source/test_parameter_value.py

@@ -11,7 +11,7 @@
 
 
 @coverage.basic
-@released(dotnet="?", golang="?", php_appsec="?", python="?", ruby="?")
+@released(dotnet="?", golang="?", php_appsec="?", python="1.18.0", ruby="?")
 @released(
     java={
         "resteasy-netty3": "1.11.0",
@@ -47,6 +47,7 @@ def setup_source_post_reported(self):
         self.source_post_fixture.setup()
 
     @bug(weblog_variant="jersey-grizzly2", reason="name field of source not set")
+    @bug(library="python", reason="Python frameworks need a header, if not, 415 status code")
     def test_source_post_reported(self):
         self.source_post_fixture.test()
 
@@ -70,7 +71,9 @@ def setup_post_telemetry_metric_instrumented_source(self):
         self.source_post_fixture.setup_telemetry_metric_instrumented_source()
 
     @missing_feature(context.library < "java@1.13.0", reason="Not implemented")
-    @missing_feature(not context.weblog_variant.startswith("spring-boot"), reason="Not implemented")
+    @missing_feature(
+        context.library == "java" and not context.weblog_variant.startswith("spring-boot"), reason="Not implemented"
+    )
     @missing_feature(library="nodejs", reason="Not implemented")
     def test_post_telemetry_metric_instrumented_source(self):
         self.source_post_fixture.test_telemetry_metric_instrumented_source()
@@ -79,7 +82,9 @@ def setup_post_telemetry_metric_executed_source(self):
         self.source_post_fixture.setup_telemetry_metric_executed_source()
 
     @missing_feature(context.library < "java@1.13.0", reason="Not implemented")
-    @missing_feature(not context.weblog_variant.startswith("spring-boot"), reason="Not implemented")
+    @missing_feature(
+        context.library == "java" and not context.weblog_variant.startswith("spring-boot"), reason="Not implemented"
+    )
     @missing_feature(library="nodejs", reason="Not implemented")
     def test_post_telemetry_metric_executed_source(self):
         self.source_post_fixture.test_telemetry_metric_executed_source()

utils/build/docker/python/django/django.app.urls.py

@@ -147,6 +147,70 @@ def view_sqli_secure(request):
     return HttpResponse("OK")
 
 
+def _sink_point(table="user", id="1"):
+    sql = "SELECT * FROM " + table + " WHERE id = '" + id + "'"
+    with connection.cursor() as cursor:
+        cursor.execute(sql)
+
+
+@csrf_exempt
+def view_iast_source_body(request):
+    # TODO: migrate to a django rest framework view with request.data
+    import json
+
+    table = json.loads(request.body).get("name")
+    user = json.loads(request.body).get("value")
+    _sink_point(table=table, id=user)
+    return HttpResponse("OK")
+
+
+def view_iast_source_cookie_name(request):
+    param = [key for key in request.COOKIES.keys() if key == "user"]
+    _sink_point(id=param[0])
+    return HttpResponse("OK")
+
+
+def view_iast_source_cookie_value(request):
+    table = request.COOKIES.get("table")
+    _sink_point(table=table)
+    return HttpResponse("OK")
+
+
+def view_iast_source_header_name(request):
+    param = [key for key in request.headers.keys() if key == "User"]
+    # param = [key for key in request.META.keys() if key == "HTTP_USER"]
+    _sink_point(id=param[0])
+    return HttpResponse("OK")
+
+
+def view_iast_source_header_value(request):
+    table = request.META.get("HTTP_TABLE")
+    _sink_point(table=table)
+    return HttpResponse("OK")
+
+
+def view_iast_source_parametername(request):
+    if request.method == "GET":
+        param = [key for key in request.GET.keys() if key == "user"]
+        _sink_point(id=param[0])
+    elif request.method == "POST":
+        param = [key for key in request.POST.keys() if key == "user"]
+        _sink_point(id=param[0])
+    return HttpResponse("OK")
+
+
+@csrf_exempt
+def view_iast_source_parameter(request):
+    if request.method == "GET":
+        table = request.GET.get("table")
+        _sink_point(table=table[0])
+    elif request.method == "POST":
+        table = request.POST.get("table")
+        _sink_point(table=table[0])
+
+    return HttpResponse("OK")
+
+
 def make_distant_call(request):
     # curl localhost:7777/make_distant_call?url=http%3A%2F%2Fweblog%3A7777 | jq
 
@@ -244,6 +308,13 @@ def get_value(request):
     path("iast/insecure_cipher/test_secure_algorithm", view_weak_cipher_secure),
     path("iast/sqli/test_secure", view_sqli_secure),
     path("iast/sqli/test_insecure", view_sqli_insecure),
+    path("iast/source/body/test", view_iast_source_body),
+    path("iast/source/cookiename/test", view_iast_source_cookie_name),
+    path("iast/source/cookievalue/test", view_iast_source_cookie_value),
+    path("iast/source/headername/test", view_iast_source_header_name),
+    path("iast/source/header/test", view_iast_source_header_value),
+    path("iast/source/parametername/test", view_iast_source_parametername),
+    path("iast/source/parameter/test", view_iast_source_parameter),
     path("make_distant_call", make_distant_call),
     path("user_login_success_event", track_user_login_success_event),
     path("user_login_failure_event", track_user_login_failure_event),

utils/build/docker/python/flask/app.py

@@ -191,6 +191,73 @@ def view_weak_cipher_secure():
     return Response("OK")
 
 
+def _sink_point(table="user", id="1"):
+    sql = "SELECT * FROM " + table + " WHERE id = '" + id + "'"
+    postgres_db = psycopg2.connect(**POSTGRES_CONFIG)
+    cursor = postgres_db.cursor()
+    cursor.execute(sql)
+
+
+@app.route("/iast/source/body/test", methods=["POST"])
+def view_iast_source_body():
+    table = flask_request.json.get("name")
+    user = flask_request.json.get("value")
+    _sink_point(table=table, id=user)
+    return Response("OK")
+
+
+@app.route("/iast/source/cookiename/test")
+def view_iast_source_cookie_name():
+    param = [key for key in flask_request.cookies.keys() if key == "user"]
+    _sink_point(id=param[0])
+    return Response("OK")
+
+
+@app.route("/iast/source/cookievalue/test")
+def view_iast_source_cookie_value():
+    table = flask_request.cookies.get("table")
+    _sink_point(table=table)
+    return Response("OK")
+
+
+@app.route("/iast/source/headername/test")
+def view_iast_source_header_name():
+    param = [key for key in flask_request.headers.keys() if key == "User"]
+    _sink_point(id=param[0])
+    return Response("OK")
+
+
+@app.route("/iast/source/header/test")
+def view_iast_source_header_value():
+    table = flask_request.headers.get("table")
+    _sink_point(table=table)
+    return Response("OK")
+
+
+@app.route("/iast/source/parametername/test", methods=["GET"])
+def view_iast_source_parametername_get():
+    param = [key for key in flask_request.args.keys() if key == "user"]
+    _sink_point(id=param[0])
+    return Response("OK")
+
+
+@app.route("/iast/source/parametername/test", methods=["POST"])
+def view_iast_source_parametername_post():
+    param = [key for key in flask_request.json.keys() if key == "user"]
+    _sink_point(id=param[0])
+    return Response("OK")
+
+
+@app.route("/iast/source/parameter/test", methods=["GET", "POST"])
+def view_iast_source_parameter():
+    if flask_request.args:
+        table = flask_request.args.get("table")
+    else:
+        table = flask_request.json.get("table")
+    _sink_point(table=table)
+    return Response("OK")
+
+
 _TRACK_METADATA = {
     "metadata0": "value0",
     "metadata1": "value1",