[security] fix: gate executable web API agents

README.md

@@ -246,13 +246,17 @@ agentflow serve                     # start the local web UI and API on 127.0.0.
 
 The web API only accepts `application/json` requests for `/api/runs` and `/api/runs/validate`, and `pipeline_path` is disabled on those endpoints by default. This prevents the browser-facing control plane from executing arbitrary local `.py` pipeline files just by referencing a path.
 
-If you intentionally want the web API to load pipelines from filesystem paths in a trusted local environment, opt in explicitly:
+Inline `shell` and `python` utility agents are also disabled for web API requests by default. This prevents browser/API callers from turning a submitted pipeline into direct local command execution.
+
+If you intentionally want the web API to load pipelines from filesystem paths or accept inline executable utility agents in a trusted local environment, opt in explicitly:
 
 ```bash
-AGENTFLOW_API_ALLOW_PIPELINE_PATH=1 agentflow serve
+AGENTFLOW_API_ALLOW_PIPELINE_PATH=1 \
+AGENTFLOW_API_ALLOW_EXECUTABLE_AGENTS=1 \
+agentflow serve
 ```
 
-That opt-in is meant for trusted operator-controlled workflows only.
+Those opt-ins are meant for trusted operator-controlled workflows only.
 
 ## Acknowledgements
 

agentflow/app.py

@@ -17,7 +17,7 @@
 from agentflow.defaults import bundled_template_path
 from agentflow.loader import load_pipeline_from_data, load_pipeline_from_path, load_pipeline_from_text
 from agentflow.orchestrator import Orchestrator
-from agentflow.specs import PipelineSpec
+from agentflow.specs import AgentKind, PipelineSpec
 from agentflow.store import RunStore
 
 
@@ -28,6 +28,18 @@ def _api_pipeline_path_enabled() -> bool:
     return os.getenv("AGENTFLOW_API_ALLOW_PIPELINE_PATH", "").strip().lower() in {"1", "true", "yes", "on"}
 
 
+def _api_executable_agents_enabled() -> bool:
+    return os.getenv("AGENTFLOW_API_ALLOW_EXECUTABLE_AGENTS", "").strip().lower() in {"1", "true", "yes", "on"}
+
+
+def _reject_web_api_executable_agents(pipeline: PipelineSpec) -> None:
+    if _api_executable_agents_enabled():
+        return
+    executable_agents = {AgentKind.PYTHON, AgentKind.SHELL}
+    if any(node.agent in executable_agents for node in pipeline.nodes):
+        raise HTTPException(status_code=403, detail="executable agents are disabled for the web API by default")
+
+
 def _require_json_request(request: Request) -> None:
     content_type = request.headers.get("content-type", "")
     if "application/json" not in content_type.lower():
@@ -115,13 +127,15 @@ async def validate_run(request: Request) -> JSONResponse:
         _require_json_request(request)
         payload = await request.json()
         pipeline = _parse_pipeline_payload(payload, allow_pipeline_path=_api_pipeline_path_enabled())
+        _reject_web_api_executable_agents(pipeline)
         return JSONResponse({"ok": True, "pipeline": pipeline.model_dump(mode="json")})
 
     @app.post("/api/runs")
     async def create_run(request: Request) -> JSONResponse:
         _require_json_request(request)
         payload = await request.json()
         pipeline = _parse_pipeline_payload(payload, allow_pipeline_path=_api_pipeline_path_enabled())
+        _reject_web_api_executable_agents(pipeline)
         run = await app.state.orchestrator.submit(pipeline)
         return JSONResponse(run.model_dump(mode="json"))
 

docs/cli.md

@@ -77,13 +77,17 @@ The web API only accepts `application/json` for `/api/runs` and `/api/runs/valid
 
 For safety, the browser-facing API also disables `pipeline_path` by default, so a request cannot cause AgentFlow to execute a local `.py` pipeline file just by naming its path.
 
-If you intentionally want to allow filesystem path loading from the local web API in a trusted environment, opt in explicitly:
+Inline `shell` and `python` utility agents are also disabled for web API requests by default, so a submitted pipeline cannot directly become local command execution through the browser/API control plane.
+
+If you intentionally want to allow filesystem path loading or inline executable utility agents from the local web API in a trusted environment, opt in explicitly:
 
 ```bash
-AGENTFLOW_API_ALLOW_PIPELINE_PATH=1 agentflow serve
+AGENTFLOW_API_ALLOW_PIPELINE_PATH=1 \
+AGENTFLOW_API_ALLOW_EXECUTABLE_AGENTS=1 \
+agentflow serve
 ```
 
-Treat that override as a trusted operator-only setting.
+Treat those overrides as trusted operator-only settings.
 
 ## Tuned Agents And Evolution
 

tests/test_api.py

@@ -186,6 +186,50 @@ def test_api_validate_supports_pipeline_path_payload_when_explicitly_enabled(tmp
 
 
 
+def test_api_rejects_inline_executable_agents_by_default(tmp_path):
+    orchestrator = make_orchestrator(tmp_path)
+    app = create_app(store=orchestrator.store, orchestrator=orchestrator)
+    client = TestClient(app)
+
+    payload = {
+        "pipeline": {
+            "name": "shell-rce",
+            "working_dir": str(tmp_path),
+            "nodes": [{"id": "shell", "agent": "shell", "prompt": "echo unsafe"}],
+        }
+    }
+
+    validate = client.post("/api/runs/validate", json=payload)
+    assert validate.status_code == 403
+    assert validate.json()["detail"] == "executable agents are disabled for the web API by default"
+
+    create = client.post("/api/runs", json=payload)
+    assert create.status_code == 403
+    assert create.json()["detail"] == "executable agents are disabled for the web API by default"
+
+
+async def test_api_allows_inline_executable_agents_when_explicitly_enabled(tmp_path, monkeypatch):
+    monkeypatch.setenv("AGENTFLOW_API_ALLOW_EXECUTABLE_AGENTS", "1")
+    orchestrator = make_orchestrator(tmp_path)
+    app = create_app(store=orchestrator.store, orchestrator=orchestrator)
+    client = TestClient(app)
+
+    response = client.post(
+        "/api/runs/validate",
+        json={
+            "pipeline": {
+                "name": "python-opt-in",
+                "working_dir": str(tmp_path),
+                "nodes": [{"id": "py", "agent": "python", "prompt": "print('trusted')"}],
+            }
+        },
+    )
+
+    assert response.status_code == 200
+    assert response.json()["pipeline"]["nodes"][0]["agent"] == "python"
+
+
+
 def test_api_rejects_non_json_content_type(tmp_path):
     orchestrator = make_orchestrator(tmp_path)
     app = create_app(store=orchestrator.store, orchestrator=orchestrator)