mirror of
https://github.com/langchain-ai/deepagents.git
synced 2026-07-22 01:35:28 -04:00
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:
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user