[security] fix: contain run artifact paths

agentflow/app.py

@@ -163,6 +163,8 @@ async def get_events(run_id: str) -> JSONResponse:
     async def get_artifact(run_id: str, node_id: str, name: str) -> PlainTextResponse:
         try:
             content = app.state.store.read_artifact_text(run_id, node_id, name)
+        except ValueError as exc:
+            raise HTTPException(status_code=400, detail="invalid artifact path") from exc
         except FileNotFoundError as exc:
             raise HTTPException(status_code=404, detail="artifact not found") from exc
         return PlainTextResponse(content)

agentflow/store.py

@@ -4,7 +4,7 @@
 import queue
 import threading
 from collections import defaultdict
-from pathlib import Path
+from pathlib import Path, PurePath
 from uuid import uuid4
 
 from pydantic import ValidationError
@@ -13,6 +13,16 @@
 from agentflow.utils import ensure_dir
 
 
+def _safe_path_segment(value: str, label: str) -> str:
+    if not isinstance(value, str) or not value:
+        raise ValueError(f"invalid {label} path segment")
+    if value in {".", ".."} or "/" in value or "\\" in value:
+        raise ValueError(f"invalid {label} path segment")
+    if PurePath(value).name != value:
+        raise ValueError(f"invalid {label} path segment")
+    return value
+
+
 class RunStore:
     def __init__(self, base_dir: str | Path = ".agentflow/runs") -> None:
         self.base_dir = ensure_dir(Path(base_dir).expanduser())
@@ -50,13 +60,13 @@ def new_run_id(self) -> str:
         return uuid4().hex
 
     def run_dir(self, run_id: str) -> Path:
-        return ensure_dir(self.base_dir / run_id)
+        return ensure_dir(self.base_dir / _safe_path_segment(run_id, "run_id"))
 
     def node_artifact_dir(self, run_id: str, node_id: str) -> Path:
-        return ensure_dir(self.run_dir(run_id) / "artifacts" / node_id)
+        return ensure_dir(self.run_dir(run_id) / "artifacts" / _safe_path_segment(node_id, "node_id"))
 
     def artifact_path(self, run_id: str, node_id: str, name: str) -> Path:
-        return self.node_artifact_dir(run_id, node_id) / name
+        return self.node_artifact_dir(run_id, node_id) / _safe_path_segment(name, "artifact name")
 
     def cancel_request_path(self, run_id: str) -> Path:
         return self.run_dir(run_id) / "cancel.requested"

tests/test_api.py

@@ -4,10 +4,12 @@
 import json
 import os
 
+import pytest
 from fastapi.testclient import TestClient
 
 from agentflow.app import create_app
 from agentflow.orchestrator import Orchestrator
+from agentflow.specs import RunRecord
 from agentflow.store import RunStore
 from tests.test_orchestrator import make_orchestrator
 
@@ -186,6 +188,51 @@ def test_api_validate_supports_pipeline_path_payload_when_explicitly_enabled(tmp
 
 
 
+def test_api_rejects_artifact_path_traversal(tmp_path):
+    orchestrator = make_orchestrator(tmp_path)
+    app = create_app(store=orchestrator.store, orchestrator=orchestrator)
+    client = TestClient(app)
+
+    outside_secret = tmp_path / "secret.txt"
+    outside_secret.write_text("outside-runs-secret", encoding="utf-8")
+
+    create = client.post(
+        "/api/runs",
+        json={"pipeline": {"name": "artifact", "working_dir": str(tmp_path), "nodes": [{"id": "alpha", "agent": "codex", "prompt": "artifact output"}]}},
+    )
+    run_id = create.json()["id"]
+    asyncio.run(orchestrator.wait(run_id, timeout=5))
+
+    sibling_run_file = client.get(f"/api/runs/{run_id}/artifacts/%2E%2E/run.json")
+    assert sibling_run_file.status_code == 400
+
+    outside_runs_file = client.get("/api/runs/%2E%2E/artifacts/%2E%2E/secret.txt")
+    assert outside_runs_file.status_code == 400
+
+
+async def test_store_rejects_artifact_write_path_traversal(tmp_path):
+    store = RunStore(tmp_path / "runs")
+    await store.create_run(
+        RunRecord(
+            id="run",
+            pipeline={"name": "p", "nodes": [{"id": "alpha", "agent": "codex", "prompt": "hi"}]},
+        )
+    )
+
+    with pytest.raises(ValueError, match="path segment"):
+        await store.write_artifact_text("run", "../../outside", "output.txt", "pwned")
+    assert not (tmp_path / "outside" / "output.txt").exists()
+
+
+async def test_store_rejects_artifact_read_path_traversal(tmp_path):
+    store = RunStore(tmp_path / "runs")
+    outside_secret = tmp_path / "secret.txt"
+    outside_secret.write_text("outside-runs-secret", encoding="utf-8")
+
+    with pytest.raises(ValueError, match="path segment"):
+        store.read_artifact_text("..", "..", "secret.txt")
+
+
 def test_api_rejects_non_json_content_type(tmp_path):
     orchestrator = make_orchestrator(tmp_path)
     app = create_app(store=orchestrator.store, orchestrator=orchestrator)