fix(sdk): surface OS errors in sandbox ls/read/edit/glob (#3359)

## Summary

Fixes #3105.

`BaseSandbox.ls`, `read`, `edit`, and `glob` all ran inline Python
scripts that either swallowed `FileNotFoundError`/`PermissionError` with
bare `pass` (ls) or left those exceptions uncaught entirely (glob),
causing the script to crash with a traceback that the host-side parser
silently turned into empty results. `read` and `edit` masked permission
failures as `file_not_found` because `os.path.isfile()` returns `False`
(not raises) on `EACCES`.

The original issue scoped this to `ls`, but the same bug class lived in
three more ops. Fixing them together avoids leaving a half-consistent
surface.

### What changed

**Inline scripts** (`sandbox.py`):
- `_LS`: emits `path_not_found` / `not_a_directory` /
`permission_denied` codes
- `_GLOB_COMMAND_TEMPLATE`: wraps body in try/except; same code
vocabulary
- `_READ_COMMAND_TEMPLATE`: replaces `os.path.isfile()` gate with
`os.stat()` + `S_ISREG`; wraps body in try/except for
`FileNotFoundError` / `PermissionError`; new `not_a_file` code
distinguishes directories/sockets/etc. from missing files
- `_EDIT_COMMAND_TEMPLATE` / `_EDIT_TMPFILE_TEMPLATE`: same refactor on
target file

**Host-side parsers**:
- `BaseSandbox.ls` / `BaseSandbox.glob`: detect error JSON line, wrap as
`f"Path '{path}': {code}"`, return `entries`/`matches=None` to match the
documented `LsResult` / `GlobResult` contracts
- `BaseSandbox._map_edit_error`: refactored to a lookup dict; learns
`permission_denied` and `not_a_file`
- `read` / `edit` host parsers unchanged — they already wrap arbitrary
codes with `f"File '{file_path}': {code}"`

### Convention

Adopts the existing pattern from `read()` / `edit()` consistently
everywhere:
1. Inline script emits short snake_case codes (no path interpolation)
2. Host wraps with the caller-supplied path prefix
This commit is contained in:
Nick Hollon
2026-05-12 12:52:01 -07:00
committed by GitHub
parent 400d40049b
commit 7598bd93f7
3 changed files with 588 additions and 187 deletions
+217 -182
View File
@@ -51,17 +51,26 @@ import base64
path = base64.b64decode('{path_b64}').decode('utf-8')
pattern = base64.b64decode('{pattern_b64}').decode('utf-8')
os.chdir(path)
matches = sorted(glob.glob(pattern, recursive=True))
for m in matches:
stat = os.stat(m)
result = {{
'path': m,
'size': stat.st_size,
'mtime': stat.st_mtime,
'is_dir': os.path.isdir(m)
}}
print(json.dumps(result))
try:
os.chdir(path)
matches = sorted(glob.glob(pattern, recursive=True))
for m in matches:
try:
st = os.stat(m)
except OSError:
continue
print(json.dumps({{
'path': m,
'size': st.st_size,
'mtime': st.st_mtime,
'is_dir': os.path.isdir(m),
}}))
except FileNotFoundError:
print(json.dumps({{'error': 'path_not_found'}}))
except NotADirectoryError:
print(json.dumps({{'error': 'not_a_directory'}}))
except PermissionError:
print(json.dumps({{'error': 'permission_denied'}}))
" 2>&1"""
"""Find files matching a pattern with metadata.
@@ -109,54 +118,60 @@ TRUNCATION_MSG: Final = (
"""Sentinel appended to `read()` content when `MAX_OUTPUT_BYTES` is hit."""
_EDIT_COMMAND_TEMPLATE = """python3 -c "
import sys, os, base64, json
import sys, os, stat as _stat, base64, json
payload = json.loads(base64.b64decode(sys.stdin.read().strip()).decode('utf-8'))
path, old, new = payload['path'], payload['old'], payload['new']
replace_all = payload.get('replace_all', False)
if not os.path.isfile(path):
print(json.dumps({{'error': 'file_not_found'}}))
sys.exit(0)
with open(path, 'rb') as f:
raw = f.read()
try:
text = raw.decode('utf-8')
except UnicodeDecodeError:
print(json.dumps({{'error': 'not_a_text_file'}}))
sys.exit(0)
st = os.stat(path)
if not _stat.S_ISREG(st.st_mode):
print(json.dumps({{'error': 'not_a_file'}}))
sys.exit(0)
# Match-driven CRLF handling (issue #2880): the read template normalizes
# CRLF to LF for the LLM, so old_string arrives LF-only even when the
# file on disk is CRLF. Try old as sent, then a CRLF variant, then an LF
# variant. The first match reveals the file line-ending style in that
# region; apply the same transform to new so the file style is preserved.
old_crlf = old.replace('\\r\\n', '\\n').replace('\\n', '\\r\\n')
old_lf = old.replace('\\r\\n', '\\n')
new_crlf = new.replace('\\r\\n', '\\n').replace('\\n', '\\r\\n')
new_lf = new.replace('\\r\\n', '\\n')
count = 0
matched_old, matched_new = old, new
for cand_old, cand_new in ((old, new), (old_crlf, new_crlf), (old_lf, new_lf)):
c = text.count(cand_old)
if c >= 1:
matched_old, matched_new, count = cand_old, cand_new, c
break
with open(path, 'rb') as f:
raw = f.read()
if count == 0:
print(json.dumps({{'error': 'string_not_found'}}))
sys.exit(0)
if count > 1 and not replace_all:
print(json.dumps({{'error': 'multiple_occurrences', 'count': count}}))
sys.exit(0)
try:
text = raw.decode('utf-8')
except UnicodeDecodeError:
print(json.dumps({{'error': 'not_a_text_file'}}))
sys.exit(0)
result = text.replace(matched_old, matched_new) if replace_all else text.replace(matched_old, matched_new, 1)
with open(path, 'wb') as f:
f.write(result.encode('utf-8'))
# Match-driven CRLF handling (issue #2880): the read template normalizes
# CRLF to LF for the LLM, so old_string arrives LF-only even when the
# file on disk is CRLF. Try old as sent, then a CRLF variant, then an LF
# variant. The first match reveals the file line-ending style in that
# region; apply the same transform to new so the file style is preserved.
old_crlf = old.replace('\\r\\n', '\\n').replace('\\n', '\\r\\n')
old_lf = old.replace('\\r\\n', '\\n')
new_crlf = new.replace('\\r\\n', '\\n').replace('\\n', '\\r\\n')
new_lf = new.replace('\\r\\n', '\\n')
count = 0
matched_old, matched_new = old, new
for cand_old, cand_new in ((old, new), (old_crlf, new_crlf), (old_lf, new_lf)):
c = text.count(cand_old)
if c >= 1:
matched_old, matched_new, count = cand_old, cand_new, c
break
print(json.dumps({{'count': count}}))
if count == 0:
print(json.dumps({{'error': 'string_not_found'}}))
sys.exit(0)
if count > 1 and not replace_all:
print(json.dumps({{'error': 'multiple_occurrences', 'count': count}}))
sys.exit(0)
result = text.replace(matched_old, matched_new) if replace_all else text.replace(matched_old, matched_new, 1)
with open(path, 'wb') as f:
f.write(result.encode('utf-8'))
print(json.dumps({{'count': count}}))
except FileNotFoundError:
print(json.dumps({{'error': 'file_not_found'}}))
except PermissionError:
print(json.dumps({{'error': 'permission_denied'}}))
" 2>&1 <<'__DEEPAGENTS_EDIT_EOF__'
{payload_b64}
__DEEPAGENTS_EDIT_EOF__
@@ -188,7 +203,7 @@ to avoid size limits on the execute() request body imposed by some sandbox provi
"""
_EDIT_TMPFILE_TEMPLATE = """python3 -c "
import os, sys, json, base64
import os, stat as _stat, sys, json, base64
old_path = base64.b64decode('{old_path_b64}').decode('utf-8')
new_path = base64.b64decode('{new_path_b64}').decode('utf-8')
@@ -206,44 +221,50 @@ finally:
try: os.remove(p)
except OSError: pass
if not os.path.isfile(target):
print(json.dumps({{'error': 'file_not_found'}}))
sys.exit(0)
with open(target, 'rb') as f:
raw = f.read()
try:
text = raw.decode('utf-8')
except UnicodeDecodeError:
print(json.dumps({{'error': 'not_a_text_file'}}))
sys.exit(0)
st = os.stat(target)
if not _stat.S_ISREG(st.st_mode):
print(json.dumps({{'error': 'not_a_file'}}))
sys.exit(0)
# Match-driven CRLF handling -- see _EDIT_COMMAND_TEMPLATE and issue #2880.
old_crlf = old.replace('\\r\\n', '\\n').replace('\\n', '\\r\\n')
old_lf = old.replace('\\r\\n', '\\n')
new_crlf = new.replace('\\r\\n', '\\n').replace('\\n', '\\r\\n')
new_lf = new.replace('\\r\\n', '\\n')
count = 0
matched_old, matched_new = old, new
for cand_old, cand_new in ((old, new), (old_crlf, new_crlf), (old_lf, new_lf)):
c = text.count(cand_old)
if c >= 1:
matched_old, matched_new, count = cand_old, cand_new, c
break
with open(target, 'rb') as f:
raw = f.read()
if count == 0:
print(json.dumps({{'error': 'string_not_found'}}))
sys.exit(0)
if count > 1 and not replace_all:
print(json.dumps({{'error': 'multiple_occurrences', 'count': count}}))
sys.exit(0)
try:
text = raw.decode('utf-8')
except UnicodeDecodeError:
print(json.dumps({{'error': 'not_a_text_file'}}))
sys.exit(0)
result = text.replace(matched_old, matched_new) if replace_all else text.replace(matched_old, matched_new, 1)
with open(target, 'wb') as f:
f.write(result.encode('utf-8'))
# Match-driven CRLF handling -- see _EDIT_COMMAND_TEMPLATE and issue #2880.
old_crlf = old.replace('\\r\\n', '\\n').replace('\\n', '\\r\\n')
old_lf = old.replace('\\r\\n', '\\n')
new_crlf = new.replace('\\r\\n', '\\n').replace('\\n', '\\r\\n')
new_lf = new.replace('\\r\\n', '\\n')
count = 0
matched_old, matched_new = old, new
for cand_old, cand_new in ((old, new), (old_crlf, new_crlf), (old_lf, new_lf)):
c = text.count(cand_old)
if c >= 1:
matched_old, matched_new, count = cand_old, cand_new, c
break
print(json.dumps({{'count': count}}))
if count == 0:
print(json.dumps({{'error': 'string_not_found'}}))
sys.exit(0)
if count > 1 and not replace_all:
print(json.dumps({{'error': 'multiple_occurrences', 'count': count}}))
sys.exit(0)
result = text.replace(matched_old, matched_new) if replace_all else text.replace(matched_old, matched_new, 1)
with open(target, 'wb') as f:
f.write(result.encode('utf-8'))
print(json.dumps({{'count': count}}))
except FileNotFoundError:
print(json.dumps({{'error': 'file_not_found'}}))
except PermissionError:
print(json.dumps({{'error': 'permission_denied'}}))
" 2>&1"""
"""Server-side file edit via temp-file upload for large payloads.
@@ -259,7 +280,7 @@ files cannot be read.
"""
_READ_COMMAND_TEMPLATE = """python3 -c "
import codecs, os, sys, base64, json
import codecs, os, stat as _stat, sys, base64, json
MAX_OUTPUT_BYTES = 500 * 1024
MAX_BINARY_BYTES = 500 * 1024
@@ -271,87 +292,92 @@ TRUNCATION_MSG = '\\n\\n' + (
path = base64.b64decode('{path_b64}').decode('utf-8')
if not os.path.isfile(path):
print(json.dumps({{'error': 'file_not_found'}}))
sys.exit(0)
if os.path.getsize(path) == 0:
print(json.dumps({{'encoding': 'utf-8', 'content': 'System reminder: File exists but has empty contents'}}))
sys.exit(0)
file_type = '{file_type}'
if file_type != 'text':
file_size = os.path.getsize(path)
if file_size > MAX_BINARY_BYTES:
print(json.dumps({{'error': 'Binary file exceeds maximum preview size of ' + str(MAX_BINARY_BYTES) + ' bytes'}}))
sys.exit(0)
with open(path, 'rb') as f:
raw = f.read()
print(json.dumps({{'encoding': 'base64', 'content': base64.b64encode(raw).decode('ascii')}}))
sys.exit(0)
with open(path, 'rb') as f:
raw_prefix = f.read(8192)
# The 8192-byte prefix can slice a multi-byte UTF-8 char (CJK is 3 bytes,
# emoji is 4); the incremental decoder buffers a trailing partial sequence
# instead of raising, so legitimate text isn't misclassified as binary.
is_binary = False
try:
codecs.getincrementaldecoder('utf-8')().decode(raw_prefix, final=False)
except UnicodeDecodeError:
is_binary = True
st = os.stat(path)
if not _stat.S_ISREG(st.st_mode):
print(json.dumps({{'error': 'not_a_file'}}))
sys.exit(0)
if st.st_size == 0:
print(json.dumps({{'encoding': 'utf-8', 'content': 'System reminder: File exists but has empty contents'}}))
sys.exit(0)
file_type = '{file_type}'
if file_type != 'text':
if st.st_size > MAX_BINARY_BYTES:
print(json.dumps({{'error': 'Binary file exceeds maximum preview size of ' + str(MAX_BINARY_BYTES) + ' bytes'}}))
sys.exit(0)
with open(path, 'rb') as f:
raw = f.read()
print(json.dumps({{'encoding': 'base64', 'content': base64.b64encode(raw).decode('ascii')}}))
sys.exit(0)
if is_binary:
with open(path, 'rb') as f:
raw = f.read()
print(json.dumps({{'encoding': 'base64', 'content': base64.b64encode(raw).decode('ascii')}}))
sys.exit(0)
raw_prefix = f.read(8192)
offset = {offset}
limit = {limit}
line_count = 0
returned_lines = 0
truncated = False
parts = []
current_bytes = 0
msg_bytes = len(TRUNCATION_MSG.encode('utf-8'))
effective_limit = MAX_OUTPUT_BYTES - msg_bytes
# The 8192-byte prefix can slice a multi-byte UTF-8 char (CJK is 3 bytes,
# emoji is 4); the incremental decoder buffers a trailing partial sequence
# instead of raising, so legitimate text isn't misclassified as binary.
is_binary = False
try:
codecs.getincrementaldecoder('utf-8')().decode(raw_prefix, final=False)
except UnicodeDecodeError:
is_binary = True
with open(path, 'r', encoding='utf-8', newline=None) as f:
for raw_line in f:
line_count += 1
if line_count <= offset:
continue
if returned_lines >= limit:
break
if is_binary:
with open(path, 'rb') as f:
raw = f.read()
print(json.dumps({{'encoding': 'base64', 'content': base64.b64encode(raw).decode('ascii')}}))
sys.exit(0)
line = raw_line.rstrip('\\n').rstrip('\\r')
piece = line if returned_lines == 0 else '\\n' + line
piece_bytes = len(piece.encode('utf-8'))
if current_bytes + piece_bytes > effective_limit:
truncated = True
remaining_bytes = effective_limit - current_bytes
if remaining_bytes > 0:
prefix = piece.encode('utf-8')[:remaining_bytes].decode('utf-8', errors='ignore')
if prefix:
parts.append(prefix)
current_bytes += len(prefix.encode('utf-8'))
break
offset = {offset}
limit = {limit}
line_count = 0
returned_lines = 0
truncated = False
parts = []
current_bytes = 0
msg_bytes = len(TRUNCATION_MSG.encode('utf-8'))
effective_limit = MAX_OUTPUT_BYTES - msg_bytes
parts.append(piece)
current_bytes += piece_bytes
returned_lines += 1
with open(path, 'r', encoding='utf-8', newline=None) as f:
for raw_line in f:
line_count += 1
if line_count <= offset:
continue
if returned_lines >= limit:
break
if returned_lines == 0 and not truncated:
print(json.dumps({{'error': 'Line offset ' + str(offset) + ' exceeds file length (' + str(line_count) + ' lines)'}}))
sys.exit(0)
line = raw_line.rstrip('\\n').rstrip('\\r')
piece = line if returned_lines == 0 else '\\n' + line
piece_bytes = len(piece.encode('utf-8'))
if current_bytes + piece_bytes > effective_limit:
truncated = True
remaining_bytes = effective_limit - current_bytes
if remaining_bytes > 0:
prefix = piece.encode('utf-8')[:remaining_bytes].decode('utf-8', errors='ignore')
if prefix:
parts.append(prefix)
current_bytes += len(prefix.encode('utf-8'))
break
text = ''.join(parts)
if truncated:
text += TRUNCATION_MSG
parts.append(piece)
current_bytes += piece_bytes
returned_lines += 1
print(json.dumps({{'encoding': 'utf-8', 'content': text}}))
if returned_lines == 0 and not truncated:
print(json.dumps({{'error': 'Line offset ' + str(offset) + ' exceeds file length (' + str(line_count) + ' lines)'}}))
sys.exit(0)
text = ''.join(parts)
if truncated:
text += TRUNCATION_MSG
print(json.dumps({{'encoding': 'utf-8', 'content': text}}))
except FileNotFoundError:
print(json.dumps({{'error': 'file_not_found'}}))
except PermissionError:
print(json.dumps({{'error': 'permission_denied'}}))
" 2>&1"""
"""Read file content with server-side pagination.
@@ -425,23 +451,31 @@ try:
}}
print(json.dumps(result))
except FileNotFoundError:
pass
print(json.dumps({{'error': 'path_not_found'}}))
except NotADirectoryError:
print(json.dumps({{'error': 'not_a_directory'}}))
except PermissionError:
pass
print(json.dumps({{'error': 'permission_denied'}}))
" 2>/dev/null"""
result = self.execute(cmd)
file_infos: list[FileInfo] = []
error: str | None = None
for line in result.output.strip().split("\n"):
if not line:
continue
try:
data = json.loads(line)
file_infos.append({"path": data["path"], "is_dir": data["is_dir"]})
except json.JSONDecodeError:
continue
if isinstance(data, dict) and "error" in data:
error = data["error"]
continue
file_infos.append({"path": data["path"], "is_dir": data["is_dir"]})
if error is not None:
return LsResult(entries=None, error=f"Path '{path}': {error}")
return LsResult(entries=file_infos)
def read(
@@ -712,23 +746,15 @@ except PermissionError:
@staticmethod
def _map_edit_error(error: str, file_path: str, old_string: str) -> EditResult:
"""Map server-side error codes to `EditResult` objects."""
if error == "file_not_found":
return EditResult(
error=f"Error: File '{file_path}' not found",
)
if error == "not_a_text_file":
return EditResult(
error=f"Error: File '{file_path}' is not a text file",
)
if error == "string_not_found":
return EditResult(
error=f"Error: String not found in file: '{old_string}'",
)
if error == "multiple_occurrences":
return EditResult(
error=f"Error: String '{old_string}' appears multiple times. Use replace_all=True to replace all occurrences.",
)
return EditResult(error=f"Error editing file '{file_path}': {error}")
messages: dict[str, str] = {
"file_not_found": f"Error: File '{file_path}' not found",
"permission_denied": f"Error: Permission denied editing file '{file_path}'",
"not_a_file": f"Error: '{file_path}' is not a regular file",
"not_a_text_file": f"Error: File '{file_path}' is not a text file",
"string_not_found": f"Error: String not found in file: '{old_string}'",
"multiple_occurrences": (f"Error: String '{old_string}' appears multiple times. Use replace_all=True to replace all occurrences."),
}
return EditResult(error=messages.get(error, f"Error editing file '{file_path}': {error}"))
def grep(
self,
@@ -798,20 +824,29 @@ except PermissionError:
if not output:
return GlobResult(matches=[])
# Parse JSON output into FileInfo dicts
# Parse JSON output into FileInfo dicts. Any error record (emitted when
# the search path itself is unreachable) wins over partial matches —
# mirrors read()/ls() convention: sandbox emits a short code, host wraps
# it with the search-path prefix.
file_infos: list[FileInfo] = []
error: str | None = None
for line in output.split("\n"):
try:
data = json.loads(line)
file_infos.append(
{
"path": data["path"],
"is_dir": data["is_dir"],
}
)
except json.JSONDecodeError:
continue
if isinstance(data, dict) and "error" in data:
error = data["error"]
continue
file_infos.append(
{
"path": data["path"],
"is_dir": data["is_dir"],
}
)
if error is not None:
return GlobResult(matches=None, error=f"Path '{path}': {error}")
return GlobResult(matches=file_infos)
@property
@@ -8,7 +8,9 @@ correctly.
import base64
import json
import os
import re
import stat
import subprocess
import sys
from pathlib import Path
@@ -267,6 +269,102 @@ def test_read_allows_truncated_paginated_output() -> None:
}
# -- ls tests -----------------------------------------------------------------
def test_ls_returns_entries_for_directory_with_files() -> None:
"""ls() parses one JSON object per line into FileInfo entries."""
sandbox = MockSandbox()
sandbox._next_output = json.dumps({"path": "/test/a.txt", "is_dir": False}) + "\n" + json.dumps({"path": "/test/sub", "is_dir": True})
result = sandbox.ls("/test")
assert result.error is None
assert result.entries == [
{"path": "/test/a.txt", "is_dir": False},
{"path": "/test/sub", "is_dir": True},
]
def test_ls_empty_directory_returns_empty_entries_no_error() -> None:
"""A genuinely empty directory yields entries=[] with error=None."""
sandbox = MockSandbox()
sandbox._next_output = ""
result = sandbox.ls("/test/empty")
assert result.error is None
assert result.entries == []
def test_ls_nonexistent_path_sets_error() -> None:
"""When the inline script reports a missing path, ls() surfaces it on .error.
Mirrors read()/edit(): the sandbox emits a short snake_case code; the host
wraps it with the path prefix.
"""
sandbox = MockSandbox()
sandbox._next_output = json.dumps({"error": "path_not_found"})
result = sandbox.ls("/test/does_not_exist")
assert result.entries is None
assert result.error == "Path '/test/does_not_exist': path_not_found"
def test_ls_permission_denied_sets_error() -> None:
"""When the inline script reports permission denied, ls() surfaces it on .error."""
sandbox = MockSandbox()
sandbox._next_output = json.dumps({"error": "permission_denied"})
result = sandbox.ls("/test/locked")
assert result.entries is None
assert result.error == "Path '/test/locked': permission_denied"
def test_ls_not_a_directory_sets_error() -> None:
"""When the inline script reports the path is a file, ls() surfaces it on .error."""
sandbox = MockSandbox()
sandbox._next_output = json.dumps({"error": "not_a_directory"})
result = sandbox.ls("/test/file.txt")
assert result.entries is None
assert result.error == "Path '/test/file.txt': not_a_directory"
def test_ls_error_line_amongst_entries_takes_precedence() -> None:
"""If any line is an error record, entries=None and error is reported.
A top-level scandir failure raises before any entries are emitted, but
`entry.is_dir()` can raise per-entry after some entries have already been
printed (e.g., a child becomes unreadable mid-scan). The parser is
defensive: any error line wins over partial entries, preserving the
documented contract (entries=None on failure).
"""
sandbox = MockSandbox()
sandbox._next_output = json.dumps({"path": "/test/a.txt", "is_dir": False}) + "\n" + json.dumps({"error": "permission_denied"})
result = sandbox.ls("/test")
assert result.entries is None
assert result.error == "Path '/test': permission_denied"
def test_ls_command_base64_encodes_path() -> None:
"""The path is base64-encoded into the inline script to prevent injection."""
sandbox = MockSandbox()
sandbox._next_output = ""
sandbox.ls("/test/dir")
assert sandbox.last_command is not None
expected_b64 = base64.b64encode(b"/test/dir").decode("ascii")
assert expected_b64 in sandbox.last_command
assert "python3 -c" in sandbox.last_command
# -- write tests --------------------------------------------------------------
@@ -912,3 +1010,185 @@ def test_read_script_ascii_larger_than_prefix(tmp_path: Path) -> None:
result = _run_read_script(target)
assert result["encoding"] == "utf-8"
# -- script-level permission/error tests --------------------------------------
# Direct subprocess runs of read/edit/glob inline scripts on the local FS,
# exercising the OSError-handling branches that mock-bypassed tests cannot
# reach.
_PERMISSION_DENIED_SKIP = pytest.mark.skipif(
sys.platform == "win32" or (hasattr(os, "geteuid") and os.geteuid() == 0),
reason="chmod 000 does not deny access on Windows or as root",
)
def _run_edit_script(
path: Path,
old: str,
new: str,
replace_all: bool = False, # noqa: FBT001, FBT002
) -> dict:
payload = json.dumps({"path": str(path), "old": old, "new": new, "replace_all": replace_all})
payload_b64 = base64.b64encode(payload.encode("utf-8")).decode("ascii")
cmd = _EDIT_COMMAND_TEMPLATE.format(payload_b64=payload_b64)
_, _, tail = cmd.partition('python3 -c "')
script, _, _ = tail.partition('" 2>&1')
proc = subprocess.run( # noqa: S603
[sys.executable, "-c", script],
input=payload_b64,
capture_output=True,
text=True,
check=False,
)
return json.loads(proc.stdout.strip())
def _run_glob_script(path: Path, pattern: str) -> str:
cmd = _GLOB_COMMAND_TEMPLATE.format(
path_b64=base64.b64encode(str(path).encode("utf-8")).decode("ascii"),
pattern_b64=base64.b64encode(pattern.encode("utf-8")).decode("ascii"),
)
_, _, tail = cmd.partition('python3 -c "')
script, _, _ = tail.partition('" 2>&1')
proc = subprocess.run( # noqa: S603
[sys.executable, "-c", script],
capture_output=True,
text=True,
check=False,
)
return proc.stdout
@_PERMISSION_DENIED_SKIP
def test_read_script_permission_denied(tmp_path: Path) -> None:
"""Read script must surface permission_denied, not crash with a traceback."""
target = tmp_path / "locked.txt"
target.write_text("secret")
target.chmod(0o000)
try:
result = _run_read_script(target)
assert result == {"error": "permission_denied"}
finally:
target.chmod(stat.S_IRUSR | stat.S_IWUSR)
def test_read_script_is_a_directory(tmp_path: Path) -> None:
"""Read script must surface a structured error when path is a directory."""
result = _run_read_script(tmp_path)
assert result.get("error") == "not_a_file"
@_PERMISSION_DENIED_SKIP
def test_edit_script_permission_denied(tmp_path: Path) -> None:
"""Edit script must surface permission_denied, not crash with a traceback."""
target = tmp_path / "locked.txt"
target.write_text("old content")
target.chmod(0o000)
try:
result = _run_edit_script(target, "old", "new")
assert result == {"error": "permission_denied"}
finally:
target.chmod(stat.S_IRUSR | stat.S_IWUSR)
def test_glob_script_path_not_found(tmp_path: Path) -> None:
"""Glob script must surface path_not_found instead of crashing on chdir."""
missing = tmp_path / "does_not_exist"
output = _run_glob_script(missing, "*.py")
data = json.loads(output.strip().split("\n")[0])
assert data == {"error": "path_not_found"}
@_PERMISSION_DENIED_SKIP
def test_glob_script_permission_denied(tmp_path: Path) -> None:
"""Glob script must surface permission_denied instead of crashing on chdir."""
locked = tmp_path / "locked_dir"
locked.mkdir()
locked.chmod(0o000)
try:
output = _run_glob_script(locked, "*.py")
data = json.loads(output.strip().split("\n")[0])
assert data == {"error": "permission_denied"}
finally:
locked.chmod(stat.S_IRWXU)
# -- glob host-side error surfacing -------------------------------------------
def test_glob_surfaces_error_from_script() -> None:
"""When the inline script emits an error JSON line, GlobResult.error is set.
Mirrors read()/ls() convention: sandbox emits a short code; host wraps
with the path prefix and reports entries=None.
"""
sandbox = MockSandbox()
sandbox._next_output = json.dumps({"error": "permission_denied"})
result = sandbox.glob("*.py", path="/locked")
assert result.matches is None
assert result.error == "Path '/locked': permission_denied"
def test_glob_path_not_found_sets_error() -> None:
"""Glob path_not_found code is wrapped with the search path."""
sandbox = MockSandbox()
sandbox._next_output = json.dumps({"error": "path_not_found"})
result = sandbox.glob("*.py", path="/missing")
assert result.matches is None
assert result.error == "Path '/missing': path_not_found"
def test_glob_empty_returns_empty_matches() -> None:
"""Empty stdout still means a successful, empty search."""
sandbox = MockSandbox()
sandbox._next_output = ""
result = sandbox.glob("*.py", path="/some/dir")
assert result.error is None
assert result.matches == []
# -- _map_edit_error coverage for new codes -----------------------------------
def test_map_edit_error_permission_denied() -> None:
"""_map_edit_error returns a readable message for permission_denied."""
result = BaseSandbox._map_edit_error("permission_denied", "/test/file.txt", "old")
assert result.error is not None
assert "permission" in result.error.lower()
assert "/test/file.txt" in result.error
# -- read host-side error wrapping for new codes -----------------------------
def test_read_permission_denied_surfaces_error() -> None:
"""read() wraps permission_denied from the inline script onto ReadResult.error."""
sandbox = MockSandbox()
sandbox._next_output = json.dumps({"error": "permission_denied"})
result = sandbox.read("/test/locked.txt")
assert result.file_data is None
assert result.error is not None
assert "permission_denied" in result.error
assert "/test/locked.txt" in result.error
def test_sandbox_edit_inline_permission_denied() -> None:
"""edit() (inline) wraps permission_denied from the inline script."""
sandbox = MockSandbox()
sandbox._next_output = json.dumps({"error": "permission_denied"})
result = sandbox.edit("/test/locked.txt", "old", "new")
assert result.error is not None
assert "permission" in result.error.lower()
assert "/test/locked.txt" in result.error
@@ -20,6 +20,7 @@ Linting exceptions:
import os
import re
import stat
import subprocess
from collections.abc import Iterator
from pathlib import Path
@@ -140,6 +141,8 @@ class LocalSubprocessSandbox(BaseSandbox):
if result.entries is not None:
for entry in result.entries:
entry["path"] = self._to_virtual_path(entry["path"])
if result.error is not None:
result.error = self._to_virtual_path(result.error)
return result
def read(self, file_path: str, offset: int = 0, limit: int = 2000) -> ReadResult:
@@ -193,7 +196,13 @@ class LocalSubprocessSandbox(BaseSandbox):
def glob(self, pattern: str, path: str = "/") -> GlobResult:
"""Run glob against mapped real paths."""
return super().glob(pattern, path=self._to_real_path(path))
result = super().glob(pattern, path=self._to_real_path(path))
if result.error is not None:
result.error = self._to_virtual_path(result.error)
if result.matches is not None:
for match in result.matches:
match["path"] = self._to_virtual_path(match["path"])
return result
@property
def id(self) -> str:
@@ -394,6 +403,30 @@ class TestLocalSandboxOperations:
assert result.error is not None
assert "not_found" in result.error.lower() or "not found" in result.error.lower()
def test_read_permission_denied(self, sandbox: LocalSubprocessSandbox) -> None:
"""Read on a chmod 000 file must surface permission_denied, not crash."""
test_path = "/tmp/test_sandbox_ops/locked_read.txt"
sandbox.write(test_path, "secret")
sandbox.execute(f"chmod 000 {test_path}")
try:
result = sandbox.read(test_path)
assert result.file_data is None
assert result.error is not None
assert "permission_denied" in result.error
finally:
sandbox.execute(f"chmod {stat.S_IRUSR | stat.S_IWUSR:o} {test_path}")
def test_read_directory_path(self, sandbox: LocalSubprocessSandbox) -> None:
"""Read on a directory must surface not_a_file, not crash."""
base_dir = "/tmp/test_sandbox_ops/read_dir_target"
sandbox.execute(f"mkdir -p {base_dir}")
result = sandbox.read(base_dir)
assert result.file_data is None
assert result.error is not None
assert "not_a_file" in result.error
def test_read_empty_file(self, sandbox: LocalSubprocessSandbox) -> None:
"""Test reading an empty file."""
test_path = "/tmp/test_sandbox_ops/empty_read.txt"
@@ -624,6 +657,18 @@ class TestLocalSandboxOperations:
assert result.error is not None
assert "not_found" in result.error.lower() or "not found" in result.error.lower()
def test_edit_permission_denied(self, sandbox: LocalSubprocessSandbox) -> None:
"""Edit on a chmod 000 file must surface permission_denied, not crash."""
test_path = "/tmp/test_sandbox_ops/locked_edit.txt"
sandbox.write(test_path, "Hello world")
sandbox.execute(f"chmod 000 {test_path}")
try:
result = sandbox.edit(test_path, "Hello", "Goodbye")
assert result.error is not None
assert "permission" in result.error.lower()
finally:
sandbox.execute(f"chmod {stat.S_IRUSR | stat.S_IWUSR:o} {test_path}")
def test_edit_special_characters(self, sandbox: LocalSubprocessSandbox) -> None:
"""Test editing with special characters and regex metacharacters."""
test_path = "/tmp/test_sandbox_ops/edit_special.txt"
@@ -1012,12 +1057,24 @@ class TestLocalSandboxOperations:
assert result.entries == []
def test_ls_info_nonexistent_directory(self, sandbox: LocalSubprocessSandbox) -> None:
"""Test listing a directory that doesn't exist."""
"""Ls on a missing path must surface the failure on .error, not return []."""
nonexistent_dir = "/tmp/test_sandbox_ops/does_not_exist"
result = sandbox.ls(nonexistent_dir)
assert result.entries == []
assert result.entries is None
assert result.error == f"Path '{nonexistent_dir}': path_not_found"
def test_ls_info_permission_denied(self, sandbox: LocalSubprocessSandbox) -> None:
"""Ls on an unreadable directory must surface the failure on .error."""
base_dir = "/tmp/test_sandbox_ops/ls_locked"
sandbox.execute(f"mkdir -p {base_dir} && chmod 000 {base_dir}")
try:
result = sandbox.ls(base_dir)
assert result.entries is None
assert result.error == f"Path '{base_dir}': permission_denied"
finally:
sandbox.execute(f"chmod {stat.S_IRWXU:o} {base_dir}")
def test_ls_info_hidden_files(self, sandbox: LocalSubprocessSandbox) -> None:
"""Test that ls_info includes hidden files (starting with .)."""
@@ -1107,10 +1164,17 @@ class TestLocalSandboxOperations:
assert f"{base_dir}/file-3.txt" in paths
def test_ls_info_path_is_sanitized(self, sandbox: LocalSubprocessSandbox) -> None:
"""Test that ls_info base64-encodes paths to prevent injection."""
"""Test that ls base64-encodes paths to prevent injection.
The malicious path is treated as a literal (non-existent) path on the
sandbox side, which must surface as a structured error rather than
executing any injected code.
"""
malicious_path = "'; import os; os.system('echo INJECTED'); #"
result = sandbox.ls(malicious_path)
assert result.entries == []
assert result.entries is None
assert result.error is not None
assert "path_not_found" in result.error
def test_read_path_is_sanitized(self, sandbox: LocalSubprocessSandbox) -> None:
"""Test that read does not execute injected code in the path.
@@ -1460,6 +1524,28 @@ class TestLocalSandboxOperations:
# Should work with explicit path
assert result.matches is not None
def test_glob_path_not_found(self, sandbox: LocalSubprocessSandbox) -> None:
"""Glob on a missing search path must surface path_not_found, not crash."""
missing = "/tmp/test_sandbox_ops/glob_missing_root"
result = sandbox.glob("*.py", path=missing)
assert result.matches is None
assert result.error is not None
assert "path_not_found" in result.error
def test_glob_permission_denied(self, sandbox: LocalSubprocessSandbox) -> None:
"""Glob on an unreadable search path must surface permission_denied."""
locked_dir = "/tmp/test_sandbox_ops/glob_locked"
sandbox.execute(f"mkdir -p {locked_dir} && chmod 000 {locked_dir}")
try:
result = sandbox.glob("*.py", path=locked_dir)
assert result.matches is None
assert result.error is not None
assert "permission_denied" in result.error
finally:
sandbox.execute(f"chmod {stat.S_IRWXU:o} {locked_dir}")
# ==================== Integration tests ====================
def test_write_read_edit_workflow(self, sandbox: LocalSubprocessSandbox) -> None: