[python] code injection

manifests/python.yml

@@ -64,7 +64,7 @@ tests/:
     iast/:
       sink/:
         test_code_injection.py:
-          TestCodeInjection: missing_feature
+          TestCodeInjection: v2.20.0
           TestCodeInjection_StackTrace: missing_feature
         test_command_injection.py:
           TestCommandInjection:

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

@@ -637,6 +637,34 @@ def view_iast_header_injection_secure(request):
     return response
 
 
+@csrf_exempt
+def view_iast_code_injection_insecure(request):
+    code_string = request.POST.get("code")
+    _ = eval(code_string)
+    return HttpResponse("OK", status=200)
+
+
+@csrf_exempt
+def view_iast_code_injection_secure(request):
+    import operator
+
+    def safe_eval(expr):
+        ops = {
+            "+": operator.add,
+            "-": operator.sub,
+            "*": operator.mul,
+            "/": operator.truediv,
+        }
+        if len(expr) != 3 or expr[1] not in ops:
+            raise ValueError("Invalid expression")
+        a, op, b = expr
+        return ops[op](float(a), float(b))
+
+    code_string = request.POST.get("code")
+    _ = safe_eval(code_string)
+    return HttpResponse("OK", status=200)
+
+
 def make_distant_call(request):
     # curl localhost:7777/make_distant_call?url=http%3A%2F%2Fweblog%3A7777 | jq
 
@@ -928,6 +956,8 @@ def s3_multipart_upload(request):
     path("iast/source/path/test", view_iast_source_path),
     path("iast/source/path_parameter/test/<str:table>", view_iast_source_path_parameter),
     path("iast/header_injection/test_secure", view_iast_header_injection_secure),
+    path("iast/code_injection/test_insecure", view_iast_code_injection_insecure),
+    path("iast/code_injection/test_secure", view_iast_code_injection_secure),
     path("iast/header_injection/test_insecure", view_iast_header_injection_insecure),
     path("make_distant_call", make_distant_call),
     path("user_login_success_event", track_user_login_success_event),

utils/build/docker/python/fastapi/main.py

@@ -841,6 +841,32 @@ async def view_cmdi_secure(cmd: typing.Annotated[str, Form()]):
 #     return "YEAH"
 
 
+@app.post("/iast/code_injection/test_insecure", response_class=PlainTextResponse)
+async def view_iast_code_injection_insecure(code: typing.Annotated[str, Form()]):
+    _ = eval(code)
+    return "OK"
+
+
+@app.post("/iast/code_injection/test_secure", response_class=PlainTextResponse)
+async def view_iast_code_injection_secure(code: typing.Annotated[str, Form()]):
+    import operator
+
+    def safe_eval(expr):
+        ops = {
+            "+": operator.add,
+            "-": operator.sub,
+            "*": operator.mul,
+            "/": operator.truediv,
+        }
+        if len(expr) != 3 or expr[1] not in ops:
+            raise ValueError("Invalid expression")
+        a, op, b = expr
+        return ops[op](float(a), float(b))
+
+    _ = safe_eval(code)
+    return "OK"
+
+
 @app.get("/createextraservice", response_class=PlainTextResponse)
 def create_extra_service(serviceName: str = ""):
     if serviceName:

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

@@ -1131,6 +1131,36 @@ def view_iast_header_injection_secure():
     return resp
 
 
+@app.route("/iast/code_injection/test_insecure", methods=["POST"])
+def view_iast_code_injection_insecure():
+    code_string = flask_request.form["code"]
+    _ = eval(code_string)
+    resp = Response("OK")
+    return resp
+
+
+@app.route("/iast/code_injection/test_secure", methods=["POST"])
+def view_iast_code_injection_secure():
+    import operator
+
+    def safe_eval(expr):
+        ops = {
+            "+": operator.add,
+            "-": operator.sub,
+            "*": operator.mul,
+            "/": operator.truediv,
+        }
+        if len(expr) != 3 or expr[1] not in ops:
+            raise ValueError("Invalid expression")
+        a, op, b = expr
+        return ops[op](float(a), float(b))
+
+    code_string = flask_request.form["code"]
+    _ = safe_eval(code_string)
+    resp = Response("OK")
+    return resp
+
+
 _TRACK_METADATA = {
     "metadata0": "value0",
     "metadata1": "value1",