fix: harden workflow execution against code injection (RCE)

flocks/server/auth.py

@@ -65,6 +65,11 @@
 # downstream handler is fully responsible for its own authentication
 # (signature checks, IP allowlists, replay protection, …).  Do NOT add
 # entries that touch user data without a per-request integrity check.
+#
+# Workflow webhook paths also need to be reachable by external systems that
+# cannot present a browser session.  They are safe to exempt here only because
+# _authorize_webhook_trigger() fails closed unless the trigger config supplies
+# api_key or hmac authentication.  See: https://github.com/AgentFlocks/flocks/issues/454
 PUBLIC_PATH_REGEXES = (
     re.compile(r"^/(?:api/)?channel/[^/]+/webhook/?$"),
     re.compile(r"^/webhook/workflows/[^/]+/[^/]+/?$"),

flocks/server/routes/workflow.py

@@ -99,6 +99,8 @@
 _PROGRESS_FLUSH_EVERY_STEPS = 5
 
 _LEGACY_SINGLETON_TRIGGER_TYPES = frozenset({"schedule", "kafka", "syslog"})
+_WEBHOOK_TRIGGER_TYPES = frozenset({"webhook", "custom_webhook"})
+_WEBHOOK_AUTH_TYPES = frozenset({"api_key", "hmac"})
 _WORKFLOW_INTEGRATION_CONFIG_VERSION = 1
 _WORKFLOW_INTEGRATION_CONFIG_KIND = "workflow.integration-config"
 _WORKFLOW_INTEGRATION_CONFIG_PREFIX = "workflow_integration_config/"
@@ -2733,6 +2735,39 @@ def _validate_trigger_type_constraints(triggers: List[TriggerDefinition]) -> Non
     raise HTTPException(status_code=409, detail=detail)
 
 
+def _webhook_auth_type(trigger: TriggerDefinition) -> str:
+    auth = trigger.auth
+    return str(getattr(auth, "type", "") or "").strip().lower()
+
+
+def _validate_webhook_trigger_auth(trigger: TriggerDefinition) -> None:
+    if trigger.type not in _WEBHOOK_TRIGGER_TYPES:
+        return
+
+    auth = trigger.auth
+    auth_type = _webhook_auth_type(trigger)
+    if auth is None or auth_type in {"", "none"}:
+        raise HTTPException(
+            status_code=400,
+            detail="Webhook triggers must configure authentication with auth.type 'api_key' or 'hmac'.",
+        )
+    if auth_type not in _WEBHOOK_AUTH_TYPES:
+        raise HTTPException(status_code=400, detail=f"Unsupported webhook auth type: {auth.type}")
+    if auth_type == "api_key" and not (auth.apiKey or auth.secretRef):
+        raise HTTPException(
+            status_code=400,
+            detail="Webhook api_key auth requires either apiKey or secretRef.",
+        )
+    if auth_type == "hmac" and not auth.secretRef:
+        raise HTTPException(status_code=400, detail="Webhook hmac auth requires secretRef.")
+
+
+def _validate_trigger_definitions(triggers: List[TriggerDefinition]) -> None:
+    _validate_trigger_type_constraints(triggers)
+    for trigger in triggers:
+        _validate_webhook_trigger_auth(trigger)
+
+
 @router.get("/workflow/{workflow_id}/triggers")
 async def list_workflow_triggers(workflow_id: str):
     """List unified triggers for a workflow with runtime status."""
@@ -2764,7 +2799,7 @@ async def create_workflow_trigger(workflow_id: str, trigger: TriggerDefinition):
         raise HTTPException(status_code=404, detail=f"Workflow not found: {workflow_id}")
     existing = await _get_workflow_trigger_defs(workflow_id, data)
     updated = _replace_or_append_trigger(existing, trigger)
-    _validate_trigger_type_constraints(updated)
+    _validate_trigger_definitions(updated)
     persisted = await _persist_workflow_triggers(workflow_id, data, updated)
     await default_trigger_runtime.restart_workflow(workflow_id, persisted.get("workflowJson") or {})
     status = await default_trigger_runtime.get_trigger_status(workflow_id, trigger)
@@ -2781,7 +2816,7 @@ async def update_workflow_trigger(workflow_id: str, trigger_id: str, trigger: Tr
     _find_trigger_or_404(existing, trigger_id)
     updated_trigger = trigger.model_copy(update={"id": trigger_id})
     updated = _replace_or_append_trigger(existing, updated_trigger)
-    _validate_trigger_type_constraints(updated)
+    _validate_trigger_definitions(updated)
     persisted = await _persist_workflow_triggers(workflow_id, data, updated)
     await default_trigger_runtime.restart_workflow(workflow_id, persisted.get("workflowJson") or {})
     status = await default_trigger_runtime.get_trigger_status(workflow_id, updated_trigger)
@@ -2908,9 +2943,13 @@ def _authorize_webhook_trigger(
     raw_body: bytes,
 ) -> None:
     auth = trigger.auth
-    if auth is None or auth.type in {"none", ""}:
-        return
-    if auth.type == "api_key":
+    auth_type = _webhook_auth_type(trigger)
+    if auth is None or auth_type in {"none", ""}:
+        raise HTTPException(
+            status_code=401,
+            detail="Webhook trigger must configure authentication with auth.type 'api_key' or 'hmac'.",
+        )
+    if auth_type == "api_key":
         expected = auth.apiKey or _resolve_trigger_secret(auth.secretRef)
         if not expected:
             raise HTTPException(status_code=401, detail="Webhook trigger API key is not configured")
@@ -2919,7 +2958,7 @@ def _authorize_webhook_trigger(
         if actual != expected:
             raise HTTPException(status_code=401, detail="Invalid webhook API key")
         return
-    if auth.type == "hmac":
+    if auth_type == "hmac":
         expected = _resolve_trigger_secret(auth.secretRef)
         if not expected:
             raise HTTPException(status_code=401, detail="Webhook trigger secret is not configured")

flocks/workflow/engine.py

@@ -841,9 +841,9 @@ def _execute_llm_node(
         """Execute an LLM node: render Jinja2 prompt template, call LLM."""
         assert node.prompt, "llm node requires prompt"
         try:
-            from jinja2 import Template, TemplateError
+            from jinja2.sandbox import SandboxedEnvironment
 
-            rendered = Template(node.prompt).render(**inputs)
+            rendered = SandboxedEnvironment().from_string(node.prompt).render(**inputs)
         except Exception as e:
             raise NodeExecutionError(
                 node_id=node.id,
@@ -871,14 +871,15 @@ def _execute_http_request_node(self, node: Node, inputs: Dict[str, Any]) -> Tupl
         assert node.url, "http_request node requires url"
         assert node.method, "http_request node requires method"
         try:
-            from jinja2 import Template
+            from jinja2.sandbox import SandboxedEnvironment
 
-            url = Template(node.url).render(**inputs)
+            _sandbox = SandboxedEnvironment()
+            url = _sandbox.from_string(node.url).render(**inputs)
             method = node.method.upper()
             headers = node.headers or {}
             body = node.body
             if isinstance(body, str):
-                body = Template(body).render(**inputs)
+                body = _sandbox.from_string(body).render(**inputs)
         except Exception as e:
             raise NodeExecutionError(
                 node_id=node.id,

flocks/workflow/repl_runtime.py

@@ -46,6 +46,12 @@ def _drain_text_stream(stream: TextIO, chunks: list[str]) -> None:
 
 @dataclass
 class PythonExecRuntime(Runtime):
+    """Trusted host-process runtime.
+
+    This class intentionally is not a security sandbox. Untrusted workflow
+    execution must use SandboxPythonExecRuntime plus authenticated entrypoints.
+    """
+
     globals: Dict[str, Any] = field(default_factory=dict)
     tool_registry: Optional[Any] = None  # FlocksToolAdapter or compatible
     cancel_checker: Optional[Callable[[], bool]] = None

tests/server/routes/test_workflow_trigger_routes.py

@@ -690,6 +690,7 @@ async def _fake_status(workflow_id: str, trigger: Any) -> dict[str, Any]:
             "type": "custom_webhook",
             "enabled": True,
             "source": {"path": "/alerts/demo", "method": "POST"},
+            "auth": {"type": "api_key", "apiKey": "demo-secret"},
             "mapping": {"payload": "$.body"},
         },
     )
@@ -874,6 +875,86 @@ async def _fake_dispatch_event(**kwargs: Any) -> dict[str, Any]:
     assert "result" not in body
 
 
+@pytest.mark.asyncio
+async def test_webhook_route_rejects_trigger_without_auth(
+    client: AsyncClient,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(
+        workflow_routes,
+        "_read_workflow_from_fs",
+        lambda workflow_id: {
+            "id": workflow_id,
+            "workflowJson": {
+                "start": "n1",
+                "nodes": [{"id": "n1", "type": "python", "code": "outputs['ok'] = True"}],
+                "edges": [],
+                "triggers": [
+                    {
+                        "id": "hook-default",
+                        "type": "custom_webhook",
+                        "enabled": True,
+                        "auth": {"type": "none"},
+                    }
+                ],
+            },
+        },
+    )
+    dispatched = False
+
+    async def _fake_dispatch_event(**_kwargs: Any) -> dict[str, Any]:
+        nonlocal dispatched
+        dispatched = True
+        return {"matched": True, "executed": True}
+
+    monkeypatch.setattr(
+        workflow_routes,
+        "default_trigger_runtime",
+        SimpleNamespace(dispatch_event=_fake_dispatch_event),
+    )
+
+    response = await client.post(
+        "/webhook/workflows/wf-1/hook-default",
+        json={"severity": "high"},
+    )
+
+    assert response.status_code == 401, response.text
+    assert dispatched is False
+
+
+@pytest.mark.asyncio
+async def test_create_webhook_trigger_requires_real_auth(
+    client: AsyncClient,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(
+        workflow_routes,
+        "_read_workflow_from_fs",
+        lambda workflow_id: {
+            "id": workflow_id,
+            "workflowJson": {
+                "start": "n1",
+                "nodes": [{"id": "n1", "type": "python", "code": "outputs['ok'] = True"}],
+                "edges": [],
+                "triggers": [],
+            },
+        },
+    )
+
+    response = await client.post(
+        "/api/workflow/wf-1/triggers",
+        json={
+            "id": "hook-default",
+            "type": "custom_webhook",
+            "enabled": True,
+            "auth": {"type": "none"},
+        },
+    )
+
+    assert response.status_code == 400, response.text
+    assert "authentication" in response.text
+
+
 @pytest.mark.asyncio
 async def test_webhook_route_rejects_disabled_trigger(
     client: AsyncClient,

tests/workflow/test_repl_runtime_security.py

@@ -0,0 +1,25 @@
+from __future__ import annotations
+
+import pytest
+
+from flocks.workflow.repl_runtime import PythonExecRuntime
+
+
+def test_python_exec_runtime_allows_installed_requirement_imports() -> None:
+    pytest.importorskip("pydantic")
+    pytest.importorskip("yaml")
+    pytest.importorskip("httpx")
+
+    outputs, _stdout = PythonExecRuntime().execute(
+        "\n".join(
+            [
+                "import httpx",
+                "import pydantic",
+                "import yaml",
+                "outputs['ok'] = bool(httpx and pydantic and yaml)",
+            ]
+        ),
+        {},
+    )
+
+    assert outputs == {"ok": True}

tests/workflow/test_workflow_template_sandbox.py

@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+import pytest
+
+from flocks.workflow.engine import WorkflowEngine
+from flocks.workflow.errors import NodeExecutionError
+from flocks.workflow.models import Node, Workflow
+
+
+def _engine() -> WorkflowEngine:
+    workflow = Workflow.from_dict(
+        {
+            "start": "start",
+            "nodes": [{"id": "start", "type": "python", "code": "outputs['ok'] = True"}],
+            "edges": [],
+        }
+    )
+    return WorkflowEngine(workflow)
+
+
+def test_llm_node_prompt_uses_jinja_sandbox() -> None:
+    node = Node.model_validate(
+        {
+            "id": "llm",
+            "type": "llm",
+            "prompt": "{{ ''.__class__.__mro__ }}",
+        }
+    )
+
+    with pytest.raises(NodeExecutionError, match="Prompt template render failed"):
+        _engine()._execute_llm_node(node, {})
+
+
+def test_http_request_node_url_uses_jinja_sandbox() -> None:
+    node = Node.model_validate(
+        {
+            "id": "http",
+            "type": "http_request",
+            "method": "GET",
+            "url": "{{ ''.__class__.__mro__ }}",
+        }
+    )
+
+    with pytest.raises(NodeExecutionError, match="HTTP request template render failed"):
+        _engine()._execute_http_request_node(node, {})