[pitboss] phase 12: Track L.10 — Flask / Django / FastAPI / Starlette adapters

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)])