Dynamic (#77)

tests/dynamic_fixtures/python_frameworks/django/benign.py

@@ -0,0 +1,22 @@
+"""Phase 12 (Track L.10) — Django CMDI benign fixture.
+
+`run_cmd(request)` reads `request.GET["cmd"]` but rejects anything
+outside an allowlist before invoking `subprocess.run` with a fixed
+argv, so the sink call is unreachable for attacker-controlled values.
+"""
+import subprocess
+from django.http import HttpResponse
+from django.urls import path
+
+_ALLOW = {"status", "uptime", "version"}
+
+
+def run_cmd(request):
+    cmd = request.GET.get("cmd", "")
+    if cmd not in _ALLOW:
+        return HttpResponse("rejected", status=400)
+    subprocess.run(["/usr/bin/echo", cmd], check=False)
+    return HttpResponse("ok")
+
+
+urlpatterns = [path("run/", run_cmd)]

tests/dynamic_fixtures/python_frameworks/django/vuln.py

@@ -0,0 +1,18 @@
+"""Phase 12 (Track L.10) — Django CMDI vuln fixture.
+
+`run_cmd(request)` reads `request.GET["cmd"]` and pipes it straight to
+`os.system`.  Adapter binding: `path("run/", run_cmd)` registration with
+`cmd` flowing through `request.GET`.
+"""
+import os
+from django.http import HttpResponse
+from django.urls import path
+
+
+def run_cmd(request):
+    cmd = request.GET.get("cmd", "")
+    os.system(cmd)
+    return HttpResponse("ok")
+
+
+urlpatterns = [path("run/", run_cmd)]

tests/dynamic_fixtures/python_frameworks/django_class_method/vuln.py

@@ -0,0 +1,9 @@
+from django.views import View
+
+import os
+
+
+class UserCommandView(View):
+    def get(self, payload):
+        os.system(payload)
+        return "ok"

tests/dynamic_fixtures/python_frameworks/fastapi/benign.py

@@ -0,0 +1,20 @@
+"""Phase 12 (Track L.10) — FastAPI CMDI benign fixture.
+
+`GET /run?cmd=<...>` rejects anything outside an allowlist before
+invoking `subprocess.run` with a fixed argv, so the sink call is
+unreachable for attacker-controlled values.
+"""
+import subprocess
+from fastapi import FastAPI
+
+app = FastAPI()
+
+_ALLOW = {"status", "uptime", "version"}
+
+
+@app.get("/run")
+def run_cmd(cmd: str = ""):
+    if cmd not in _ALLOW:
+        return {"rejected": True}
+    subprocess.run(["/usr/bin/echo", cmd], check=False)
+    return {"ok": True}

tests/dynamic_fixtures/python_frameworks/fastapi/vuln.py

@@ -0,0 +1,16 @@
+"""Phase 12 (Track L.10) — FastAPI CMDI vuln fixture.
+
+`GET /run?cmd=<...>` forwards the `cmd` query parameter straight into
+`os.system`.  Adapter binding: `@app.get("/run")` with `cmd` flowing
+through the function formal.
+"""
+import os
+from fastapi import FastAPI
+
+app = FastAPI()
+
+
+@app.get("/run")
+def run_cmd(cmd: str = ""):
+    os.system(cmd)
+    return {"ok": True}

tests/dynamic_fixtures/python_frameworks/flask/benign.py

@@ -0,0 +1,21 @@
+"""Phase 12 (Track L.10) — Flask CMDI benign fixture.
+
+The `/run` route accepts a `cmd` query parameter but rejects everything
+outside an allowlist before invoking `subprocess.run` with a fixed argv,
+so the sink call is unreachable for attacker-controlled values.
+"""
+import subprocess
+from flask import Flask, request
+
+app = Flask(__name__)
+
+_ALLOW = {"status", "uptime", "version"}
+
+
+@app.route("/run", methods=["GET"])
+def run_cmd():
+    cmd = request.args.get("cmd", "")
+    if cmd not in _ALLOW:
+        return "rejected", 400
+    subprocess.run(["/usr/bin/echo", cmd], check=False)
+    return "ok"

tests/dynamic_fixtures/python_frameworks/flask/vuln.py

@@ -0,0 +1,18 @@
+"""Phase 12 (Track L.10) — Flask CMDI vuln fixture.
+
+The `/run` route forwards a `cmd` query parameter straight into
+`os.system`, so any attacker who reaches the route can execute
+arbitrary shell.  Adapter binding: `@app.route("/run", methods=["GET"])`
+with `cmd` flowing through `request.args.get`.
+"""
+import os
+from flask import Flask, request
+
+app = Flask(__name__)
+
+
+@app.route("/run", methods=["GET"])
+def run_cmd():
+    cmd = request.args.get("cmd", "")
+    os.system(cmd)
+    return "ok"

tests/dynamic_fixtures/python_frameworks/starlette/benign.py

@@ -0,0 +1,23 @@
+"""Phase 12 (Track L.10) — Starlette CMDI benign fixture.
+
+`run_cmd(request)` reads the `cmd` query parameter but rejects anything
+outside an allowlist before invoking `subprocess.run` with a fixed
+argv, so the sink call is unreachable for attacker-controlled values.
+"""
+import subprocess
+from starlette.applications import Starlette
+from starlette.responses import PlainTextResponse
+from starlette.routing import Route
+
+_ALLOW = {"status", "uptime", "version"}
+
+
+async def run_cmd(request):
+    cmd = request.query_params.get("cmd", "")
+    if cmd not in _ALLOW:
+        return PlainTextResponse("rejected", status_code=400)
+    subprocess.run(["/usr/bin/echo", cmd], check=False)
+    return PlainTextResponse("ok")
+
+
+app = Starlette(routes=[Route("/run", endpoint=run_cmd)])

tests/dynamic_fixtures/python_frameworks/starlette/vuln.py

@@ -0,0 +1,19 @@
+"""Phase 12 (Track L.10) — Starlette CMDI vuln fixture.
+
+`run_cmd(request)` reads the `cmd` query parameter and pipes it
+straight to `os.system`.  Adapter binding: `Route("/run", endpoint=run_cmd)`
+registration with `cmd` flowing through `request.query_params`.
+"""
+import os
+from starlette.applications import Starlette
+from starlette.responses import PlainTextResponse
+from starlette.routing import Route
+
+
+async def run_cmd(request):
+    cmd = request.query_params.get("cmd", "")
+    os.system(cmd)
+    return PlainTextResponse("ok")
+
+
+app = Starlette(routes=[Route("/run", endpoint=run_cmd)])