diff --git a/flocks/server/auth.py b/flocks/server/auth.py index 654f7d381..da231d312 100644 --- a/flocks/server/auth.py +++ b/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/[^/]+/[^/]+/?$"), diff --git a/flocks/server/routes/workflow.py b/flocks/server/routes/workflow.py index c480e0b1f..1bf338498 100644 --- a/flocks/server/routes/workflow.py +++ b/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") diff --git a/flocks/workflow/engine.py b/flocks/workflow/engine.py index e0621865a..a0dcffe27 100644 --- a/flocks/workflow/engine.py +++ b/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, diff --git a/flocks/workflow/repl_runtime.py b/flocks/workflow/repl_runtime.py index ad78f0b59..4b9618e96 100644 --- a/flocks/workflow/repl_runtime.py +++ b/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 diff --git a/tests/server/routes/test_workflow_trigger_routes.py b/tests/server/routes/test_workflow_trigger_routes.py index bcd8d08e2..5d3e5e9c9 100644 --- a/tests/server/routes/test_workflow_trigger_routes.py +++ b/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, diff --git a/tests/workflow/test_repl_runtime_security.py b/tests/workflow/test_repl_runtime_security.py new file mode 100644 index 000000000..1c0d455a0 --- /dev/null +++ b/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} diff --git a/tests/workflow/test_workflow_template_sandbox.py b/tests/workflow/test_workflow_template_sandbox.py new file mode 100644 index 000000000..12f141508 --- /dev/null +++ b/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, {})