code review jr: store path traversal...

emit workflow_run_failed

again?

pyproject.toml

@@ -28,6 +28,7 @@ dev = [
 
 [tool.pytest.ini_options]
 addopts = "-p no:cacheprovider"
+pythonpath = ["."]
 
 [tool.uv]
 package = true

src/wf_artifacts/store.py

@@ -1,10 +1,23 @@
 from __future__ import annotations
 
 import json
+import re
 from pathlib import Path
 
 from .models import WorkflowArtifact, WorkflowDeployment
 
+STORE_ID_PATTERN = r"^[A-Za-z0-9_][A-Za-z0-9_.-]*$"
+
+
+def ensure_store_id(value: str, *, field_name: str) -> str:
+    """Reject ids that cannot safely map to one local store path component."""
+    if not re.fullmatch(STORE_ID_PATTERN, value):
+        raise ValueError(
+            f"{field_name} must start with alphanumeric or underscore and contain "
+            "only [A-Za-z0-9_.-]"
+        )
+    return value
+
 
 class WorkflowArtifactStore:
     """Storage boundary for workflow artifacts and deployments."""
@@ -51,16 +64,16 @@ class FileWorkflowArtifactStore(WorkflowArtifactStore):
         return self.root / "deployments"
 
     def save_artifact(self, artifact: WorkflowArtifact) -> None:
-        artifact_dir = self.artifacts_dir / artifact.id
+        artifact_dir = self._artifact_dir(artifact.id)
         artifact_dir.mkdir(parents=True, exist_ok=True)
-        path = artifact_dir / f"{artifact.version}.json"
+        path = artifact_dir / self._artifact_filename(artifact.version)
         path.write_text(
             json.dumps(artifact.model_dump(mode="json"), indent=2),
             encoding="utf-8",
         )
 
     def get_artifact(self, artifact_id: str, version: int) -> WorkflowArtifact:
-        path = self.artifacts_dir / artifact_id / f"{version}.json"
+        path = self._artifact_dir(artifact_id) / self._artifact_filename(version)
         if not path.exists():
             raise KeyError(f"unknown workflow artifact {artifact_id}@{version}")
         return WorkflowArtifact.model_validate_json(path.read_text(encoding="utf-8"))
@@ -74,9 +87,10 @@ class FileWorkflowArtifactStore(WorkflowArtifactStore):
         return artifacts
 
     def resolve_latest(self, artifact_id: str) -> WorkflowArtifact:
+        artifact_dir = self._artifact_dir(artifact_id)
         versions = [
             int(path.stem)
-            for path in (self.artifacts_dir / artifact_id).glob("*.json")
+            for path in artifact_dir.glob("*.json")
             if path.stem.isdecimal()
         ]
         if not versions:
@@ -84,14 +98,14 @@ class FileWorkflowArtifactStore(WorkflowArtifactStore):
         return self.get_artifact(artifact_id, max(versions))
 
     def save_deployment(self, deployment: WorkflowDeployment) -> None:
-        path = self.deployments_dir / f"{deployment.id}.json"
+        path = self._deployment_path(deployment.id)
         path.write_text(
             json.dumps(deployment.model_dump(mode="json"), indent=2),
             encoding="utf-8",
         )
 
     def get_deployment(self, deployment_id: str) -> WorkflowDeployment:
-        path = self.deployments_dir / f"{deployment_id}.json"
+        path = self._deployment_path(deployment_id)
         if not path.exists():
             raise KeyError(f"unknown workflow deployment {deployment_id!r}")
         return WorkflowDeployment.model_validate_json(path.read_text(encoding="utf-8"))
@@ -106,7 +120,31 @@ class FileWorkflowArtifactStore(WorkflowArtifactStore):
 
     def delete_deployment(self, deployment_id: str) -> None:
         """Remove one mutable deployment binding record from the store."""
-        path = self.deployments_dir / f"{deployment_id}.json"
+        path = self._deployment_path(deployment_id)
         if not path.exists():
             raise KeyError(f"unknown workflow deployment {deployment_id!r}")
         path.unlink()
+
+    def _artifact_dir(self, artifact_id: str) -> Path:
+        safe_id = ensure_store_id(artifact_id, field_name="artifact_id")
+        root = self.artifacts_dir.resolve()
+        path = (self.artifacts_dir / safe_id).resolve()
+        if path.parent != root:
+            raise ValueError(f"artifact_id escapes artifact store: {artifact_id!r}")
+        return path
+
+    @staticmethod
+    def _artifact_filename(version: int) -> str:
+        if version < 1:
+            raise ValueError("artifact version must be >= 1")
+        return f"{version}.json"
+
+    def _deployment_path(self, deployment_id: str) -> Path:
+        safe_id = ensure_store_id(deployment_id, field_name="deployment_id")
+        root = self.deployments_dir.resolve()
+        path = (self.deployments_dir / f"{safe_id}.json").resolve()
+        if path.parent != root:
+            raise ValueError(
+                f"deployment_id escapes deployment store: {deployment_id!r}"
+            )
+        return path

src/wf_mcp/broker/service/workflow_runtime.py

@@ -9,6 +9,7 @@ from wf_authoring import NodeSpec
 from wf_core import (
     NodeUse,
     RunState,
+    RunStatus,
     Workflow,
     execute_workflow_result_async,
     resume_workflow_result_async,
@@ -168,7 +169,9 @@ class WorkflowRuntimeService:
         )
         self.emit_event(
             make_event(
-                "workflow_run_completed",
+                "workflow_run_failed"
+                if run.status == RunStatus.FAILED
+                else "workflow_run_completed",
                 workflow_name=plan.name,
                 payload={"status": run.status.value},
             )

src/wf_mcp/connections.py

@@ -1,11 +1,22 @@
 from __future__ import annotations
 
+import re
 from dataclasses import dataclass, field
 
 from .models import ConnectionConfig
 
+CONNECTION_ID_PATTERN = r"^[A-Za-z0-9_][A-Za-z0-9_.-]*$"
+
 
 def parse_connection_id(connection_id: str) -> tuple[str, str]:
+    # Connection ids are logical source ids, but they also key persisted auth and
+    # catalog files. Keep this parser conservative so unsafe ids are rejected
+    # before they reach either registry or store boundaries.
+    if not re.fullmatch(CONNECTION_ID_PATTERN, connection_id):
+        raise ValueError(
+            "connection id must start with alphanumeric or underscore and contain "
+            "only [A-Za-z0-9_.-]"
+        )
     if "." not in connection_id:
         raise ValueError("connection id must look like '<server>.<account>'")
     server, account = connection_id.split(".", 1)

src/wf_mcp/storage/store.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import json
 from pathlib import Path
 
+from ..connections import parse_connection_id
 from ..models import (
     AuthRecord,
     CatalogNodeEntry,
@@ -43,10 +44,22 @@ class FileStore(Store):
         return self.root / "catalog"
 
     def _auth_path(self, connection_id: str) -> Path:
-        return self.auth_dir / f"{connection_id}.json"
+        return self._connection_path(self.auth_dir, connection_id)
 
     def _catalog_path(self, connection_id: str) -> Path:
-        return self.catalog_dir / f"{connection_id}.json"
+        return self._connection_path(self.catalog_dir, connection_id)
+
+    @staticmethod
+    def _connection_path(directory: Path, connection_id: str) -> Path:
+        """Map one validated connection id to one file inside a store directory."""
+        parse_connection_id(connection_id)
+        root = directory.resolve()
+        path = (directory / f"{connection_id}.json").resolve()
+        if path.parent != root:
+            raise ValueError(
+                f"connection id escapes store directory: {connection_id!r}"
+            )
+        return path
 
     def save_auth(self, record: AuthRecord) -> None:
         self._auth_path(record.connection_id).write_text(

tests/artifacts/test_store.py

@@ -2,6 +2,8 @@ from __future__ import annotations
 
 import json
 
+import pytest
+
 from wf_artifacts import (
     FileWorkflowArtifactStore,
     WorkflowArtifact,
@@ -179,3 +181,41 @@ def test_file_store_deletes_deployment(tmp_path) -> None:
     store.delete_deployment("summarize_docs.personal")
 
     assert store.list_deployments() == []
+
+
+def test_file_store_rejects_artifact_id_path_traversal(tmp_path) -> None:
+    store = FileWorkflowArtifactStore(tmp_path / "store")
+    bad_artifact = artifact(1).model_copy(update={"id": "../outside"})
+
+    with pytest.raises(ValueError, match="artifact_id"):
+        store.save_artifact(bad_artifact)
+
+    assert not (tmp_path / "outside").exists()
+
+
+def test_file_store_rejects_artifact_lookup_path_traversal(tmp_path) -> None:
+    store = FileWorkflowArtifactStore(tmp_path / "store")
+
+    with pytest.raises(ValueError, match="artifact_id"):
+        store.get_artifact("../outside", 1)
+
+
+def test_file_store_rejects_deployment_id_path_traversal(tmp_path) -> None:
+    store = FileWorkflowArtifactStore(tmp_path / "store")
+    bad_deployment = WorkflowDeployment(
+        id="../outside",
+        artifact_id="summarize_docs",
+        artifact_version=1,
+    )
+
+    with pytest.raises(ValueError, match="deployment_id"):
+        store.save_deployment(bad_deployment)
+
+    assert not (tmp_path / "outside.json").exists()
+
+
+def test_file_store_rejects_deployment_lookup_path_traversal(tmp_path) -> None:
+    store = FileWorkflowArtifactStore(tmp_path / "store")
+
+    with pytest.raises(ValueError, match="deployment_id"):
+        store.get_deployment("../outside")

tests/wf_mcp/service/test_workflow_runtime.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import asyncio
 
-from wf_core import NodeUse
+from wf_core import END, NodeUse, RunStatus
 from wf_mcp.broker import WfMcpService
 from wf_mcp.broker.service.source_catalog import SourceCatalogService
 from wf_mcp.broker.service.specs import qualify_spec
@@ -12,7 +12,7 @@ from wf_mcp.storage import FileStore
 from wf_platform import CapabilityBuckets, CapabilitySource, SourceVisibility
 
 from ..test_support import echo_tool, local_temp_root
-from .conftest import single_echo_plan
+from .conftest import raw_plan, single_echo_plan
 
 
 def _unused_tool_executor(connection: ConnectionConfig):
@@ -133,3 +133,44 @@ def test_workflow_runtime_service_runs_plan_and_emits_events() -> None:
         "workflow_run_completed",
     ]
     assert events[1].payload["status"] == "completed"
+
+
+def test_workflow_runtime_service_emits_failed_event_for_failed_run() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "runtime_failed_event"))
+
+    run = asyncio.run(
+        service.workflow_runtime.run_workflow_from_plan(
+            raw_plan(
+                name="runtime_failed_event",
+                input_schema={"type": "object", "properties": {}},
+                state_schema={"type": "object", "properties": {}},
+                output_schema={"type": "object", "properties": {}},
+                start="fail",
+                nodes=[
+                    {
+                        "id": "fail",
+                        "type": "node",
+                        "node": "wf.std.runtime_error",
+                        "input": [
+                            {
+                                "value": "boom",
+                                "target": {
+                                    "root": "local",
+                                    "parts": ["message"],
+                                },
+                            }
+                        ],
+                    }
+                ],
+                edges=[{"from": "fail", "outcome": "ok", "to": END}],
+            ),
+            {},
+        )
+    )
+
+    assert run.status == RunStatus.FAILED
+    assert [event.kind for event in service.list_events()][-2:] == [
+        "workflow_run_started",
+        "workflow_run_failed",
+    ]
+    assert service.list_events()[-1].payload["status"] == "failed"

tests/wf_mcp/test_store.py

@@ -1,6 +1,9 @@
 from __future__ import annotations
 
-from wf_mcp.models import AuthRecord
+import pytest
+
+from wf_mcp.connections import parse_connection_id
+from wf_mcp.models import AuthRecord, CatalogSnapshot
 from wf_mcp.storage import FileStore
 
 from .test_support import local_temp_root
@@ -18,3 +21,40 @@ def test_file_store_round_trips_auth() -> None:
     loaded = store.load_auth("demo.personal")
 
     assert loaded == record
+
+
+def test_parse_connection_id_rejects_path_traversal() -> None:
+    for connection_id in ("../personal", "demo/../../personal", ".hidden.personal"):
+        with pytest.raises(ValueError, match="connection id"):
+            parse_connection_id(connection_id)
+
+
+def test_file_store_rejects_auth_connection_id_path_traversal(tmp_path) -> None:
+    store = FileStore(tmp_path / "store")
+    record = AuthRecord(
+        connection_id="../outside",
+        scheme="oauth",
+        payload={"token": "secret"},
+    )
+
+    with pytest.raises(ValueError, match="connection id"):
+        store.save_auth(record)
+
+    assert not (tmp_path / "outside.json").exists()
+
+
+def test_file_store_rejects_catalog_connection_id_path_traversal(tmp_path) -> None:
+    store = FileStore(tmp_path / "store")
+
+    with pytest.raises(ValueError, match="connection id"):
+        store.load_catalog("../outside")
+
+    snapshot = CatalogSnapshot(
+        connection_id="../outside",
+        fetched_at_epoch_ms=0,
+        max_age_seconds=60,
+    )
+    with pytest.raises(ValueError, match="connection id"):
+        store.save_catalog(snapshot)
+
+    assert not (tmp_path / "outside.json").exists()