Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 21 additions & 5 deletions libs/deepagents/deepagents/backends/sandbox.py
Original file line number Diff line number Diff line change
Expand Up @@ -1138,24 +1138,40 @@ async def _aedit_via_upload(
return EditResult(path=file_path, occurrences=data.get("count", 1))

def delete(self, file_path: str) -> DeleteResult:
"""Delete a file or directory from the sandbox via a server-side ``rm``.
"""Delete a file or directory from the sandbox via a server-side `rm`.

Uses ``rm -rf``, so directories are removed recursively along with their
contents, and deleting a path that does not exist succeeds silently. A
non-zero exit (e.g. a permission error) is reported as a failure.
Runs `test -e || test -L` first: a path that does not exist (and is not
a broken symlink) returns a not-found error, matching the contract of
`FilesystemBackend` and `StateBackend`. Because a shell `test` has no
error channel, a non-zero probe conflates "absent" with "unstattable"
(e.g. an unsearchable parent directory); an unknown exit code is not
treated as absent and falls through to the delete.

Uses `rm -rf`, so directories are removed recursively along with their
contents. A recursive delete may remove some entries before failing
partway; a non-zero `rm` exit (e.g. a permission error) is reported as
a failure.

Args:
file_path: Absolute path to the file or directory to delete.

Returns:
`DeleteResult` with the deleted path on success, or an error if the
deletion command fails.
path does not exist or the deletion command fails.
"""
# `shlex.quote` only neutralizes shell metacharacters so the path is
# passed to `rm` as a single literal argument. It is NOT a security
# boundary: it does not confine the deletion to any sandbox root or
# block traversal. Whatever the sandbox shell can reach, this can delete.
quoted = shlex.quote(file_path)
exists = self.execute(f"test -e {quoted} || test -L {quoted}")
# `exit_code` may be None when the backend cannot determine a status;
# only a definite non-zero means the path is absent. Treating None as
# not-found would fabricate a diagnosis and skip the delete, so fall
# through to `rm` on an unknown probe result (matches the `rm` check
# below and `_parse_grep_output`, which both guard `is not None`).
if exists.exit_code is not None and exists.exit_code != 0:
return DeleteResult(error=f"Error: '{file_path}' not found")
result = self.execute(f"rm -rf {quoted}")

if result.exit_code == 0:
Expand Down
78 changes: 74 additions & 4 deletions libs/deepagents/deepagents/middleware/filesystem.py
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,7 @@
)
from deepagents.backends.sandbox import BaseSandbox
from deepagents.backends.utils import (
_GLOB_WILDCARD_CHARS,
_VIDEO_EXTRA_EXTENSIONS,
MAX_VIDEO_INPUT_BYTES,
FileType,
Expand Down Expand Up @@ -296,12 +297,72 @@ def _check_fs_permission(
return "allow"


def _wildcard_delete_overlap(pattern: str, anchor: str, target: str) -> bool:
"""Check whether a wildcard deny pattern overlaps a recursive delete target.

Args:
pattern: The original glob pattern (e.g. ``/work/*.log``).
anchor: The longest wildcard-free prefix of ``pattern``.
target: The absolute path being recursively deleted.

Returns:
True if the pattern's matches intersect the delete subtree.
"""
# Root anchor ("/**/x"): pattern can match anywhere, block all.
if anchor == "/":
return True
# Target directly matches the glob: block.
if wcglob.globmatch(target, pattern, flags=_FS_WCMATCH_FLAGS):
return True
# Anchor is inside the delete subtree: recursive delete would remove
# matching descendants — block.
if PurePosixPath(anchor).is_relative_to(PurePosixPath(target)):
return True
# Target is below the anchor: safe to allow ONLY when the pattern suffix
# is a single, non-** component (fixed depth) AND no ancestor of the
# target matches the glob. "/work/*.log" can never match anything under
# "/work/notes.txt". But "/work/*" matches "/work/app", so deleting
# "/work/app/child" mutates a denied path's contents and must be blocked.
# Patterns with directory wildcards ("/work/*/secrets") could match
# descendants of the target, so fail closed for those.
if not PurePosixPath(target).is_relative_to(PurePosixPath(anchor)):
return False
anchor_parts = PurePosixPath(anchor).parts
pattern_parts = PurePosixPath(pattern).parts
suffix = pattern_parts[len(anchor_parts) :]
if len(suffix) != 1 or "**" in suffix[0]:
return True
# Check whether any ancestor of the target (between anchor and target)
# matches the glob. If so, the target is inside a denied directory's
# subtree.
target_parts = PurePosixPath(target).parts
return any(
wcglob.globmatch(
str(PurePosixPath(*target_parts[:depth])),
pattern,
flags=_FS_WCMATCH_FLAGS,
)
for depth in range(len(anchor_parts), len(target_parts))
)


def _find_delete_deny_patterns(rules: list[FilesystemPermission], target: str) -> list[str]:
"""Return deny-write patterns that block deleting `target`.

A recursive delete removes `target` and all descendants, so any overlapping
deny-write pattern prevents the operation. The check is based only on
permission rules and returns all matching patterns.
A recursive delete removes `target` and all descendants, so a deny-write
pattern blocks the operation when it could match `target` or anything in
its subtree. Sibling file globs that cannot match anything inside the
deleted subtree (e.g. deny `/work/*.log` when deleting `/work/notes.txt`)
do not block. The check is based only on permission rules and returns all
matching patterns.

Literal (wildcard-free) deny patterns use a subtree-overlap check: a deny
on a directory blocks deleting anything inside it and blocks deleting an
ancestor that contains it. Wildcard patterns are handled by
`_wildcard_delete_overlap`, which also blocks when the glob matches an
ancestor of `target` (deleting `/work/app/child` under a deny on `/work/*`
mutates the denied `/work/app`), while still allowing siblings that can
never contain a match (deny `/work/*.log` vs `/work/notes.txt`).

Args:
rules: Filesystem permission rules.
Expand All @@ -316,7 +377,16 @@ def _find_delete_deny_patterns(rules: list[FilesystemPermission], target: str) -
if rule.mode != "deny" or "write" not in rule.operations:
continue
for pattern in rule.paths:
if pattern not in seen and _paths_overlap(target, _glob_anchor(pattern)):
if pattern in seen:
continue
anchor = _glob_anchor(pattern)
if any(c in _GLOB_WILDCARD_CHARS for c in pattern):
overlaps = _wildcard_delete_overlap(pattern, anchor, target)
else:
# Literal pattern (no wildcards): keep the original subtree-overlap
# check so that a deny on "/work" blocks deletes of "/work/sub".
overlaps = _paths_overlap(target, anchor)
if overlaps:
seen.add(pattern)
denying.append(pattern)
return denying
Expand Down
74 changes: 58 additions & 16 deletions libs/deepagents/tests/unit_tests/backends/test_sandbox_backend.py
Original file line number Diff line number Diff line change
Expand Up @@ -42,26 +42,33 @@ class MockSandbox(BaseSandbox):

def __init__(self) -> None:
self.last_command: str | None = None
self.commands: list[str] = []
self._next_output: str = "1"
self._next_exit_code: int = 0
self._uploaded: list[tuple[str, bytes]] = []
self._file_store: dict[str, bytes] = {}
# exit_code is int | None (a backend may report an unknown status).
self._responses: list[tuple[str, int | None]] = []

@property
def id(self) -> str:
return "mock-sandbox"

def execute(self, command: str, *, timeout: int | None = None) -> ExecuteResponse:
self.last_command = command
self.commands.append(command)
# Detect temp-file upload path: upload_files() stores .deepagents_edit_*
# keys in _file_store before execute() is called.
has_tmp = any(".deepagents_edit_" in k for k in self._file_store)
if "old_path = base64.b64decode(" in command and has_tmp:
return self._simulate_edit_tmpfile(command)
output = self._next_output
exit_code = self._next_exit_code
self._next_output = "1"
self._next_exit_code = 0
if self._responses:
output, exit_code = self._responses.pop(0)
else:
output = self._next_output
exit_code = self._next_exit_code
self._next_output = "1"
self._next_exit_code = 0
return ExecuteResponse(output=output, exit_code=exit_code, truncated=False)

def _simulate_edit_tmpfile(self, command: str) -> ExecuteResponse:
Expand Down Expand Up @@ -1555,7 +1562,7 @@ def test_parse_grep_output_non_integer_line_number_is_skipped() -> None:


class TestSandboxDelete:
"""BaseSandbox.delete maps the `rm -f` exit code onto DeleteResult."""
"""BaseSandbox.delete probes existence then maps the `rm -rf` exit onto DeleteResult."""

def test_delete_success(self) -> None:
sandbox = MockSandbox()
Expand All @@ -1580,31 +1587,57 @@ def test_delete_directory_uses_recursive_rm(self) -> None:
assert "rm -rf" in sandbox.last_command
assert "/some/dir" in sandbox.last_command

def test_delete_missing_is_noop_success(self) -> None:
# `rm -f` ignores a missing path: exit 0, so delete reports success.
def test_delete_missing_returns_not_found(self) -> None:
# `test -e` exits 1 for a missing path; delete must return a not-found error.
sandbox = MockSandbox()
sandbox._next_output = ""
sandbox._next_exit_code = 0
sandbox._next_exit_code = 1 # test -e reports path absent
result = sandbox.delete("/missing.txt")
assert result.path is None
assert result.error is not None
assert "not found" in result.error

def test_delete_probe_checks_broken_symlink(self) -> None:
# The existence probe must also `test -L` so a broken symlink (where
# `test -e` fails but the link exists) is still deleted, not reported
# missing. Guards the `|| test -L` clause, which the mock's single
# exit code cannot otherwise distinguish.
sandbox = MockSandbox()
sandbox._responses = [("", 0), ("", 0)] # probe ok, rm ok
sandbox.delete("/link")
probe = sandbox.commands[0]
assert "test -e" in probe
assert "test -L" in probe

def test_delete_unknown_probe_exit_is_not_treated_as_missing(self) -> None:
# `exit_code` may be None when the backend cannot determine a status.
# An unknown probe result must NOT be reported as not-found; it falls
# through to `rm` instead of fabricating a diagnosis.
sandbox = MockSandbox()
sandbox._responses = [("", None), ("", 0)] # probe unknown, rm ok
result = sandbox.delete("/file.txt")
assert result.error is None
assert result.path == "/missing.txt"
assert result.path == "/file.txt"
assert len(sandbox.commands) == 2 # probe did not short-circuit

def test_delete_failure_reports_output(self) -> None:
# A non-zero exit (e.g. a permission error) surfaces rm's stderr.
# A non-zero exit from rm (e.g. a permission error) surfaces rm's stderr.
sandbox = MockSandbox()
sandbox._next_output = "rm: cannot remove '/some/dir': Is a directory"
sandbox._next_exit_code = 1
# Queue: test -e succeeds (file exists), then rm -rf fails with output.
sandbox._responses = [
("", 0),
("rm: cannot remove '/some/dir': Is a directory", 1),
]
result = sandbox.delete("/some/dir")
assert result.path is None
assert result.error is not None
assert "Error deleting file" in result.error
assert "Is a directory" in result.error

def test_delete_failure_unknown_error(self) -> None:
# Non-zero exit with no output falls back to a generic message.
# Non-zero exit from rm with no output falls back to a generic message.
sandbox = MockSandbox()
sandbox._next_output = ""
sandbox._next_exit_code = 1
# Queue: test -e succeeds (file exists), then rm -rf fails silently.
sandbox._responses = [("", 0), ("", 1)]
result = sandbox.delete("/file.txt")
assert result.path is None
assert result.error is not None
Expand All @@ -1617,3 +1650,12 @@ async def test_adelete_success(self) -> None:
result = await sandbox.adelete("/file.txt")
assert result.error is None
assert result.path == "/file.txt"

async def test_adelete_missing_returns_not_found(self) -> None:
# `adelete` delegates to `delete`, so the not-found contract holds async.
sandbox = MockSandbox()
sandbox._next_exit_code = 1 # test -e reports path absent
result = await sandbox.adelete("/missing.txt")
assert result.path is None
assert result.error is not None
assert "not found" in result.error
24 changes: 24 additions & 0 deletions libs/deepagents/tests/unit_tests/test_permissions.py
Original file line number Diff line number Diff line change
Expand Up @@ -253,6 +253,30 @@ class TestFindDeleteDenyPatterns:
pytest.param("/workshop/**", "/work", [], id="sibling-prefix-glob"),
pytest.param("/work2", "/work", [], id="sibling-prefix-literal"),
pytest.param("/work/secrets", "/work/logs", [], id="sibling-leaf"),
# Single-component file glob: non-matching siblings are allowed
pytest.param("/work/*.log", "/work/notes.txt", [], id="file-glob-does-not-block-non-matching-sibling"),
# Single-component wildcard: ancestor that matches the glob blocks
# delete of its descendants (wildcard-denied dirs get the same
# protection as literal-denied dirs)
pytest.param("/work/*", "/work/app/child", ["/work/*"], id="wildcard-ancestor-blocks-descendant-delete"),
pytest.param("/work/*.log", "/work/app.log/child", ["/work/*.log"], id="file-glob-ancestor-blocks-descendant-delete"),
pytest.param("/work/*", "/work/app/deep/nested", ["/work/*"], id="wildcard-ancestor-blocks-deep-descendant-delete"),
# Directory wildcard after anchor: target below anchor is blocked (fail closed)
pytest.param("/work/*/secrets", "/work/app", ["/work/*/secrets"], id="dir-wildcard-blocks-ancestor-target"),
pytest.param("/work/**/secrets", "/work/app", ["/work/**/secrets"], id="globstar-wildcard-blocks-ancestor-target"),
# Recursive glob with suffix: deleting /work/sub would remove /work/sub/a.log
pytest.param("/work/**/*.log", "/work/sub", ["/work/**/*.log"], id="recursive-glob-blocks-descendant-that-contains-match"),
# --- glob that matches the target itself -> blocked --------------
# Guards the linchpin: the "allow non-matching sibling" path is only
# safe because a target that matches the glob is blocked first. If
# this direct-match check regresses, a denied file becomes deletable.
pytest.param("/work/*.log", "/work/app.log", ["/work/*.log"], id="file-glob-blocks-matching-target"),
pytest.param("/work/*", "/work/app", ["/work/*"], id="single-wildcard-blocks-matching-child"),
# Non-`*` wildcard classes (brace, char-class, `?`) are supported via
# BRACE|GLOBSTAR flags and the `_GLOB_WILDCARD_CHARS` frozenset.
pytest.param("/work/{secrets,keys}", "/work/secrets", ["/work/{secrets,keys}"], id="brace-glob-blocks-matching"),
pytest.param("/work/[ab].txt", "/work/a.txt", ["/work/[ab].txt"], id="charclass-glob-blocks-matching"),
pytest.param("/work/f?le.txt", "/work/other.txt", [], id="question-glob-non-matching-sibling-allowed"),
# --- overlap in either direction -> blocked ----------------------
pytest.param("/work/a.txt", "/work/a.txt", ["/work/a.txt"], id="exact-file"),
pytest.param("/work", "/work", ["/work"], id="exact-dir-literal"),
Expand Down