fix: require recorded approval before /api/code/execute runs code (fixes #311)

Co-Authored-By: Claude Opus 5.5 <[email protected]>

src/praisonaiui/features/code_execution.py

@@ -141,6 +141,80 @@ def set_execution_manager(manager: CodeExecutionProtocol) -> None:
     _execution_manager = manager
 
 
+# ── Approval gate ────────────────────────────────────────────────────
+
+
+async def _approval_gate(
+    code: str, language: str, approval_id: Optional[str]
+) -> Optional[JSONResponse]:
+    """Return None if execution is approved, else the response to send.
+
+    Each run is recorded via the approvals feature as a ``critical``
+    ``execute_code`` request. A pending request returns 202 with its id;
+    re-submit with ``approval_id`` once approved (single use, bound to
+    the exact code/language).
+    """
+    from .approvals import _notify_subscribers, get_approval_manager
+
+    approvals = get_approval_manager()
+    arguments = {"code": code, "language": language}
+
+    if approval_id:
+        entry = approvals.get(approval_id)
+        if (
+            entry
+            and entry.get("tool_name") == "execute_code"
+            and entry.get("status") == "approved"
+            and entry.get("arguments") == arguments
+            and not entry.get("consumed")
+        ):
+            entry["consumed"] = True
+            return None
+        return JSONResponse(
+            {
+                "status": "error",
+                "language": language,
+                "error": "Approval not found, not approved, already used, "
+                "or does not match this code",
+            },
+            status_code=403,
+        )
+
+    entry = approvals.request_approval(
+        {
+            "tool_name": "execute_code",
+            "arguments": arguments,
+            "risk_level": "critical",
+            "description": f"Execute {language} code ({len(code)} chars)",
+        }
+    )
+    status = entry.get("status")
+    await _notify_subscribers("new" if status == "pending" else "resolved", entry)
+    if status == "approved":
+        entry["consumed"] = True
+        return None
+    if status == "denied":
+        return JSONResponse(
+            {
+                "status": "error",
+                "language": language,
+                "approval_id": entry["id"],
+                "error": entry.get("reason") or "Code execution denied",
+            },
+            status_code=403,
+        )
+    return JSONResponse(
+        {
+            "status": "pending_approval",
+            "language": language,
+            "approval_id": entry["id"],
+            "output": f"Awaiting approval {entry['id']}. Approve it, then re-run "
+            "with this approval_id.",
+        },
+        status_code=202,
+    )
+
+
 # ── Feature class ────────────────────────────────────────────────────
 
 
@@ -173,9 +247,20 @@ class CodeExecutionFeature(BaseFeatureProtocol):
     async def _execute(self, request: Request) -> JSONResponse:
         mgr = get_execution_manager()
         body = await request.json()
+        code = body.get("code", "")
+        language = body.get("language", "python")
+        allowed = {lang["id"] for lang in mgr.list_languages()}
+        if language not in allowed:
+            return JSONResponse(
+                {"status": "error", "error": f"Language '{language}' not allowed"},
+                status_code=400,
+            )
+        blocked = await _approval_gate(code, language, body.get("approval_id"))
+        if blocked is not None:
+            return blocked
         result = mgr.execute(
-            code=body.get("code", ""),
-            language=body.get("language", "python"),
+            code=code,
+            language=language,
             timeout=body.get("timeout", 30),
         )
         status = 200 if result.get("status") != "error" else 400

tests/test_all_gaps.py

@@ -141,7 +141,7 @@ check("GET nonexistent plugin", client.get("/api/marketplace/plugins/nonexistent
 print("\n═══ Code Execution API ═══")
 check("GET /api/code/languages", client.get("/api/code/languages"), 200, ["languages", "count"])
 r = client.post("/api/code/execute", json={"code": "print('hello')", "language": "python"})
-check("POST /api/code/execute", r, 200, ["status", "language"])
+check("POST /api/code/execute (pending approval)", r, 202, ["status", "approval_id"])
 r = client.post("/api/code/execute", json={"code": "ls", "language": "ruby"})
 check("POST execute disallowed lang", r, 400, ["error"])
 

tests/unit/test_code_studio.py

@@ -2,6 +2,7 @@
 
 from pathlib import Path
 
+import pytest
 from starlette.testclient import TestClient
 
 from praisonaiui.server import create_app
@@ -34,16 +35,68 @@ class TestCodeExecutionApi:
         d = r.json()
         assert "languages" in d and "count" in d
 
-    def test_execute(self):
+    @pytest.fixture(autouse=True)
+    def _fresh_approvals(self, monkeypatch):
+        from praisonaiui.features import approvals
+
+        monkeypatch.setattr(approvals, "_approval_manager", approvals.SimpleApprovalManager())
+        return approvals._approval_manager
+
+    def test_execute_requires_approval(self, _fresh_approvals):
         client = TestClient(create_app())
+        payload = {"code": "print('hi')", "language": "python", "timeout": 30}
+        r = client.post("/api/code/execute", json=payload)
+        assert r.status_code == 202
+        d = r.json()
+        assert d["status"] == "pending_approval"
+        approval_id = d["approval_id"]
+        assert _fresh_approvals.get(approval_id)["tool_name"] == "execute_code"
+
+        # Not yet approved
+        r = client.post("/api/code/execute", json={**payload, "approval_id": approval_id})
+        assert r.status_code == 403
+
+        assert client.post(f"/api/approvals/{approval_id}/approve", json={}).status_code == 200
+        # Approval is bound to the exact code
         r = client.post(
             "/api/code/execute",
-            json={"code": "print('hi')", "language": "python", "timeout": 30},
+            json={**payload, "code": "print('x')", "approval_id": approval_id},
         )
+        assert r.status_code == 403
+
+        r = client.post("/api/code/execute", json={**payload, "approval_id": approval_id})
         assert r.status_code == 200
         d = r.json()
         assert "status" in d and "language" in d
 
+        # Single use
+        r = client.post("/api/code/execute", json={**payload, "approval_id": approval_id})
+        assert r.status_code == 403
+
+    def test_execute_denied(self, _fresh_approvals):
+        client = TestClient(create_app())
+        r = client.post("/api/code/execute", json={"code": "1", "language": "python"})
+        approval_id = r.json()["approval_id"]
+        client.post(f"/api/approvals/{approval_id}/deny", json={})
+        r = client.post(
+            "/api/code/execute",
+            json={"code": "1", "language": "python", "approval_id": approval_id},
+        )
+        assert r.status_code == 403
+
+    def test_execute_auto_approved_by_policy(self, _fresh_approvals):
+        _fresh_approvals.update_policies({"auto_approve_tools": ["execute_code"]})
+        client = TestClient(create_app())
+        r = client.post("/api/code/execute", json={"code": "1", "language": "python"})
+        assert r.status_code == 200
+        assert _fresh_approvals.list_history()[0]["approver"] == "auto-policy"
+
+    def test_execute_disallowed_language(self, _fresh_approvals):
+        client = TestClient(create_app())
+        r = client.post("/api/code/execute", json={"code": "ls", "language": "ruby"})
+        assert r.status_code == 400
+        assert _fresh_approvals.list_pending() == []
+
 
 class TestCodeStudioFrontend:
     def test_view_file_exists(self):