diff --git a/roboco/runtime/orchestrator.py b/roboco/runtime/orchestrator.py index 806e5ff6..be4bbde0 100644 --- a/roboco/runtime/orchestrator.py +++ b/roboco/runtime/orchestrator.py @@ -16,6 +16,7 @@ import asyncio import contextlib import json import os +import re import shutil import tempfile from dataclasses import dataclass @@ -877,6 +878,20 @@ class AgentOrchestrator: img, f"{docker_dir}/{dockerfile}", build_context ) + @staticmethod + def _safe_agent_path_segment(agent_id: str) -> str: + """Return ``agent_id`` if it is safe as a single path segment, else raise. + + ``agent_id`` reaches the grok usage dir from request-facing call sites, so + it must not be able to traverse the path. Allow only the slug / uuid + charset the orchestrator actually assigns (alphanumerics, hyphen, + underscore); a value with a separator, ``..``, or any other character is + rejected rather than stripped. + """ + if re.fullmatch(r"[A-Za-z0-9_-]+", agent_id): + return agent_id + raise ValueError(f"unsafe agent id for a filesystem path: {agent_id!r}") + @staticmethod def _grok_usage_dir(agent_id: str) -> Path: """Per-agent grok usage dir, branched compose-vs-local. @@ -885,11 +900,14 @@ class AgentOrchestrator: (``_ensure_grok_usage_dir``) and the finalize read side (``_grok_usage_json``) so they can never drift: in compose the orchestrator sees the mounted host dir at ``GROK_USAGE_DATA_DIR``; in local mode the - container's usage.json lands under the shared tempdir. + container's usage.json lands under the shared tempdir. ``agent_id`` is + validated as a single safe path segment first (path-injection barrier for + both the mount and the finalize read). """ + safe_agent_id = AgentOrchestrator._safe_agent_path_segment(agent_id) if PROJECT_HOST_PATH: - return Path(GROK_USAGE_DATA_DIR) / agent_id - return Path(tempfile.gettempdir()) / "roboco-grok-usage" / agent_id + return Path(GROK_USAGE_DATA_DIR) / safe_agent_id + return Path(tempfile.gettempdir()) / "roboco-grok-usage" / safe_agent_id def _ensure_grok_usage_dir(self, agent_id: str) -> None: """Pre-create the agent's grok usage dir (world-writable) before the mount. diff --git a/tests/unit/runtime/test_grok_usage_finalize.py b/tests/unit/runtime/test_grok_usage_finalize.py index bbf00a2b..497fca74 100644 --- a/tests/unit/runtime/test_grok_usage_finalize.py +++ b/tests/unit/runtime/test_grok_usage_finalize.py @@ -94,6 +94,22 @@ def test_grok_usage_dir_branches_compose_vs_local( ) +@pytest.mark.parametrize( + "bad", + ["..", ".", "../etc", "a/b", "a\\b", "", "be-dev-1/../x", "x\x00y"], +) +def test_grok_usage_dir_rejects_path_traversal(bad: str) -> None: + # agent_id reaches the usage dir from request-facing call sites; a value that + # could traverse the path must be rejected, not used to build a Path. + with pytest.raises(ValueError, match="unsafe agent id"): + AgentOrchestrator._grok_usage_dir(bad) + + +def test_safe_agent_path_segment_accepts_real_slugs() -> None: + for slug in ("be-dev-1", "pr-reviewer-1", "main-pm", "intake", "secretary"): + assert AgentOrchestrator._safe_agent_path_segment(slug) == slug + + def test_grok_usage_json_reads_the_real_local_dir( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: