diff --git a/CHANGELOG.md b/CHANGELOG.md index c91ae055..e9346035 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,11 @@ ## [Unreleased] +## [v0.51.221] — 2026-06-02 — Release GO (stage-p3e — block all workspace symlink escapes [security]) + +### Security +- The workspace file API now blocks **all** symlink escapes from the selected workspace, not just symlinks pointing at system directories. Previously a symlink placed inside a workspace could resolve to an arbitrary external host path (e.g. `~/.ssh`, `~/.hermes/auth.json`) and be read through `/api/list` / `read_file_content` — and since that API is reachable by LLM agent tool calls, an imported or crafted workspace could expose credentials. `safe_resolve_ws` now requires the resolved path stay under the workspace root, `list_dir` hides escaping symlinks (they could never be opened anyway), and `read_file_content` rejects them. Symlinks that resolve back under the workspace still work normally. The directory-list, file-read, file-upload, and archive-extraction paths are additionally hardened against a symlink-swap **TOCTOU** race: each path is opened component-by-component from the workspace root with `O_NOFOLLOW` (an anchored `openat` walk on Linux/macOS, with a plain-open fallback on platforms without `dir_fd` support such as Windows, where creating symlinks needs admin anyway), so a symlink raced into any component after the containment check cannot redirect the read/list/write outside the workspace. Note: an intentional in-workspace symlink pointing to an external directory is no longer followed (#3398, @Hinotoi-agent). + ## [v0.51.220] — 2026-06-02 — Release GN (stage-p3c — fix aux title generation with @provider: model ids) ### Fixed diff --git a/api/upload.py b/api/upload.py index 9399c197..2ce1bb87 100644 --- a/api/upload.py +++ b/api/upload.py @@ -11,7 +11,7 @@ from pathlib import Path from api.config import MAX_UPLOAD_BYTES, STATE_DIR from api.helpers import j, bad from api.models import get_session -from api.workspace import safe_resolve_ws, resolve_trusted_workspace +from api.workspace import safe_resolve_ws, resolve_trusted_workspace, open_anchored_create_fd, make_anchored_dir def _max_extracted_bytes() -> int: @@ -211,7 +211,8 @@ def extract_archive(file_bytes: bytes, filename: str, workspace: Path): dest_dir = safe_resolve_ws(workspace, stem).with_name(stem + '_' + suffix) else: raise ValueError('Could not allocate a unique extraction directory') - dest_dir.mkdir(parents=True, exist_ok=True) + # #3398: create the extraction root race-safely under the true workspace root. + make_anchored_dir(workspace, dest_dir) # Member-count cap: a tiny archive with millions of (possibly empty) members # slips under the byte cap but can exhaust inodes / file descriptors. Bound it. @@ -243,8 +244,12 @@ def extract_archive(file_bytes: bytes, filename: str, workspace: Path): f'{cap // (1024*1024)} MB limit). ' f'Possible zip bomb.' ) - member_path.parent.mkdir(parents=True, exist_ok=True) - with zf.open(member) as src, open(member_path, 'wb') as dst: + # #3398: open_anchored_create_fd creates intermediate dirs + # race-safely under the true workspace root (anchored mkdirat), + # so no pathname member_path.parent.mkdir() before it (which + # could be redirected outside by a raced symlink component). + _mfd = open_anchored_create_fd(workspace, member_path) + with zf.open(member) as src, os.fdopen(_mfd, 'wb', closefd=True) as dst: _chunk_size = 65536 while True: chunk = src.read(_chunk_size) @@ -281,10 +286,13 @@ def extract_archive(file_bytes: bytes, filename: str, workspace: Path): f'{cap // (1024*1024)} MB limit). ' f'Possible zip bomb.' ) - member_path.parent.mkdir(parents=True, exist_ok=True) + # #3398: anchored member create makes intermediate dirs + # race-safely; no pathname member_path.parent.mkdir() first. src_obj = tf.extractfile(member) if src_obj: - with src_obj as src, open(member_path, 'wb') as dst: + # #3398: fd-anchored member create under the TRUE workspace root. + _mfd = open_anchored_create_fd(workspace, member_path) + with src_obj as src, os.fdopen(_mfd, 'wb', closefd=True) as dst: _chunk_size = 65536 while True: chunk = src.read(_chunk_size) @@ -428,7 +436,12 @@ def handle_workspace_upload(handler): # workspace==target equality case, so the normal subpath='' path passes.) if not target_dir.resolve().is_relative_to(workspace.resolve()): return j(handler, {'error': 'Upload target escapes workspace'}, status=403) - target_dir.mkdir(parents=True, exist_ok=True) + # #3398: create the upload target dir race-safely under the workspace root + # (anchored mkdirat) so a raced symlink subpath can't mkdir outside. + try: + make_anchored_dir(workspace, target_dir) + except (ValueError, OSError): + return j(handler, {'error': 'Upload target escapes workspace'}, status=403) results = [] for _field_name, (filename, file_bytes) in files.items(): @@ -458,7 +471,19 @@ def handle_workspace_upload(handler): else: return j(handler, {'error': 'Too many uploads with the same filename'}, status=400) - dest.write_bytes(file_bytes) + # #3398 TOCTOU hardening: create the destination via an anchored + # openat-walk from the true workspace root with O_CREAT|O_EXCL| + # O_NOFOLLOW, so a symlink raced into any path component after the + # containment checks above cannot redirect the write outside the + # workspace. The dedup loop guarantees `dest` does not exist. + try: + _wfd = open_anchored_create_fd(workspace, dest.resolve()) + except FileExistsError: + return j(handler, {'error': f'Upload destination already exists: {safe_name}'}, status=409) + except (ValueError, OSError): + return j(handler, {'error': f'Path traversal blocked: {safe_name}'}, status=403) + with os.fdopen(_wfd, 'wb', closefd=True) as _wfh: + _wfh.write(file_bytes) mime = mimetypes.guess_type(safe_name)[0] or 'application/octet-stream' # For archives, optionally extract into the target directory. diff --git a/api/workspace.py b/api/workspace.py index d76f3908..14990e6c 100644 --- a/api/workspace.py +++ b/api/workspace.py @@ -671,75 +671,233 @@ def validate_workspace_to_add(path: str) -> Path: def safe_resolve_ws(root: Path, requested: str) -> Path: """Resolve a relative path inside a workspace root, raising ValueError on traversal. - Symlinks whose *unresolved* path is within the workspace root are allowed — - the user placed them there intentionally. Only raw ``..`` traversal outside - the root is blocked. + Both raw ``..`` traversal and symlink escapes are blocked. Workspace file + APIs can be reached by browser UI actions and agent/tool calls, so a symlink + inside the workspace must not expand the trusted workspace boundary to an + arbitrary host path. """ - import os - unresolved = root / requested - resolved = unresolved.resolve() - # Fast path: resolved path is inside root (covers most cases) + root_resolved = root.resolve() + resolved = (root / requested).resolve() try: - resolved.relative_to(root.resolve()) - return resolved - except ValueError: - pass - # Symlink path: normalize '..' (without following symlinks) and check - # os.path.normpath collapses '..' but does NOT follow symlinks. - norm = Path(os.path.normpath(str(unresolved))) - try: - norm.relative_to(root) + resolved.relative_to(root_resolved) except ValueError: raise ValueError(f"Path traversal blocked: {requested}") - # Symlink points outside workspace root — additionally block system directories. - # Even if the user placed the symlink intentionally, prevent reads from - # /etc, /proc, /sys, /dev and other blocked roots (LLM agents can call - # read_file_content via tool calls, not just human users). - if _is_blocked_system_path(resolved): - raise ValueError(f"Path traversal blocked (system dir): {requested}") return resolved +# ── Race-safe (TOCTOU) anchored open ───────────────────────────────────────── +# safe_resolve_ws() validates a path, but if callers then re-open by pathname a +# symlink swapped in AFTER the check could still escape the workspace. To close +# that window we open the (already symlink-resolved) target component-by-component +# from the workspace root using openat (dir_fd) + O_NOFOLLOW: every component must +# be a real, non-symlink entry, so a component swapped to a symlink mid-flight is +# refused. Legit in-workspace symlinks still work because safe_resolve_ws() has +# already collapsed them to their real in-workspace target, and we walk that real +# (symlink-free) path. Portable: uses os.supports_dir_fd where available (Linux, +# macOS); on platforms without dir_fd support (Windows — where creating symlinks +# also requires admin) we fall back to a plain pathname open, matching the prior +# behaviour with no regression. + +_DIR_FD_OK = os.open in getattr(os, "supports_dir_fd", set()) +_O_NOFOLLOW = getattr(os, "O_NOFOLLOW", 0) +_O_DIRECTORY = getattr(os, "O_DIRECTORY", 0) + + +def open_anchored_fd(workspace: Path, target: Path, *, want_dir: bool) -> int: + """Open ``target`` race-safely and return an owned file descriptor. + + ``target`` must be the symlink-resolved path returned by safe_resolve_ws() + (i.e. already verified to live under the workspace). Raises FileNotFoundError + if a component is missing / wrong-type, or ValueError if a component was + swapped to a symlink (escape attempt). Caller owns and must close the fd. + """ + root_resolved = workspace.resolve() + # Relative, symlink-free component list (resolve() already collapsed any links). + try: + rel_parts = target.relative_to(root_resolved).parts + except ValueError: + raise ValueError(f"Path traversal blocked: {target}") from None + + if not _DIR_FD_OK: + # Windows / no openat: fall back to a plain pathname open. No new race + # protection, but no regression vs the prior path-based behaviour, and + # symlink creation needs admin on Windows anyway. + flags = os.O_RDONLY | (_O_DIRECTORY if want_dir else 0) | _O_NOFOLLOW + try: + return os.open(str(target), flags) + except OSError: + raise FileNotFoundError(f"Not found: {target}") from None + + # Open the (trusted) workspace root. root_resolved is canonical (resolve() + # collapsed any symlinks to REACH it, e.g. macOS /tmp -> /private/tmp), so its + # final component is legitimately a real directory — O_NOFOLLOW here only fires + # if the root itself was raced into a symlink after resolve() (escape attempt). + fd = os.open(str(root_resolved), os.O_RDONLY | _O_DIRECTORY | _O_NOFOLLOW) + try: + for i, part in enumerate(rel_parts): + is_last = i == len(rel_parts) - 1 + want_directory = (not is_last) or want_dir + flags = os.O_RDONLY | _O_NOFOLLOW | (_O_DIRECTORY if want_directory else 0) + try: + nfd = os.open(part, flags, dir_fd=fd) + except OSError: + # ELOOP (component is a symlink — swapped in) or missing/wrong type. + raise FileNotFoundError(f"Not found: {target}") from None + os.close(fd) + fd = nfd + return fd + except BaseException: + try: + os.close(fd) + except OSError: + pass + raise + + +def open_anchored_create_fd(root: Path, dest: Path) -> int: + """Create ``dest`` for exclusive writing race-safely, anchored under ``root``. + + Walks from ``root`` via openat + O_NOFOLLOW (creating missing intermediate + directories with mkdir(dir_fd=...)), then creates the leaf with + O_CREAT|O_EXCL|O_NOFOLLOW so a symlink raced into any component cannot + redirect the write outside ``root``. ``dest`` must be the resolved path and + must not already exist (callers dedup first). Raises ValueError if ``dest`` + is not under ``root``, FileExistsError if it exists, FileNotFoundError if a + component was swapped to a symlink. Caller owns and must close the returned + write fd. On platforms without dir_fd support (Windows) falls back to a plain + exclusive create — no new race protection but no regression. + """ + root_resolved = root.resolve() + try: + rel_parts = dest.relative_to(root_resolved).parts + except ValueError: + raise ValueError(f"Path traversal blocked: {dest}") from None + if not rel_parts: + raise ValueError(f"Invalid destination: {dest}") + + if not _DIR_FD_OK: + # Windows / no openat: create parent dirs then exclusively create the leaf. + dest.parent.mkdir(parents=True, exist_ok=True) + return os.open(str(dest), os.O_WRONLY | os.O_CREAT | os.O_EXCL | _O_NOFOLLOW, 0o644) + + fd = os.open(str(root_resolved), os.O_RDONLY | _O_DIRECTORY | _O_NOFOLLOW) + try: + for part in rel_parts[:-1]: + try: + nfd = os.open(part, os.O_RDONLY | _O_DIRECTORY | _O_NOFOLLOW, dir_fd=fd) + except FileNotFoundError: + os.mkdir(part, 0o755, dir_fd=fd) + nfd = os.open(part, os.O_RDONLY | _O_DIRECTORY | _O_NOFOLLOW, dir_fd=fd) + except OSError: + # ELOOP — component swapped to a symlink (escape attempt). + raise FileNotFoundError(f"Not found: {dest}") from None + os.close(fd) + fd = nfd + return os.open( + rel_parts[-1], + os.O_WRONLY | os.O_CREAT | os.O_EXCL | _O_NOFOLLOW, + 0o644, + dir_fd=fd, + ) + finally: + try: + os.close(fd) + except OSError: + pass + + +def make_anchored_dir(root: Path, dest: Path) -> None: + """Create directory ``dest`` (and any missing parents) race-safely under ``root``. + + Walks from ``root`` via openat + O_NOFOLLOW, creating each missing component + with mkdir(dir_fd=...), so a symlink raced into any component cannot make the + server create directories outside ``root``. Idempotent (existing dirs are + fine). Raises ValueError if ``dest`` is not under ``root``, FileNotFoundError + if a component was swapped to a symlink. On platforms without dir_fd support + (Windows) falls back to a plain Path.mkdir — no regression. + """ + root_resolved = root.resolve() + dest_resolved = dest.resolve() + if dest_resolved == root_resolved: + return + try: + rel_parts = dest_resolved.relative_to(root_resolved).parts + except ValueError: + raise ValueError(f"Path traversal blocked: {dest}") from None + + if not _DIR_FD_OK: + dest.mkdir(parents=True, exist_ok=True) + return + + fd = os.open(str(root_resolved), os.O_RDONLY | _O_DIRECTORY | _O_NOFOLLOW) + try: + for part in rel_parts: + try: + nfd = os.open(part, os.O_RDONLY | _O_DIRECTORY | _O_NOFOLLOW, dir_fd=fd) + except FileNotFoundError: + os.mkdir(part, 0o755, dir_fd=fd) + nfd = os.open(part, os.O_RDONLY | _O_DIRECTORY | _O_NOFOLLOW, dir_fd=fd) + except OSError: + # ELOOP — component swapped to a symlink (escape attempt). + raise FileNotFoundError(f"Not found: {dest}") from None + os.close(fd) + fd = nfd + finally: + try: + os.close(fd) + except OSError: + pass + + def list_dir(workspace: Path, rel: str='.'): target = safe_resolve_ws(workspace, rel) if not target.is_dir(): raise FileNotFoundError(f"Not a directory: {rel}") ws_resolved = workspace.resolve() + target_resolved = target.resolve() entries = [] - for item in sorted(target.iterdir(), key=lambda p: (not p.is_symlink(), p.is_file(), p.name.lower())): - if item.is_symlink(): - # Resolve the symlink target and check if it stays within workspace + + def _process(name, is_symlink, raw_link, lstat_result, reachable): + """Append one directory entry. ``raw_link`` is the os.readlink() result + for symlinks (else None); ``lstat_result`` is an os.stat_result obtained + with follow_symlinks=False (else None); ``reachable`` is False when a + follow_symlinks=True stat raised (broken target or symlink loop).""" + if is_symlink: + if raw_link is None: + return + # A symlink whose follow-stat raised (ELOOP / broken target) can never + # be opened — filter it. This catches mutual/self loops portably across + # Python versions where Path.resolve() loop handling differs (3.11 + # raises RuntimeError, 3.13 can return a path), so do not rely on + # resolve() raising for cycle detection. + if not reachable: + return try: - link_target = item.resolve() - except OSError: - continue - # Cycle detection: skip if symlink points back to current dir, - # workspace root, or any ancestor of current dir. - # This must run REGARDLESS of whether target is inside workspace. - if (link_target == target.resolve() or link_target == target - or link_target == ws_resolved): - continue + link_target = (target_resolved / raw_link).resolve() + except (OSError, RuntimeError): + return + # Cycle detection: skip if symlink points back to current dir or root. + if link_target == target_resolved or link_target == ws_resolved: + return try: - target.resolve().relative_to(link_target) - # target is under link_target — link_target is an ancestor → cycle - continue + target_resolved.relative_to(link_target) + return # target is under link_target — ancestor → cycle except ValueError: pass - # Block symlinks that resolve to system directories. + # Hide symlinks that resolve outside the workspace (can never be opened). + try: + link_target.relative_to(ws_resolved) + except ValueError: + return if _is_blocked_system_path(link_target): - continue + return is_dir = link_target.is_dir() - # Keep the display path relative to workspace (don't follow the link) - display_path = str(Path(item.name)) + display_path = name if rel and rel != '.': display_path = rel + '/' + display_path - try: - item_stat = item.lstat() - mtime_ns = item_stat.st_mtime_ns - except OSError: - mtime_ns = None + mtime_ns = lstat_result.st_mtime_ns if lstat_result is not None else None entry = { - 'name': item.name, + 'name': name, 'path': display_path, 'type': 'symlink', 'target': str(link_target), @@ -753,27 +911,118 @@ def list_dir(workspace: Path, rel: str='.'): entry['size'] = None entries.append(entry) else: - # Use rel-based path so entries under symlink targets (outside - # the workspace root) still get a valid workspace-relative path. - entry_path = item.name + entry_path = name if rel and rel != '.': - entry_path = rel + '/' + item.name - try: - item_stat = item.stat() - size = item_stat.st_size if item.is_file() else None - mtime_ns = item_stat.st_mtime_ns - except OSError: + entry_path = rel + '/' + name + if lstat_result is not None: + is_file = stat.S_ISREG(lstat_result.st_mode) + size = lstat_result.st_size if is_file else None + mtime_ns = lstat_result.st_mtime_ns + is_dir_entry = stat.S_ISDIR(lstat_result.st_mode) + else: size = None mtime_ns = None + is_dir_entry = False entries.append({ - 'name': item.name, + 'name': name, 'path': entry_path, - 'type': 'dir' if item.is_dir() else 'file', + 'type': 'dir' if is_dir_entry else 'file', 'size': size, 'mtime_ns': mtime_ns, }) - if len(entries) >= 200: - break + + if _DIR_FD_OK: + # #3398 TOCTOU hardening (Linux/macOS): open the directory via an anchored + # openat-walk (O_NOFOLLOW on every component) and enumerate via the verified + # fd (os.scandir(fd) + fd-relative fstatat/readlinkat), so a path component + # swapped to an escaping symlink after safe_resolve_ws() cannot redirect the + # listing. + def _sort_key_de(de): + try: + is_link = de.is_symlink() + except OSError: + is_link = False + is_file = False + if not is_link: + try: + is_file = de.is_file() + except OSError: + pass + return (not is_link, is_file, de.name.lower()) + + dir_fd = open_anchored_fd(workspace, target, want_dir=True) + try: + st = os.fstat(dir_fd) + if not stat.S_ISDIR(st.st_mode): + raise FileNotFoundError(f"Not a directory: {rel}") + with os.scandir(dir_fd) as scan: + scandir_entries = sorted(scan, key=_sort_key_de) + for de in scandir_entries: + name = de.name + is_symlink = de.is_symlink() + raw_link = None + if is_symlink: + try: + raw_link = os.readlink(name, dir_fd=dir_fd) + except OSError: + raw_link = None + try: + lst = os.stat(name, dir_fd=dir_fd, follow_symlinks=False) + except OSError: + lst = None + # reachable: follow-stat succeeds (filters ELOOP/broken symlinks). + reachable = True + if is_symlink: + try: + os.stat(name, dir_fd=dir_fd, follow_symlinks=True) + except OSError: + reachable = False + _process(name, is_symlink, raw_link, lst, reachable) + if len(entries) >= 200: + break + finally: + try: + os.close(dir_fd) + except OSError: + pass + else: + # Portability fallback (Windows / no dir_fd): path-based enumeration after + # safe_resolve_ws(). No anchored-fd race protection on these platforms, but + # no regression vs the prior behaviour (creating symlinks on Windows needs + # admin anyway), and safe_resolve_ws() still blocks the static escape. + def _sort_key_p(p: Path): + is_link = p.is_symlink() + is_file = False + if not is_link: + try: + is_file = p.is_file() + except OSError: + pass + return (not is_link, is_file, p.name.lower()) + + for item in sorted(target.iterdir(), key=_sort_key_p): + name = item.name + is_symlink = item.is_symlink() + raw_link = None + if is_symlink: + try: + raw_link = os.readlink(str(item)) + except OSError: + raw_link = None + try: + lst = item.lstat() + except OSError: + lst = None + # reachable: follow-stat succeeds (filters ELOOP/broken symlinks). + reachable = True + if is_symlink: + try: + os.stat(str(item), follow_symlinks=True) + except OSError: + reachable = False + _process(name, is_symlink, raw_link, lst, reachable) + if len(entries) >= 200: + break return entries @@ -805,11 +1054,20 @@ def read_file_content(workspace: Path, rel: str) -> dict: target = safe_resolve_ws(workspace, rel) if not target.is_file(): raise FileNotFoundError(f"Not a file: {rel}") - size = target.stat().st_size - if size > MAX_FILE_BYTES: - raise ValueError(f"File too large ({size} bytes, max {MAX_FILE_BYTES})") - content = target.read_text(encoding='utf-8', errors='replace') - return {'path': rel, 'content': content, 'size': size, 'lines': content.count('\n') + 1} + # #3398 TOCTOU hardening: open the resolved file via an anchored openat-walk + # (O_NOFOLLOW on every component) so a path swapped to an escaping symlink + # after safe_resolve_ws() cannot be followed, then read from the fd (not the + # pathname) so the bytes returned are guaranteed to be the verified file. + fd = open_anchored_fd(workspace, target, want_dir=False) + with os.fdopen(fd, 'rb', closefd=True) as fh: + st = os.fstat(fh.fileno()) + if not stat.S_ISREG(st.st_mode): + raise FileNotFoundError(f"Not a file: {rel}") + if st.st_size > MAX_FILE_BYTES: + raise ValueError(f"File too large ({st.st_size} bytes, max {MAX_FILE_BYTES})") + raw = fh.read(MAX_FILE_BYTES + 1) + content = raw.decode('utf-8', errors='replace') + return {'path': rel, 'content': content, 'size': len(raw), 'lines': content.count('\n') + 1} # ── Git detection ────────────────────────────────────────────────────────── diff --git a/tests/test_symlink_cycle_detection.py b/tests/test_symlink_cycle_detection.py index 01a4e162..2878c5c2 100644 --- a/tests/test_symlink_cycle_detection.py +++ b/tests/test_symlink_cycle_detection.py @@ -8,8 +8,8 @@ recursion. Covers: - External symlink dirs (e.g. ln -s /some/path ~/workspace/link) - Self-referencing symlink (ln -s . ~/workspace/loop) - Ancestor symlink (ln -s .. ~/workspace/up) -- Symlink entries carry correct type / is_dir / target fields -- Browsing into a symlink directory via workspace-relative path works +- Internal symlink entries carry correct type / is_dir / target fields +- External symlink directories are hidden from listings and cannot be traversed """ import json import os @@ -57,8 +57,8 @@ def make_session(created_list, ws=None): class TestSymlinkCycleDetection: """Symlink cycle detection in list_dir / safe_resolve_ws.""" - def test_external_symlink_listed_as_symlink(self, cleanup_test_sessions, tmp_path_factory): - """External symlink dir should appear with type='symlink', is_dir=True.""" + def test_external_symlink_filtered_from_listing(self, cleanup_test_sessions, tmp_path_factory): + """External symlink dirs should be hidden from workspace listings.""" ws = tmp_path_factory.mktemp("ws") target = tmp_path_factory.mktemp("target") (target / "file.txt").write_text("hello") @@ -67,25 +67,40 @@ class TestSymlinkCycleDetection: sid, _ = make_session(cleanup_test_sessions, ws) listing = get(f"/api/list?session_id={sid}&path=.") - entries = listing["entries"] - ext = [e for e in entries if e["name"] == "ext"] - assert len(ext) == 1 - assert ext[0]["type"] == "symlink" - assert ext[0]["is_dir"] is True - assert ext[0]["target"] == str(target) + names = [e["name"] for e in listing["entries"]] + assert "ext" not in names - def test_external_symlink_browsable(self, cleanup_test_sessions, tmp_path_factory): - """Listing inside an external symlink dir returns its contents.""" + def test_internal_symlink_listed_as_symlink(self, cleanup_test_sessions, tmp_path_factory): + """Internal symlink dirs should appear with type='symlink', is_dir=True.""" + ws = tmp_path_factory.mktemp("ws") + target = ws / "target" + target.mkdir() + (target / "file.txt").write_text("hello") + link = ws / "internal" + link.symlink_to(target) + + sid, _ = make_session(cleanup_test_sessions, ws) + listing = get(f"/api/list?session_id={sid}&path=.") + entries = listing["entries"] + internal = [e for e in entries if e["name"] == "internal"] + assert len(internal) == 1 + assert internal[0]["type"] == "symlink" + assert internal[0]["is_dir"] is True + assert internal[0]["target"] == str(target) + + def test_external_symlink_not_browsable(self, cleanup_test_sessions, tmp_path_factory): + """Listing inside an external symlink dir is blocked at the workspace boundary.""" ws = tmp_path_factory.mktemp("ws") target = tmp_path_factory.mktemp("target") (target / "inner.txt").write_text("data") (ws / "ext").symlink_to(target) sid, _ = make_session(cleanup_test_sessions, ws) - listing = get(f"/api/list?session_id={sid}&path=ext") - entries = listing["entries"] - names = [e["name"] for e in entries] - assert "inner.txt" in names + try: + get(f"/api/list?session_id={sid}&path=ext") + assert False, "External symlink traversal should be blocked" + except urllib.error.HTTPError as e: + assert e.code in (400, 404, 500) def test_self_referencing_symlink_filtered(self, cleanup_test_sessions, tmp_path_factory): """Symlink pointing to the workspace root itself must be filtered out.""" @@ -112,8 +127,20 @@ class TestSymlinkCycleDetection: names = [e["name"] for e in listing["entries"]] assert "up" not in names, "Ancestor symlink should be filtered" + def test_mutual_symlink_loop_filtered(self, cleanup_test_sessions, tmp_path_factory): + """Mutually recursive symlinks should be skipped instead of raising RuntimeError.""" + ws = tmp_path_factory.mktemp("ws") + (ws / "a").symlink_to(ws / "b") + (ws / "b").symlink_to(ws / "a") + + sid, _ = make_session(cleanup_test_sessions, ws) + listing = get(f"/api/list?session_id={sid}&path=.") + names = [e["name"] for e in listing["entries"]] + assert "a" not in names + assert "b" not in names + def test_symlink_cycle_in_subdir(self, cleanup_test_sessions, tmp_path_factory): - """Symlink cycle inside a symlink target's subtree must not recurse.""" + """External symlink subpaths must be blocked instead of traversed.""" ws = tmp_path_factory.mktemp("ws") target = tmp_path_factory.mktemp("target") (target / "subdir").mkdir() @@ -122,20 +149,23 @@ class TestSymlinkCycleDetection: (ws / "ext").symlink_to(target) sid, _ = make_session(cleanup_test_sessions, ws) - # List root — should show ext but not recurse + # List root — should hide the external symlink and not recurse. listing = get(f"/api/list?session_id={sid}&path=.") names = [e["name"] for e in listing["entries"]] - assert "ext" in names + assert "ext" not in names - # List inside ext/subdir — 'back' should be filtered - listing2 = get(f"/api/list?session_id={sid}&path=ext/subdir") - names2 = [e["name"] for e in listing2["entries"]] - assert "back" not in names2, "Cycle symlink inside external target should be filtered" + # Traversing into ext/subdir crosses the workspace boundary and is blocked. + try: + get(f"/api/list?session_id={sid}&path=ext/subdir") + assert False, "External symlink subpath traversal should be blocked" + except urllib.error.HTTPError as e: + assert e.code in (400, 404, 500) - def test_symlink_file_entry(self, cleanup_test_sessions, tmp_path_factory): - """Symlink to a file should have is_dir=False and include size.""" + def test_internal_symlink_file_entry(self, cleanup_test_sessions, tmp_path_factory): + """Internal symlink to a file should have is_dir=False and include size.""" ws = tmp_path_factory.mktemp("ws") - real = tmp_path_factory.mktemp("real") + real = ws / "real" + real.mkdir() (real / "data.txt").write_text("hello world") (ws / "link.txt").symlink_to(real / "data.txt") @@ -147,6 +177,18 @@ class TestSymlinkCycleDetection: assert link[0]["is_dir"] is False assert link[0]["size"] == 11 # len("hello world") + def test_external_symlink_file_filtered_from_listing(self, cleanup_test_sessions, tmp_path_factory): + """External symlink files should be hidden from workspace listings.""" + ws = tmp_path_factory.mktemp("ws") + real = tmp_path_factory.mktemp("real") + (real / "data.txt").write_text("hello world") + (ws / "link.txt").symlink_to(real / "data.txt") + + sid, _ = make_session(cleanup_test_sessions, ws) + listing = get(f"/api/list?session_id={sid}&path=.") + names = [e["name"] for e in listing["entries"]] + assert "link.txt" not in names + def test_path_traversal_still_blocked(self, cleanup_test_sessions, tmp_path_factory): """Raw .. traversal must still be blocked even with symlink support.""" ws = tmp_path_factory.mktemp("ws") diff --git a/tests/test_workspace_symlink_containment.py b/tests/test_workspace_symlink_containment.py new file mode 100644 index 00000000..9bdb6ecb --- /dev/null +++ b/tests/test_workspace_symlink_containment.py @@ -0,0 +1,251 @@ +import pytest + +from api.workspace import list_dir, read_file_content, safe_resolve_ws + + +def test_safe_resolve_blocks_external_symlink_directory(tmp_path): + workspace = tmp_path / "workspace" + outside = tmp_path / "outside" + workspace.mkdir() + outside.mkdir() + (outside / "secret.txt").write_text("outside", encoding="utf-8") + (workspace / "escape").symlink_to(outside) + + with pytest.raises(ValueError, match="Path traversal blocked"): + safe_resolve_ws(workspace, "escape") + + with pytest.raises(ValueError, match="Path traversal blocked"): + list_dir(workspace, "escape") + + assert "escape" not in {entry["name"] for entry in list_dir(workspace, ".")} + + +def test_read_file_blocks_external_symlink_file(tmp_path): + workspace = tmp_path / "workspace" + outside = tmp_path / "outside" + workspace.mkdir() + outside.mkdir() + (outside / "secret.txt").write_text("outside", encoding="utf-8") + (workspace / "secret-link.txt").symlink_to(outside / "secret.txt") + + with pytest.raises(ValueError, match="Path traversal blocked"): + read_file_content(workspace, "secret-link.txt") + + assert "secret-link.txt" not in {entry["name"] for entry in list_dir(workspace, ".")} + + +def test_internal_symlink_still_resolves_within_workspace(tmp_path): + workspace = tmp_path / "workspace" + workspace.mkdir() + nested = workspace / "nested" + nested.mkdir() + (nested / "inside.txt").write_text("inside", encoding="utf-8") + (workspace / "inside-link.txt").symlink_to(nested / "inside.txt") + + resolved = safe_resolve_ws(workspace, "inside-link.txt") + + assert resolved == (nested / "inside.txt").resolve() + assert read_file_content(workspace, "inside-link.txt")["content"] == "inside" + assert "inside-link.txt" in {entry["name"] for entry in list_dir(workspace, ".")} + + +# ── TOCTOU hardening (#3398): a path that passes safe_resolve_ws() but is then +# swapped to an external symlink before the open must not read/list/write +# outside the workspace. The read/list/write paths use a portable anchored +# openat-walk (openat + O_NOFOLLOW per component, dir_fd where supported). ── + + +def test_read_file_toctou_swap_to_external_symlink_blocked(tmp_path, monkeypatch): + """If the resolved path is swapped to an external symlink AFTER the + safe_resolve_ws() check, read_file_content must refuse, not follow the + symlink and leak external content.""" + import api.workspace as w + + workspace = tmp_path / "workspace" + workspace.mkdir() + (workspace / "data.txt").write_text("LEGIT", encoding="utf-8") + outside = tmp_path / "outside" + outside.mkdir() + (outside / "secret.txt").write_text("SECRET-LEAK", encoding="utf-8") + + real_resolve = w.safe_resolve_ws + + def racing_resolve(root, rel): + p = real_resolve(root, rel) + if rel == "data.txt": + try: + p.unlink() + except OSError: + pass + p.symlink_to(outside / "secret.txt") + return p + + monkeypatch.setattr(w, "safe_resolve_ws", racing_resolve) + try: + result = w.read_file_content(workspace, "data.txt") + assert "SECRET" not in result["content"], "TOCTOU symlink swap leaked external content" + except (FileNotFoundError, ValueError): + pass # refused — the correct outcome + + +def test_list_dir_toctou_swap_to_external_symlink_blocked(tmp_path, monkeypatch): + """If a checked directory path is swapped to an external symlink after + safe_resolve_ws(), list_dir must refuse rather than enumerate the external + directory.""" + import api.workspace as w + + workspace = tmp_path / "workspace" + workspace.mkdir() + (workspace / "sub").mkdir() + outside = tmp_path / "outside" + outside.mkdir() + (outside / "secret.txt").write_text("x", encoding="utf-8") + + real_resolve = w.safe_resolve_ws + + def racing_resolve(root, rel): + p = real_resolve(root, rel) + if rel == "sub": + try: + p.rmdir() + except OSError: + pass + p.symlink_to(outside) + return p + + monkeypatch.setattr(w, "safe_resolve_ws", racing_resolve) + try: + entries = w.list_dir(workspace, "sub") + names = {e["name"] for e in entries} + assert "secret.txt" not in names, "TOCTOU symlink swap leaked external dir listing" + except (FileNotFoundError, ValueError): + pass # refused — the correct outcome + + +def test_anchored_create_blocks_symlinked_component(tmp_path): + """open_anchored_create_fd must refuse to write through a symlinked path + component (the upload / archive-extraction write race), landing nothing + outside the workspace.""" + from api.workspace import open_anchored_create_fd + + workspace = tmp_path / "workspace" + workspace.mkdir() + outside = tmp_path / "outside" + outside.mkdir() + (workspace / "evil").symlink_to(outside) # symlinked intermediate dir + + with pytest.raises((FileNotFoundError, ValueError, OSError)): + open_anchored_create_fd(workspace, (workspace / "evil" / "pwned.txt")) + assert not (outside / "pwned.txt").exists() + + +def test_anchored_create_no_fd_leak_on_rejection(tmp_path): + """Repeated rejected anchored creates must not leak file descriptors.""" + import os + + from api.workspace import open_anchored_create_fd + + if not os.path.isdir("/proc/self/fd"): + pytest.skip("fd-count check requires /proc/self/fd") + + workspace = tmp_path / "workspace" + workspace.mkdir() + outside = tmp_path / "outside" + outside.mkdir() + (workspace / "evil").symlink_to(outside) + + before = len(os.listdir("/proc/self/fd")) + for _ in range(200): + try: + open_anchored_create_fd(workspace, (workspace / "evil" / "x.txt")) + except Exception: + pass + after = len(os.listdir("/proc/self/fd")) + assert after <= before + 2, f"fd leak: before={before} after={after}" + + +def test_anchored_create_nested_autocreates_dirs(tmp_path): + """A normal (non-escaping) nested create works and lands under the workspace.""" + import os + + from api.workspace import open_anchored_create_fd + + workspace = tmp_path / "workspace" + workspace.mkdir() + fd = open_anchored_create_fd(workspace, workspace / "a" / "b" / "file.txt") + os.write(fd, b"hello") + os.close(fd) + assert (workspace / "a" / "b" / "file.txt").read_text() == "hello" + + +def test_list_read_create_work_on_no_dir_fd_fallback(tmp_path, monkeypatch): + """The no-dir_fd portability fallback (Windows path) must still list, read, + and create within the workspace, and still hide/block external symlinks via + the static safe_resolve_ws guard — no fd-relative API that would brick on + platforms without os.supports_dir_fd.""" + import os + + import api.workspace as w + + monkeypatch.setattr(w, "_DIR_FD_OK", False) + + workspace = tmp_path / "workspace" + workspace.mkdir() + (workspace / "a.txt").write_text("hi", encoding="utf-8") + (workspace / "sub").mkdir() + (workspace / "internal").symlink_to(workspace / "sub") + outside = tmp_path / "outside" + outside.mkdir() + (outside / "s.txt").write_text("x", encoding="utf-8") + (workspace / "escape").symlink_to(outside) + + names = {e["name"] for e in w.list_dir(workspace, ".")} + assert "a.txt" in names + assert "internal" in names # legit internal symlink listed + assert "escape" not in names # external symlink hidden + assert w.read_file_content(workspace, "a.txt")["content"] == "hi" + + fd = w.open_anchored_create_fd(workspace, workspace / "new" / "f.txt") + os.write(fd, b"ok") + os.close(fd) + assert (workspace / "new" / "f.txt").read_text() == "ok" + + +def test_read_blocked_when_workspace_root_raced_to_symlink(tmp_path): + """If the workspace root itself is swapped to an external symlink after + resolve() but before the anchored open, read_file_content must refuse + (O_NOFOLLOW on the root open), not follow it and leak external content.""" + import os + import shutil + + import api.workspace as w + + if not w._DIR_FD_OK: + pytest.skip("anchored root-open race only applies on dir_fd platforms") + + outside = tmp_path / "evil" + outside.mkdir() + (outside / "f.txt").write_text("SECRET-LEAK", encoding="utf-8") + wsroot = tmp_path / "wsroot" + wsroot.mkdir() + (wsroot / "f.txt").write_text("LEGIT", encoding="utf-8") + + real_open = os.open + state = {"swapped": False} + + def racing_open(path, *args, **kwargs): + if (not state["swapped"]) and "dir_fd" not in kwargs and str(path) == str(wsroot.resolve()): + state["swapped"] = True + shutil.rmtree(str(wsroot)) + os.symlink(str(outside), str(wsroot)) + return real_open(path, *args, **kwargs) + + os.open = racing_open + try: + try: + result = w.read_file_content(wsroot, "f.txt") + assert "SECRET" not in result["content"], "root-swap race leaked external content" + except (FileNotFoundError, ValueError, NotADirectoryError, OSError): + pass # refused — correct + finally: + os.open = real_open diff --git a/tests/test_workspace_upload.py b/tests/test_workspace_upload.py index 25cfb73a..440bccbb 100644 --- a/tests/test_workspace_upload.py +++ b/tests/test_workspace_upload.py @@ -514,8 +514,12 @@ class TestWorkspaceUploadSymlinkTarget: {"file": ("pwned.txt", b"should not land outside")}, ) # The escaping target must be rejected outright, and nothing may land - # outside the workspace. - assert status == 403, f"expected 403, got status={status} result={result}" + # outside the workspace. Either a 403 (upload-handler symlink-target + # rejection) or a 400 ("Path traversal blocked" from safe_resolve_ws, + # which #3398 made the workspace boundary enforce consistently for all + # symlink escapes) is an acceptable rejection — the invariant is that + # the upload does NOT land outside the workspace. + assert status in (400, 403), f"expected 400/403, got status={status} result={result}" assert not (escape / "pwned.txt").exists() finally: import shutil