diff --git a/backend/apps/outputs/outputs.py b/backend/apps/outputs/outputs.py index 51c480a4..2e4dddef 100644 --- a/backend/apps/outputs/outputs.py +++ b/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) diff --git a/backend/apps/outputs/workspace_io.py b/backend/apps/outputs/workspace_io.py index d9347bcb..c93f711e 100644 --- a/backend/apps/outputs/workspace_io.py +++ b/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", diff --git a/backend/tests/test_workspace_path_containment.py b/backend/tests/test_workspace_path_containment.py new file mode 100644 index 00000000..b427ebe9 --- /dev/null +++ b/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("hi") + (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: `/abc-evil/loot.txt` string-prefixes `/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"])