fix(sandbox): surface local directory search permission errors (#7234)

* fix(sandbox): surface local directory search permission errors

Permission failures during local list_dir and glob walks must reach the
caller instead of looking like an empty successful result. Transient
filesystem races keep using the existing walk race allowlist.

Fixes #7210

* fix(sandbox): keep denied path in ls/glob permission errors

Guard chmod-based permission tests for non-root POSIX, reverse-resolve
PermissionError.filename in LocalSandbox.list_dir/glob like grep, and
surface that path from ls_tool/glob_tool instead of only the requested root.

* test(sandbox): use workspace/tmp_path leak guards for ls/glob denials
This commit is contained in:
DDT 2026-10-22 08:38:35 +08:00 committed by GitHub
parent 6160304d70
commit a125277984
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 141 additions and 29 deletions

View File

@ -19,6 +19,8 @@ def list_dir(path: str, max_depth: int = 2) -> list[str]:
Raises:
FileNotFoundError: If ``path`` does not exist or is not a directory.
PermissionError: If a directory cannot be listed because of access
permissions. Callers must not treat this as an empty listing.
"""
result: list[str] = []
root_path = Path(path).resolve()
@ -38,34 +40,31 @@ def list_dir(path: str, max_depth: int = 2) -> list[str]:
if current_depth > max_depth:
return
try:
for item in current_path.iterdir():
if should_ignore_name(item.name):
continue
for item in current_path.iterdir():
if should_ignore_name(item.name):
continue
if item.is_symlink():
try:
item_resolved = item.resolve()
if not _is_within_root(item_resolved):
continue
except OSError:
if item.is_symlink():
try:
item_resolved = item.resolve()
if not _is_within_root(item_resolved):
continue
post_fix = "/" if item_resolved.is_dir() else ""
result.append(str(item_resolved) + post_fix)
except OSError:
continue
item_resolved = item.resolve()
if not _is_within_root(item_resolved):
continue
post_fix = "/" if item.is_dir() else ""
post_fix = "/" if item_resolved.is_dir() else ""
result.append(str(item_resolved) + post_fix)
continue
# Recurse into subdirectories if not at max depth
if item.is_dir() and current_depth < max_depth:
_traverse(item, current_depth + 1)
except PermissionError:
pass
item_resolved = item.resolve()
if not _is_within_root(item_resolved):
continue
post_fix = "/" if item.is_dir() else ""
result.append(str(item_resolved) + post_fix)
# Recurse into subdirectories if not at max depth
if item.is_dir() and current_depth < max_depth:
_traverse(item, current_depth + 1)
_traverse(root_path, 1)

View File

@ -1029,6 +1029,13 @@ class LocalSandbox(Sandbox):
if not virtual_children or os.path.exists(resolved_path):
raise
entries = []
except OSError as e:
if e.filename is None:
raise
# Keep custom-mount host paths out of public listing errors.
filename = self._reverse_resolve_path(os.fsdecode(e.filename))
filename2 = self._reverse_resolve_path(os.fsdecode(e.filename2)) if e.filename2 is not None else None
raise type(e)(e.errno, e.strerror, filename, getattr(e, "winerror", None), filename2) from None
# Reverse resolve local paths back to container paths and preserve
# list_dir's trailing "/" marker for directories.
result: list[str] = []
@ -1144,7 +1151,15 @@ class LocalSandbox(Sandbox):
def glob(self, path: str, pattern: str, *, include_dirs: bool = False, max_results: int = 200) -> tuple[list[str], bool]:
resolved_path = Path(self._resolve_path(path))
matches, truncated = find_glob_matches(resolved_path, pattern, include_dirs=include_dirs, max_results=max_results)
try:
matches, truncated = find_glob_matches(resolved_path, pattern, include_dirs=include_dirs, max_results=max_results)
except OSError as e:
if e.filename is None:
raise
# Keep custom-mount host paths out of public search errors.
filename = self._reverse_resolve_path(os.fsdecode(e.filename))
filename2 = self._reverse_resolve_path(os.fsdecode(e.filename2)) if e.filename2 is not None else None
raise type(e)(e.errno, e.strerror, filename, getattr(e, "winerror", None), filename2) from None
return [self._reverse_resolve_path(match) for match in matches], truncated
def grep(

View File

@ -169,7 +169,14 @@ def find_glob_matches(root: Path, pattern: str, *, include_dirs: bool = False, m
if not root.is_dir():
raise NotADirectoryError(root)
for current_root, dirs, files in os.walk(root):
def handle_walk_error(error: OSError) -> None:
# Entries may disappear or change type during enumeration. Other
# errors, such as permission failures, must reach the caller rather
# than being reported as an incomplete successful search.
if error.errno not in _GREP_RACE_ERRNOS:
raise error
for current_root, dirs, files in os.walk(root, onerror=handle_walk_error):
dirs[:] = [name for name in dirs if not should_ignore_name(name)]
# root is already resolved; os.walk builds current_root by joining under root,
# so relative_to() works without an extra stat()/resolve() per directory.

View File

@ -3004,8 +3004,9 @@ def ls_tool(runtime: Runtime, path: str, description: str = "") -> str:
return f"Error: {e}"
except FileNotFoundError:
return f"Error: Directory not found: {requested_path}"
except PermissionError:
return f"Error: Permission denied: {requested_path}"
except PermissionError as e:
detail = _sanitize_error(e, runtime) if e.filename is not None else requested_path
return f"Error: Permission denied: {detail}"
except Exception as e:
return f"Error: Unexpected error listing directory: {_sanitize_error(e, runtime)}"
@ -3069,8 +3070,9 @@ def glob_tool(
return f"Error: Directory not found: {requested_path}"
except NotADirectoryError:
return f"Error: Path is not a directory: {requested_path}"
except PermissionError:
return f"Error: Permission denied: {requested_path}"
except PermissionError as e:
detail = _sanitize_error(e, runtime) if e.filename is not None else requested_path
return f"Error: Permission denied: {detail}"
except Exception as e:
return f"Error: Unexpected error searching paths: {_sanitize_error(e, runtime)}"

View File

@ -1,4 +1,6 @@
import json
import os
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch
@ -489,6 +491,93 @@ def test_find_glob_matches_raises_not_a_directory(tmp_path) -> None:
pass
_NON_ROOT_POSIX = pytest.mark.skipif(
os.name == "nt" or not hasattr(os, "geteuid") or os.geteuid() == 0,
reason="requires non-root POSIX directory permissions",
)
@_NON_ROOT_POSIX
def test_local_list_dir_propagates_permission_error(tmp_path) -> None:
"""Permission-denied roots must not look like an empty successful listing."""
from deerflow.sandbox.local.list_dir import list_dir
locked = tmp_path / "locked"
locked.mkdir()
(locked / "secret.txt").write_text("hidden\n", encoding="utf-8")
locked.chmod(0o000)
try:
with pytest.raises(PermissionError) as caught:
list_dir(str(locked))
assert caught.value.filename is not None
assert Path(os.fsdecode(caught.value.filename)).resolve() == locked.resolve()
finally:
locked.chmod(0o700)
@_NON_ROOT_POSIX
def test_find_glob_matches_propagates_permission_error(tmp_path) -> None:
"""Inaccessible subtrees must surface instead of being skipped as no matches."""
workspace = tmp_path / "workspace"
workspace.mkdir()
locked = workspace / "locked"
locked.mkdir()
(locked / "secret.py").write_text("x = 1\n", encoding="utf-8")
locked.chmod(0o000)
try:
with pytest.raises(PermissionError) as caught:
find_glob_matches(workspace, "**/*.py")
assert caught.value.filename is not None
assert Path(os.fsdecode(caught.value.filename)).resolve() == locked.resolve()
finally:
locked.chmod(0o700)
@_NON_ROOT_POSIX
def test_ls_tool_permission_error_names_denied_entry(tmp_path, monkeypatch) -> None:
"""ls must name the denied entry, not only the requested root."""
runtime = _make_runtime(tmp_path)
workspace = tmp_path / "workspace"
locked = workspace / "locked"
locked.mkdir()
(locked / "secret.txt").write_text("hidden\n", encoding="utf-8")
monkeypatch.setattr("deerflow.sandbox.tools.ensure_sandbox_initialized", lambda _: LocalSandbox(id="local"))
locked.chmod(0o000)
try:
result = ls_tool.func(runtime=runtime, description="list locked", path="/mnt/user-data/workspace/locked")
finally:
locked.chmod(0o700)
assert result.startswith("Error: Permission denied:"), result
assert "/mnt/user-data/workspace/locked" in result
assert str(workspace) not in result
assert str(tmp_path) not in result
@_NON_ROOT_POSIX
def test_glob_tool_permission_error_names_denied_subdirectory(tmp_path, monkeypatch) -> None:
"""glob must name the denied subdirectory, not only the search root."""
runtime = _make_runtime(tmp_path)
workspace = tmp_path / "workspace"
locked = workspace / "locked"
locked.mkdir()
(locked / "secret.py").write_text("x = 1\n", encoding="utf-8")
monkeypatch.setattr("deerflow.sandbox.tools.ensure_sandbox_initialized", lambda _: LocalSandbox(id="local"))
locked.chmod(0o000)
try:
result = glob_tool.func(
runtime=runtime,
description="search locked",
path="/mnt/user-data/workspace",
pattern="**/*.py",
)
finally:
locked.chmod(0o700)
assert result.startswith("Error: Permission denied:"), result
assert "/mnt/user-data/workspace/locked" in result
assert str(workspace) not in result
assert str(tmp_path) not in result
def test_find_grep_matches_accepts_single_file(tmp_path) -> None:
file_path = tmp_path / "file.txt"
file_path.write_text("TODO\n", encoding="utf-8")