fix: only gate-created approvals unlock code execution; Code Studio re-sends approval_id

A low-risk execute_code request filed via POST /api/approvals is
auto-approved by the default policy and could be used as approval_id.
Require risk_level=critical (what the gate records). Code Studio now
re-submits the pending approval_id for the same code/language so Run
works after approval.

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

src/praisonaiui/features/code_execution.py

@@ -141,6 +141,81 @@ 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("risk_level") == "critical"
+            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 +248,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

src/praisonaiui/templates/frontend/plugins/views/code-studio.js

@@ -23,6 +23,7 @@ let editor = null;
 let monacoNs = null;
 let running = false;
 let timerId = null;
+let pendingApproval = null; // { id, language, code } awaiting approval
 const history = [];
 
 function escapeHtml(s) {
@@ -178,14 +179,19 @@ async function runCode(container) {
     runBtn.textContent = `Executing… ${((Date.now() - start) / 1000).toFixed(1)}s`;
   }, 100);
 
+  const p = pendingApproval;
+  const approval_id = p && p.language === lang && p.code === code ? p.id : undefined;
   let result;
   try {
     const res = await fetch('/api/code/execute', {
       method: 'POST',
       headers: { 'Content-Type': 'application/json' },
-      body: JSON.stringify({ language: lang, code, timeout: 30 }),
+      body: JSON.stringify({ language: lang, code, timeout: 30, approval_id }),
     });
     result = await res.json();
+    pendingApproval = result.status === 'pending_approval'
+      ? { id: result.approval_id, language: lang, code }
+      : null;
   } catch (e) {
     result = { status: 'failed', error: String(e), language: lang };
   } finally {

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,87 @@ 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_rejects_non_gate_approval(self, _fresh_approvals):
+        # A low-risk request filed via /api/approvals is auto-approved by the
+        # default policy; it must not unlock execution.
+        client = TestClient(create_app())
+        r = client.post(
+            "/api/approvals",
+            json={
+                "tool_name": "execute_code",
+                "arguments": {"code": "1", "language": "python"},
+                "risk_level": "low",
+            },
+        )
+        assert r.json()["status"] == "approved"
+        r = client.post(
+            "/api/code/execute",
+            json={"code": "1", "language": "python", "approval_id": r.json()["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):