Release v0.51.221 — Release GO (stage-p3e — block all workspace symlink escapes + portable TOCTOU hardening [security]) (#3398) (#3451)
Some checks failed
Release & Docker / release (push) Has been cancelled

* [security] fix(workspace): block all symlink escapes from the selected workspace (#3398, @Hinotoi-agent)

Previously safe_resolve_ws allowed a symlink placed inside a workspace to resolve
to an external host path as long as it wasn't a system dir (/etc, /proc, etc).
But the workspace file API is reachable by LLM agent tool calls (read_file_content),
so an in-workspace symlink to ~/.ssh, ~/.hermes/auth.json (credentials), etc. was a
real read path. Now ANY symlink escape is blocked: safe_resolve_ws resolves and
requires the result stay under the workspace root; list_dir hides escaping symlinks
(they could never be opened anyway); internal symlinks resolving back under the
workspace still work. Updated the upload symlink-target test to accept the new
400 'Path traversal blocked' rejection (was 403) — the invariant (nothing lands
outside the workspace) is unchanged.

Co-authored-by: Hinotoi-agent <Hinotoi-agent@users.noreply.github.com>

* docs(changelog): v0.51.221 release header for #3398 symlink-escape security fix

* [security] harden workspace file API against symlink-swap TOCTOU via portable anchored openat-walk (#3398 follow-up)

Codex review of #3398 flagged that safe_resolve_ws() validates a path but
list_dir/read_file_content/upload/extraction then re-open by pathname, leaving a
TOCTOU window: a symlink swapped in AFTER the check could still escape. (This
race pre-existed #3398; closing it here so the containment is complete.)

A first attempt used /proc/self/fd for the post-open containment check, but that
BRICKS workspace browsing on macOS/Windows (no /proc → every read/list rejected).
This version is portable:

- open_anchored_fd(): opens the (already symlink-resolved) target
  component-by-component from the workspace root via 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. No /proc dependency. Used by read_file_content
  (read from the fd) and list_dir (enumerate via os.scandir(fd), per-entry
  fstatat/readlinkat).
- open_anchored_create_fd(): same anchored walk for writes, creating missing
  intermediate dirs with mkdir(dir_fd=) and the leaf with O_CREAT|O_EXCL|
  O_NOFOLLOW. Used by the workspace upload write AND archive (zip+tar) member
  writes, anchored against the TRUE workspace root (not the mutable extraction
  dest_dir, closing Codex's root-swap finding). fd-leak-safe on rejection.
- Portability: gated on os.supports_dir_fd; platforms without it (Windows, where
  symlink creation needs admin) fall back to a plain O_NOFOLLOW open/exclusive
  create — no new race protection but no regression vs the prior path-based code.

Legit in-workspace symlinks still resolve and read/list fine (safe_resolve_ws
collapses them to a real in-workspace path, which the anchored walk then opens).
Verified: the swap-race leaks external content against the old path-based read
and is blocked here; macOS-class symlinked-root workspaces work; no fd leak over
300 rejected creates. Adds TOCTOU + anchored-create regression tests.

* [security] close 3 more #3398 TOCTOU gaps from Codex r3: root-swap, pre-create mkdir, Windows list_dir fallback

Codex round-3 review found three residual issues in the anchored openat-walk:

1. (CORE) The workspace ROOT itself could be swapped to a symlink after
   resolve() but before the root os.open() — add _O_NOFOLLOW to the root open in
   open_anchored_fd() and open_anchored_create_fd() (and make_anchored_dir()), so
   a raced root symlink is refused. Verified: root-swap race now blocked.

2. (SILENT) Upload/extraction still did pathname Path.mkdir() AFTER the
   containment check, so a raced symlink component could make the server create
   dirs outside the workspace before the anchored file create rejected. Removed
   the redundant member_path.parent.mkdir() calls (open_anchored_create_fd
   already creates intermediates via anchored mkdirat) and replaced the two
   base-dir mkdirs (upload target dir + archive extraction root) with a new
   make_anchored_dir() that walks from the true workspace root via
   openat+O_NOFOLLOW + mkdir(dir_fd=).

3. (CORE) list_dir() unconditionally used os.scandir(fd)/os.stat(dir_fd=)/
   os.readlink(dir_fd=), which would brick workspace browsing on platforms
   without os.supports_dir_fd (Windows). Split list_dir() into a _DIR_FD_OK
   anchored branch and a path-based fallback branch (prior behaviour) sharing one
   _process() entry builder. open_anchored_create_fd()'s Windows fallback now also
   creates parent dirs.

Adds regression tests: no-dir_fd fallback (list+read+create+symlink filtering)
and the root-swap race. All prior TOCTOU + anchored-create tests still green.

* fix(workspace): portable symlink-loop filtering in list_dir via follow-stat ELOOP

CI on Python 3.13 caught test_mutual_symlink_loop_filtered failing: a mutual
symlink loop (a->b->a) was NOT filtered from the listing. Root cause: the new
readlink-based cycle detection relied on (target_resolved / raw_link).resolve()
RAISING on a loop, but Path.resolve() loop handling differs by Python version
(3.11 raises RuntimeError, 3.13 can return a path), so the loop slipped through
on 3.13.

Fix: compute a version-independent 'reachable' flag per symlink via
os.stat(..., follow_symlinks=True) — the syscall reliably returns ELOOP for
mutual/self loops and ENOENT for broken targets on every platform/version. A
symlink whose follow-stat raises can never be opened, so list_dir filters it.
Applied in both the dir_fd-anchored branch (fd-relative stat) and the Windows
path-based fallback branch. Mutual loop now filtered on all versions.

---------

Co-authored-by: nesquena-hermes <[email protected]>
Co-authored-by: Hinotoi-agent <Hinotoi-agent@users.noreply.github.com>
This commit is contained in:
nesquena-hermes
2026-06-02 16:56:30 -07:00
committed by GitHub
parent 42aef61b31
commit 7c48c37629
6 changed files with 686 additions and 101 deletions

View File

@@ -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

View File

@@ -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.

View File

@@ -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 ──────────────────────────────────────────────────────────

View File

@@ -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")

View File

@@ -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

View File

@@ -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