[eric] security: contain workspace file paths with realpath and a whole-component boundary

backend/apps/outputs/outputs.py

@@ -41,7 +41,9 @@ from backend.apps.outputs.workspace_io import (
     save,
     load,
     load_output,
+    resolve_in_workspace,
     walk_directory,
+    workspace_root,
     would_shrink_oversize_file,
 )
 from backend.apps.outputs.prompts import VIBE_CODE_SYSTEM_PROMPT
@@ -75,8 +77,8 @@ outputs = SubApp("outputs", outputs_lifespan)
 async def serve_workspace_file(workspace_id: str, filepath: str, p_d: str = ""):
     """Serve a file from a workspace folder. For index.html, inject OUTPUT data."""
     folder = os.path.join(WORKSPACE_DIR, workspace_id)
-    full_path = os.path.normpath(os.path.join(folder, filepath))
-    if not full_path.startswith(os.path.normpath(folder)):
+    full_path = resolve_in_workspace(folder, filepath)
+    if full_path is None:
         raise HTTPException(status_code=403, detail="Path traversal not allowed")
     if not os.path.isfile(full_path):
         raise HTTPException(status_code=404, detail="File not found")
@@ -362,8 +364,8 @@ async def seed_workspace(body: WorkspaceSeedRequest):
     # Legacy flat path. Seed only fills in MISSING files; it never overwrites what's already on disk. A reopen re-sends the inline output.files snapshot, which lags behind whatever the agent just wrote to the workspace; writing it back reverted every edited file (new files survived, edited ones snapped to the snapshot). Disk wins once an app exists.
     if body.files:
         for rel_path, content in body.files.items():
-            full_path = os.path.normpath(os.path.join(folder, rel_path))
-            if not full_path.startswith(os.path.normpath(folder)):
+            full_path = resolve_in_workspace(folder, rel_path)
+            if full_path is None:
                 continue
             if os.path.exists(full_path):
                 continue
@@ -523,10 +525,8 @@ async def write_workspace_file(workspace_id: str, filepath: str, body: dict):
     folder = os.path.join(WORKSPACE_DIR, workspace_id)
     if not os.path.isdir(folder):
         raise HTTPException(status_code=404, detail="Workspace not found")
-    folder_norm = os.path.normpath(folder)
-    full_path = os.path.normpath(os.path.join(folder, filepath))
-    # `startswith(folder_norm + os.sep)` (not just folder_norm) so a workspace `abc-123` can't be tricked into writing into a sibling `abc-1234-evil`, prefix-string collision rather than path-component containment. Today's UUID-format ids make the collision unlikely in practice, but the check is one character and immunizes future id schemes.
-    if full_path != folder_norm and not full_path.startswith(folder_norm + os.sep):
+    full_path = resolve_in_workspace(folder, filepath)
+    if full_path is None:
         raise HTTPException(status_code=403, detail="Path traversal not allowed")
     content = body.get("content", "")
     if would_shrink_oversize_file(full_path, content):
@@ -543,14 +543,14 @@ async def delete_workspace_file(workspace_id: str, filepath: str):
     folder = os.path.join(WORKSPACE_DIR, workspace_id)
     if not os.path.isdir(folder):
         raise HTTPException(status_code=404, detail="Workspace not found")
-    folder_norm = os.path.normpath(folder)
-    full_path = os.path.normpath(os.path.join(folder, filepath))
-    if full_path != folder_norm and not full_path.startswith(folder_norm + os.sep):
+    full_path = resolve_in_workspace(folder, filepath)
+    if full_path is None:
         raise HTTPException(status_code=403, detail="Path traversal not allowed")
     if os.path.isfile(full_path):
         os.remove(full_path)
+        root = workspace_root(folder)
         parent = os.path.dirname(full_path)
-        while parent != os.path.normpath(folder):
+        while parent != root:
             if os.path.isdir(parent) and not os.listdir(parent):
                 os.rmdir(parent)
                 parent = os.path.dirname(parent)

backend/apps/outputs/workspace_io.py

@@ -3,6 +3,7 @@ the workspace file-tree walk used by the polling read endpoint."""
 
 import logging
 import os
+from typing import Optional
 
 from fastapi import HTTPException
 
@@ -58,6 +59,29 @@ def app_workspace_dir(output_id: str) -> str | None:
     return path if os.path.isdir(path) else None
 
 
+def workspace_root(folder: str) -> str:
+    """The canonical form of a workspace folder. Every containment check and
+    every path handed back to a caller is anchored here."""
+    return os.path.realpath(folder)
+
+
+def resolve_in_workspace(folder: str, relative_path: str) -> Optional[str]:
+    """Resolve `relative_path` under workspace `folder`, or None if it escapes.
+
+    The single containment guard for every workspace read/write/delete.
+    `realpath`, never `normpath`: normpath is string math that strolls straight
+    through a symlink pointing out of the workspace. Both sides are resolved so
+    a symlinked ancestor (macOS `/var` -> `/private/var`, every tempdir) doesn't
+    reject a legitimate file. It also compares whole path COMPONENTS, not
+    strings: without the trailing separator, workspace `abc` owns `abc-evil`.
+    """
+    root = workspace_root(folder)
+    target = os.path.realpath(os.path.join(root, relative_path))
+    if target != root and not target.startswith(root + os.sep):
+        return None
+    return target
+
+
 # Build/install/cache directories that the polling endpoint must never descend into. Without this skip-list the workspace endpoint reads `node_modules/` (300 MB of MUI source, when it's a real dir and not a symlink), `.venv/` (10k+ Python files from the hardlinked cache), `__pycache__/`, `dist/`, `.git/`, etc; every 2 seconds while the agent is active. Result: backend CPU pegged on JSON-serializing auto-generated chunks the frontend will then throw away. The frontend already filters these for display; this skip is the real fix.
 WALK_SKIP_DIRS = frozenset({
     "node_modules",

backend/tests/test_workspace_path_containment.py

@@ -0,0 +1,188 @@
+"""Workspace path containment for the Output file endpoints (issues #135, #136).
+
+#135: all four guards used `os.path.normpath`, which is pure string math and
+never resolves symlinks, so a symlink planted inside a workspace pointed the
+read/write/delete anywhere on disk.
+
+#136: `serve_workspace_file` and `seed_workspace` compared with a bare
+`startswith(folder)`, so a sibling folder whose name merely PREFIXES the
+workspace (`abc` vs `abc-evil`) sailed through a guard meant for `abc`.
+
+Run:
+    backend/.venv/bin/python -m pytest backend/tests/test_workspace_path_containment.py -v
+"""
+
+import asyncio
+import os
+from typing import Any, Dict
+
+import pytest
+from fastapi import HTTPException
+
+from backend.apps.outputs import outputs as outputs_mod
+from backend.apps.outputs.models import WorkspaceSeedRequest
+from backend.apps.outputs.workspace_io import resolve_in_workspace
+
+
+def p_run(coro: Any) -> Any:
+    return asyncio.new_event_loop().run_until_complete(coro)
+
+
+@pytest.fixture
+def workspace(monkeypatch: pytest.MonkeyPatch, tmp_path: Any) -> Dict[str, str]:
+    """A workspace root with one real workspace, a prefix-colliding sibling, and
+    an out-of-tree secret, wired into every endpoint under test."""
+    root = tmp_path / "workspaces"
+    (root / "abc").mkdir(parents=True)
+    (root / "abc-evil").mkdir()
+    (root / "abc" / "index.html").write_text("<html>hi</html>")
+    (root / "abc-evil" / "loot.txt").write_text("sibling loot")
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    (outside / "secret.txt").write_text("top secret")
+    monkeypatch.setattr(outputs_mod, "WORKSPACE_DIR", str(root))
+    return {"root": str(root), "folder": str(root / "abc"), "outside": str(outside)}
+
+
+def p_symlink(target: str, link: str) -> None:
+    try:
+        os.symlink(target, link, target_is_directory=os.path.isdir(target))
+    except (OSError, NotImplementedError) as e:
+        pytest.skip(f"symlinks unavailable on this platform: {e}")
+
+
+# --- the resolver itself -----------------------------------------------------
+
+def test_plain_relative_path_resolves(workspace: Dict[str, str]) -> None:
+    got = resolve_in_workspace(workspace["folder"], "sub/dir/file.txt")
+    assert got == os.path.join(os.path.realpath(workspace["folder"]), "sub", "dir", "file.txt")
+
+
+def test_folder_itself_is_inside(workspace: Dict[str, str]) -> None:
+    assert resolve_in_workspace(workspace["folder"], "") == os.path.realpath(workspace["folder"])
+
+
+def test_dotdot_escape_is_rejected(workspace: Dict[str, str]) -> None:
+    assert resolve_in_workspace(workspace["folder"], "../../outside/secret.txt") is None
+
+
+def test_absolute_path_is_rejected(workspace: Dict[str, str]) -> None:
+    assert resolve_in_workspace(workspace["folder"], "/etc/hosts") is None
+
+
+def test_sibling_prefix_collision_is_rejected(workspace: Dict[str, str]) -> None:
+    """#136: `<root>/abc-evil/loot.txt` string-prefixes `<root>/abc`."""
+    assert resolve_in_workspace(workspace["folder"], "../abc-evil/loot.txt") is None
+
+
+def test_symlink_inside_workspace_is_rejected(workspace: Dict[str, str]) -> None:
+    """#135: normpath happily walks `esc` because it never resolves the link."""
+    p_symlink(workspace["outside"], os.path.join(workspace["folder"], "esc"))
+    assert resolve_in_workspace(workspace["folder"], "esc/secret.txt") is None
+
+
+def test_symlinked_file_inside_workspace_is_rejected(workspace: Dict[str, str]) -> None:
+    p_symlink(os.path.join(workspace["outside"], "secret.txt"),
+              os.path.join(workspace["folder"], "leak.txt"))
+    assert resolve_in_workspace(workspace["folder"], "leak.txt") is None
+
+
+def test_symlinked_workspace_root_is_not_false_rejected(tmp_path: Any) -> None:
+    """Both sides get realpath'd, so a symlinked ancestor (macOS /var ->
+    /private/var, every tempfile path) must not reject a legitimate file."""
+    real = tmp_path / "real_ws"
+    real.mkdir()
+    (real / "index.html").write_text("hi")
+    link = tmp_path / "linked_ws"
+    p_symlink(str(real), str(link))
+    got = resolve_in_workspace(str(link), "index.html")
+    assert got == os.path.join(os.path.realpath(str(real)), "index.html")
+
+
+def test_symlink_pointing_back_inside_is_allowed(workspace: Dict[str, str]) -> None:
+    """Containment, not a symlink ban: a link that stays in the workspace works."""
+    inner = os.path.join(workspace["folder"], "assets")
+    os.makedirs(inner)
+    p_symlink(inner, os.path.join(workspace["folder"], "static"))
+    assert resolve_in_workspace(workspace["folder"], "static/logo.svg") == os.path.join(
+        os.path.realpath(inner), "logo.svg"
+    )
+
+
+# --- the four endpoints that guard with it -----------------------------------
+
+def test_serve_rejects_symlink_escape(workspace: Dict[str, str]) -> None:
+    p_symlink(workspace["outside"], os.path.join(workspace["folder"], "esc"))
+    with pytest.raises(HTTPException) as e:
+        p_run(outputs_mod.serve_workspace_file("abc", "esc/secret.txt"))
+    assert e.value.status_code == 403
+
+
+def test_serve_rejects_sibling_prefix(workspace: Dict[str, str]) -> None:
+    with pytest.raises(HTTPException) as e:
+        p_run(outputs_mod.serve_workspace_file("abc", "../abc-evil/loot.txt"))
+    assert e.value.status_code == 403
+
+
+def test_serve_still_serves_a_real_file(workspace: Dict[str, str]) -> None:
+    resp = p_run(outputs_mod.serve_workspace_file("abc", "index.html"))
+    assert b"hi" in resp.body
+
+
+def test_seed_skips_symlink_escape(workspace: Dict[str, str]) -> None:
+    p_symlink(workspace["outside"], os.path.join(workspace["folder"], "esc"))
+    p_run(outputs_mod.seed_workspace(WorkspaceSeedRequest(
+        workspace_id="abc", files={"esc/planted.txt": "x"}, template_mode="flat",
+    )))
+    assert not os.path.exists(os.path.join(workspace["outside"], "planted.txt"))
+
+
+def test_seed_skips_sibling_prefix(workspace: Dict[str, str]) -> None:
+    p_run(outputs_mod.seed_workspace(WorkspaceSeedRequest(
+        workspace_id="abc", files={"../abc-evil/planted.txt": "x"}, template_mode="flat",
+    )))
+    assert not os.path.exists(os.path.join(workspace["root"], "abc-evil", "planted.txt"))
+
+
+def test_seed_still_writes_a_real_file(workspace: Dict[str, str]) -> None:
+    p_run(outputs_mod.seed_workspace(WorkspaceSeedRequest(
+        workspace_id="abc", files={"app/main.py": "print(1)"}, template_mode="flat",
+    )))
+    assert os.path.isfile(os.path.join(workspace["folder"], "app", "main.py"))
+
+
+def test_write_rejects_symlink_escape(workspace: Dict[str, str]) -> None:
+    p_symlink(workspace["outside"], os.path.join(workspace["folder"], "esc"))
+    with pytest.raises(HTTPException) as e:
+        p_run(outputs_mod.write_workspace_file("abc", "esc/planted.txt", {"content": "x"}))
+    assert e.value.status_code == 403
+    assert not os.path.exists(os.path.join(workspace["outside"], "planted.txt"))
+
+
+def test_write_still_writes_a_real_file(workspace: Dict[str, str]) -> None:
+    p_run(outputs_mod.write_workspace_file("abc", "notes/todo.md", {"content": "hello"}))
+    assert (open(os.path.join(workspace["folder"], "notes", "todo.md")).read()) == "hello"
+
+
+def test_delete_rejects_symlink_escape(workspace: Dict[str, str]) -> None:
+    p_symlink(workspace["outside"], os.path.join(workspace["folder"], "esc"))
+    with pytest.raises(HTTPException) as e:
+        p_run(outputs_mod.delete_workspace_file("abc", "esc/secret.txt"))
+    assert e.value.status_code == 403
+    assert os.path.exists(os.path.join(workspace["outside"], "secret.txt"))
+
+
+def test_delete_still_deletes_a_real_file(workspace: Dict[str, str]) -> None:
+    victim = os.path.join(workspace["folder"], "deep", "gone.txt")
+    os.makedirs(os.path.dirname(victim))
+    open(victim, "w").write("x")
+    p_run(outputs_mod.delete_workspace_file("abc", "deep/gone.txt"))
+    assert not os.path.exists(victim)
+    assert not os.path.isdir(os.path.dirname(victim))
+
+
+def test_delete_prunes_up_to_but_not_past_the_workspace(workspace: Dict[str, str]) -> None:
+    victim = os.path.join(workspace["folder"], "only.txt")
+    open(victim, "w").write("x")
+    p_run(outputs_mod.delete_workspace_file("abc", "only.txt"))
+    assert os.path.isdir(workspace["folder"])