feat(sdk): add interrupt mode to filesystem permissions (#3505)

## Example

```python
from deepagents import create_deep_agent
from deepagents.middleware.filesystem import FilesystemPermission

agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-5",
    permissions=[
        # Pause for human approval on any write under /secrets/**
        FilesystemPermission(
            operations=["write", "edit"],
            paths=["/secrets/**"],
            mode="interrupt",
        ),
    ],
)
```

When the agent calls a filesystem tool whose access matches an
interrupt-mode rule, the run pauses for approval via
`HumanInTheLoopMiddleware` instead of erroring (`mode="deny"`) or
running silently (`mode="allow"`). HITL is auto-installed whenever any
rule uses `mode="interrupt"`.

## Summary

- Adds `mode="interrupt"` to `FilesystemPermission` — matching tool
calls pause for human approval via `HumanInTheLoopMiddleware` instead of
erroring out or running silently.
- `create_deep_agent` auto-installs HITL when any interrupt-mode rule is
present and merges generated `interrupt_on` entries with the caller's
`interrupt_on` (caller wins per tool name). Wired at all three sites:
main agent, general-purpose subagent, and declarative subagents.
- Uses the `when` predicate on `InterruptOnConfig`
(langchain-ai/langchain#37579), so interrupts fire per matching
path/subtree rather than per tool name. The gate is **scope-aware** per
tool, so it fires precisely without leaking protected paths.
- The interrupt offers the approver the full decision set — `approve` /
`edit` / `reject` / `respond` — matching langchain's default for
user-supplied `interrupt_on` tools.

## Implementation

- `deepagents/middleware/filesystem.py`
  - `FilesystemPermission.mode: Literal["allow", "deny", "interrupt"]`.
- `_check_fs_permission` returns the matched rule's mode (first-match
precedence — list a `deny` before an `interrupt` if you want deny to
win).
- Result-filtering helpers (`_filter_paths_by_permission`,
`_filter_file_infos_by_permission`,
`_filter_grep_matches_by_permission`) drop only `deny` matches;
interrupt-mode paths remain in `ls`/`glob`/`grep` output because the
HITL gate fires **before** execution (the predicate), so an approved
bulk listing still returns its results.
- `deepagents/middleware/_fs_interrupt.py` (new) — the HITL glue, kept
out of `FilesystemMiddleware` since only graph assembly needs it.
`_FS_TOOL_PATH_ARGS` tags each tool with a `ToolScope`, and
`_build_interrupt_on_from_permissions(rules)` returns a `{tool_name:
InterruptOnConfig}` map with a scope-aware `when` predicate:
- **exact** (`read_file` / `write_file` / `edit_file`) — fires iff the
call's literal path matches an interrupt-mode rule.
- **bulk** (`ls` / `glob` / `grep`) — fires iff the call's search
subtree could intersect an interrupt-mode rule. A pathless call
(`grep(path=None)`) and current-directory aliases (`.`, `""`, `./`) fire
unconditionally because they can't be localized; `glob`'s `pattern` is
also gated, since an absolute pattern (or `..` traversal) redirects the
search root independently of `path`.
- `deepagents/backends/utils.py` — generic path helpers `_glob_anchor`
(longest wildcard-free prefix of a glob) and `_paths_overlap`
(component-aware subtree intersection), alongside `validate_path` /
`to_posix_path`.
- `deepagents/graph.py` — merges fs-derived configs with `interrupt_on`
and conditionally adds `HumanInTheLoopMiddleware` for the main agent, GP
subagent, and declarative subagents.

## Security hardening (from review)

Corridor flagged, and this PR closes, three HITL-bypass shapes for bulk
tools — each with a regression test:

- **Pathless `grep`** (`grep(path=None)`) — the predicate now fires
unconditionally when the call can't be localized.
- **Current-directory aliases** (`.` / `""` / `./`, which
`validate_path` normalizes to `/.`) — collapsed to `/` so the
root-overlaps-everything branch fires.
- **Absolute / `..` glob `pattern`** (`glob(pattern="/secrets/**",
path="/workspace")`) — Python's `glob` ignores the working directory for
absolute patterns, so the predicate now gates `pattern` in addition to
`path`.

## Dependency

This needs the `when` predicate / `interrupt_mode` on
`HumanInTheLoopMiddleware`, which shipped in **langchain 1.3.3**. No git
override or branch pin — the dependency uses the released package:

```toml
"langchain>=1.3.3,<2.0.0",
```

(The earlier `[tool.uv.sources]` override that pinned `langchain` to the
`sr/hitl-refactor` branch has been removed now that
langchain-ai/langchain#37579 is merged and released.)

The smoke-test snapshots track langchain's released tool descriptions
(e.g. the `write_todos` description).

## Test plan

- [x] `make test` — 1727 passed, 90 skipped, 4 xfailed
- [x] `make lint` — clean (ruff + ty)
- [x] Unit tests in `tests/unit_tests/test_permissions.py` cover:
  - The `"interrupt"` literal on the dataclass.
- `_check_fs_permission` returns `"interrupt"` on match and
`deny`-before-`interrupt` precedence.
- `_filter_paths_by_permission` keeps interrupt paths (only deny is
filtered).
- `_build_interrupt_on_from_permissions` registers only the tools whose
op could be triggered.
- The scope-aware `when` predicate: **exact** behavior, **bulk** subtree
overlap, `_glob_anchor` / `_paths_overlap` edge cases (root, equal,
ancestor, descendant, prefix lookalike, disjoint), and all three bypass
regressions above (pathless, current-dir aliases, absolute/`..` glob
patterns).
This commit is contained in:
Nick Hollon
2026-06-03 10:58:58 -04:00
committed by GitHub
parent e052e08993
commit a090162ea9
5 changed files with 502 additions and 11 deletions
@@ -388,6 +388,45 @@ def truncate_if_too_long(result: list[str] | str) -> list[str] | str:
return result
# Characters that mark a glob path component as a wildcard segment for the
# purposes of `_glob_anchor`. Keep in sync with the wcmatch flags used by the
# filesystem middleware (`BRACE | GLOBSTAR`).
_GLOB_WILDCARD_CHARS = frozenset("*?[{")
def _glob_anchor(pattern: str) -> str:
"""Return the longest leading directory of `pattern` with no wildcards.
For `/secrets/**` returns `/secrets`; for `/a/*/b` returns `/a`; for a
pattern with a wildcard at or near the root (`/**/secrets`, `/*/foo`)
falls back to `/`. The root fallback causes overlap checks to match
*any* subtree — conservative over-gating, since we cannot statically
pin down where the rule could resolve. Callers wanting precise gating
should anchor the rule's leading components.
"""
parts = PurePosixPath(to_posix_path(pattern)).parts
safe: list[str] = []
for part in parts:
if any(c in _GLOB_WILDCARD_CHARS for c in part):
break
safe.append(part)
if not safe:
return "/"
return str(PurePosixPath(*safe))
def _paths_overlap(call_path: str, rule_anchor: str) -> bool:
"""Return True if the subtree at `call_path` intersects the subtree at `rule_anchor`.
Two subtrees overlap when one is a (component-wise) prefix of the other,
or they're equal. Comparison runs on `PurePosixPath` components, so
`/secret` does not overlap `/secrets`. The root `/` overlaps everything.
"""
a = PurePosixPath(call_path)
b = PurePosixPath(rule_anchor)
return a == b or a.is_relative_to(b) or b.is_relative_to(a)
def to_posix_path(path: str) -> str:
r"""Normalize backslash separators to forward slashes for `PurePosixPath` use.
+47 -4
View File
@@ -37,6 +37,7 @@ from deepagents._tools import _apply_tool_description_overrides
from deepagents._version import __version__
from deepagents.backends import StateBackend
from deepagents.backends.protocol import BackendFactory, BackendProtocol
from deepagents.middleware._fs_interrupt import _build_interrupt_on_from_permissions
from deepagents.middleware._tool_exclusion import _ToolExclusionMiddleware
from deepagents.middleware.async_subagents import AsyncSubAgent, AsyncSubAgentMiddleware
from deepagents.middleware.filesystem import FilesystemMiddleware, FilesystemPermission
@@ -183,6 +184,24 @@ def get_default_model() -> ChatAnthropic:
return _build_default_model()
def _merge_fs_interrupt_on(
fs_interrupt_on: dict[str, InterruptOnConfig],
user_interrupt_on: dict[str, bool | InterruptOnConfig] | None,
) -> dict[str, bool | InterruptOnConfig] | None:
"""Merge fs-permission-derived configs with user-supplied `interrupt_on`.
User-supplied entries override generated ones per tool name. Returns
`None` when both inputs are empty so callers can skip installing
`HumanInTheLoopMiddleware`.
"""
if not fs_interrupt_on and not user_interrupt_on:
return None
merged: dict[str, bool | InterruptOnConfig] = {**fs_interrupt_on}
if user_interrupt_on:
merged.update(user_interrupt_on)
return merged
_REQUIRED_MIDDLEWARE: tuple[tuple[type[AgentMiddleware[Any, Any, Any]], tuple[str, ...]], ...] = (
(FilesystemMiddleware, ()),
(SubAgentMiddleware, ()),
@@ -401,6 +420,18 @@ def create_deep_agent( # noqa: C901, PLR0912, PLR0915 # Complex graph assembly
Rules are evaluated in declaration order; the first match wins.
If no rule matches, the call is allowed.
Each rule's ``mode`` can be:
- ``"allow"`` (default): the call proceeds.
- ``"deny"``: the tool returns a permission-denied error.
- ``"interrupt"``: the call pauses for human approval via
`HumanInTheLoopMiddleware`. A `HumanInTheLoopMiddleware` is
auto-installed when any interrupt-mode rule is present, and the
generated `interrupt_on` entries are merged with the
`interrupt_on` argument below (user-supplied entries win per
tool name). Requires a `langchain` version that supports the
``when`` predicate on `InterruptOnConfig`.
Subagents inherit these rules unless they specify their own
`permissions` field, which replaces the parent's rules entirely.
@@ -629,6 +660,10 @@ def create_deep_agent( # noqa: C901, PLR0912, PLR0915 # Complex graph assembly
)
subagent_interrupt_on = spec.get("interrupt_on", interrupt_on)
subagent_interrupt_on = _merge_fs_interrupt_on(
_build_interrupt_on_from_permissions(subagent_permissions or []),
subagent_interrupt_on,
)
# Inherit parent tools unless the subagent declares its own.
# Descriptions are rewritten; exclusion is handled by middleware.
@@ -702,8 +737,12 @@ def create_deep_agent( # noqa: C901, PLR0912, PLR0915 # Complex graph assembly
general_purpose_spec["system_prompt"] = gp_prompt
else:
general_purpose_spec["system_prompt"] = _apply_profile_prompt(_profile, GENERAL_PURPOSE_SUBAGENT["system_prompt"])
if interrupt_on is not None:
general_purpose_spec["interrupt_on"] = interrupt_on
gp_interrupt_on = _merge_fs_interrupt_on(
_build_interrupt_on_from_permissions(permissions or []),
interrupt_on,
)
if gp_interrupt_on is not None:
general_purpose_spec["interrupt_on"] = gp_interrupt_on
inline_subagents.insert(0, general_purpose_spec)
@@ -766,8 +805,12 @@ def create_deep_agent( # noqa: C901, PLR0912, PLR0915 # Complex graph assembly
add_cache_control=True,
)
)
if interrupt_on is not None:
deepagent_middleware.append(HumanInTheLoopMiddleware(interrupt_on=interrupt_on))
main_interrupt_on = _merge_fs_interrupt_on(
_build_interrupt_on_from_permissions(permissions or []),
interrupt_on,
)
if main_interrupt_on is not None:
deepagent_middleware.append(HumanInTheLoopMiddleware(interrupt_on=main_interrupt_on))
deepagent_middleware = _apply_excluded_middleware(
deepagent_middleware,
_profile,
@@ -0,0 +1,182 @@
"""Glue between `FilesystemPermission` rules and `HumanInTheLoopMiddleware`.
`FilesystemMiddleware` itself doesn't know about HITL — it only enforces deny
rules and filters denied results. The graph-assembly code in
`deepagents.graph` calls `_build_interrupt_on_from_permissions` to turn the
filesystem permissions into an `interrupt_on` mapping for
`HumanInTheLoopMiddleware`, using a `when` predicate that decides per call
whether the access intersects an interrupt-mode rule.
"""
from collections.abc import Callable
from pathlib import PurePosixPath
from typing import Literal
from langchain.agents.middleware import InterruptOnConfig
from langchain.tools.tool_node import ToolCallRequest
from deepagents.backends.utils import _glob_anchor, _paths_overlap, to_posix_path, validate_path
from deepagents.middleware.filesystem import FilesystemOperation, FilesystemPermission, _check_fs_permission
# Scope of a filesystem tool's path argument:
# - "exact": the call operates on exactly the named path (read_file,
# write_file, edit_file). Interrupt fires iff that path matches an
# interrupt-mode rule.
# - "bulk": the call's path argument names a search root and the call may
# surface any descendant (ls, glob, grep). Interrupt fires whenever the
# search subtree intersects an interrupt-mode rule's pattern, and — when
# the path argument is omitted (`grep(path=None)`) — fires unconditionally
# for any interrupt-mode rule, because a pathless bulk call can touch
# anything.
ToolScope = Literal["exact", "bulk"]
# Map filesystem tool name → (operation, path-arg name, scope, pattern-arg name).
# Drives `_build_interrupt_on_from_permissions` when synthesizing `when`
# predicates per tool. The optional pattern-arg name is set only for `glob`,
# whose `pattern` argument can itself redirect the search root (an absolute
# pattern ignores the call's `path`); see `_make_bulk_when_predicate`.
_FS_TOOL_PATH_ARGS: dict[str, tuple[FilesystemOperation, str, ToolScope, str | None]] = {
"ls": ("read", "path", "bulk", None),
"read_file": ("read", "file_path", "exact", None),
"write_file": ("write", "file_path", "exact", None),
"edit_file": ("write", "file_path", "exact", None),
"glob": ("read", "path", "bulk", "pattern"),
"grep": ("read", "path", "bulk", None),
}
def _make_fs_when_predicate(
rules: list[FilesystemPermission],
operation: FilesystemOperation,
path_arg_name: str,
scope: ToolScope,
pattern_arg_name: str | None = None,
) -> Callable[[ToolCallRequest], bool]:
"""Build a `when` predicate that fires on interrupt-mode rule matches.
The predicate's behavior depends on the tool's `ToolScope`:
- `"exact"`: fire iff the call's path matches an interrupt-mode rule
with normal first-match precedence. A preceding `deny` rule wins and
the interrupt does not fire — the tool returns a permission-denied
error instead.
- `"bulk"`: fire iff the call's search subtree could intersect an
interrupt-mode rule. With no path argument (e.g. `grep(path=None)`)
we cannot localize the call, so we fire unconditionally for any
interrupt-mode rule on the operation. `pattern_arg_name` (set for
`glob`) additionally gates the call's `pattern`, which can redirect
the search root independently of `path`.
"""
if scope == "exact":
return _make_exact_when_predicate(rules, operation, path_arg_name)
return _make_bulk_when_predicate(rules, operation, path_arg_name, pattern_arg_name)
def _make_exact_when_predicate(
rules: list[FilesystemPermission],
operation: FilesystemOperation,
path_arg_name: str,
) -> Callable[[ToolCallRequest], bool]:
def when(req: ToolCallRequest) -> bool:
raw_path = req.tool_call.get("args", {}).get(path_arg_name)
if not isinstance(raw_path, str):
return False
try:
normalized = validate_path(raw_path)
except ValueError:
return False
return _check_fs_permission(rules, operation, normalized) == "interrupt"
return when
def _make_bulk_when_predicate(
rules: list[FilesystemPermission],
operation: FilesystemOperation,
path_arg_name: str,
pattern_arg_name: str | None = None,
) -> Callable[[ToolCallRequest], bool]:
# Precompute interrupt-mode rule anchors for this op so the predicate is
# a single pass per call.
interrupt_anchors: list[str] = [
_glob_anchor(pattern) for rule in rules if rule.mode == "interrupt" and operation in rule.operations for pattern in rule.paths
]
def when(req: ToolCallRequest) -> bool:
if not interrupt_anchors:
return False
args = req.tool_call.get("args", {})
raw_path = args.get(path_arg_name)
if not isinstance(raw_path, str):
# A missing path (pathless bulk call) can't be localized, so fire;
# any other non-string is malformed, so don't.
return raw_path is None
try:
normalized = validate_path(raw_path)
except ValueError:
return False
# `validate_path` returns `/.` for current-directory aliases like
# `"."`, `""`, and `"./"`. Those refer to the whole accessible tree
# just like a missing path arg, so collapse to `/` so the
# root-overlaps-everything branch in `_paths_overlap` fires. Without
# this, an agent could pass `path="."` to bypass HITL.
if normalized == "/.":
normalized = "/"
if any(_paths_overlap(normalized, anchor) for anchor in interrupt_anchors):
return True
# `glob`'s `pattern` can redirect the search root away from `path`, so
# gating on `path` alone would let `glob(pattern="/secrets/**",
# path="/workspace")` bypass an interrupt rule on `/secrets/**`.
if pattern_arg_name is not None:
raw_pattern = args.get(pattern_arg_name)
if isinstance(raw_pattern, str) and _bulk_pattern_fires(raw_pattern, interrupt_anchors):
return True
return False
return when
def _bulk_pattern_fires(raw_pattern: str, interrupt_anchors: list[str]) -> bool:
"""Whether a glob `pattern` reaches an interrupt-mode subtree regardless of `path`.
An absolute pattern is matched from its own root — Python's `glob` ignores
the backend's `os.chdir(path)` — so gate on the pattern's anchor. A relative
pattern containing `..` can climb out of `path`; we cannot localize where it
lands, so treat it as firing. Absoluteness comes from the raw pattern, not
`_glob_anchor`: the anchor of a leading-wildcard relative pattern (`*.txt`)
collapses to `/`, which would otherwise look absolute.
"""
posix_pattern = to_posix_path(raw_pattern)
if posix_pattern.startswith("/"):
return any(_paths_overlap(_glob_anchor(raw_pattern), anchor) for anchor in interrupt_anchors)
return ".." in PurePosixPath(posix_pattern).parts
def _build_interrupt_on_from_permissions(
rules: list[FilesystemPermission],
) -> dict[str, InterruptOnConfig]:
"""Generate `interrupt_on` configs from interrupt-mode permissions.
Returns an entry for each filesystem tool whose operation could be triggered
by at least one interrupt-mode rule. Each entry uses a `when` predicate so
the interrupt only fires when the tool call's path argument matches an
interrupt-mode rule.
"""
if not any(r.mode == "interrupt" for r in rules):
return {}
# Offer the approver the full decision set, matching the default for
# user-supplied `interrupt_on` tools. All four are human-controlled, so the
# human stays the authorization gate: `edit`ed calls still re-enter the tool
# and hit its pre-execution deny check, and `respond` skips execution.
# Annotated so ty narrows to `list[DecisionType]` instead of `list[str]`.
allowed: list[Literal["approve", "edit", "reject", "respond"]] = ["approve", "edit", "reject", "respond"]
result: dict[str, InterruptOnConfig] = {}
for tool_name, (op, arg, scope, pattern_arg) in _FS_TOOL_PATH_ARGS.items():
if not any(r.mode == "interrupt" and op in r.operations for r in rules):
continue
result[tool_name] = InterruptOnConfig(
allowed_decisions=allowed,
when=_make_fs_when_predicate(rules, op, arg, scope, pattern_arg),
)
return result
@@ -87,7 +87,21 @@ class FilesystemPermission:
operations: list[FilesystemOperation]
paths: list[str]
mode: Literal["allow", "deny"] = "allow"
mode: Literal["allow", "deny", "interrupt"] = "allow"
"""Effect when a tool call matches this rule:
- ``"allow"`` (default): the call proceeds.
- ``"deny"``: the tool returns a permission-denied error.
- ``"interrupt"``: the call is paused for human approval via
[`HumanInTheLoopMiddleware`][langchain.agents.middleware.HumanInTheLoopMiddleware].
Best paired with patterns that have a literal leading anchor (e.g.,
``/secrets/**``, ``/projects/*/secrets/**``). Bulk tools
(``ls``/``glob``/``grep``) fire the interrupt based on whether their
search subtree could overlap the rule's anchored prefix, so a fully
unanchored pattern (``/**/secrets``) collapses to ``/`` and
conservatively over-fires for any bulk call.
"""
def __post_init__(self) -> None:
"""Validate permission path patterns."""
@@ -108,7 +122,7 @@ def _check_fs_permission(
rules: list[FilesystemPermission],
operation: FilesystemOperation,
path: str,
) -> Literal["allow", "deny"]:
) -> Literal["allow", "deny", "interrupt"]:
for rule in rules:
if operation not in rule.operations:
continue
@@ -122,9 +136,17 @@ def _filter_paths_by_permission(
operation: FilesystemOperation,
paths: list[str],
) -> list[str]:
"""Filter paths, removing only those denied by a rule.
Interrupt-mode paths pass through here: the interrupt fires at the HITL
stage *before* the tool runs (see `_build_interrupt_on_from_permissions`
and its scope-aware predicate), so by the time result-filtering runs the
user has already approved (or no rule matched). Filtering interrupt-mode
results out here would silently empty the listing the user just approved.
"""
if not rules:
return paths
return [p for p in paths if _check_fs_permission(rules, operation, p) == "allow"]
return [p for p in paths if _check_fs_permission(rules, operation, p) != "deny"]
def _all_paths_scoped_to_routes(
@@ -151,8 +173,12 @@ def _filter_file_infos_by_permission(
*,
operation: FilesystemOperation,
) -> list[FileInfo]:
"""Filter file-info entries according to filesystem permissions."""
return [fi for fi in infos if _check_fs_permission(rules, operation, fi.get("path", "")) == "allow"]
"""Filter file-info entries, removing only those denied by a rule.
See `_filter_paths_by_permission` for why interrupt-mode entries
pass through.
"""
return [fi for fi in infos if _check_fs_permission(rules, operation, fi.get("path", "")) != "deny"]
def _filter_grep_matches_by_permission(
@@ -161,8 +187,12 @@ def _filter_grep_matches_by_permission(
*,
operation: FilesystemOperation,
) -> list[GrepMatch]:
"""Filter grep matches according to filesystem permissions."""
return [m for m in matches if _check_fs_permission(rules, operation, m.get("path", "")) == "allow"]
"""Filter grep matches, removing only those denied by a rule.
See `_filter_paths_by_permission` for why interrupt-mode entries
pass through.
"""
return [m for m in matches if _check_fs_permission(rules, operation, m.get("path", "")) != "deny"]
def _format_grep_tool_result(
@@ -10,7 +10,9 @@ from langgraph.store.memory import InMemoryStore
from deepagents.backends import StateBackend, StoreBackend
from deepagents.backends.composite import CompositeBackend
from deepagents.backends.protocol import EditResult, ExecuteResponse, ReadResult, SandboxBackendProtocol, WriteResult
from deepagents.backends.utils import _glob_anchor, _paths_overlap
from deepagents.graph import create_deep_agent
from deepagents.middleware._fs_interrupt import _build_interrupt_on_from_permissions, _make_fs_when_predicate
from deepagents.middleware.filesystem import (
FilesystemMiddleware,
FilesystemPermission,
@@ -164,6 +166,201 @@ class TestFilesystemPermission:
rule = FilesystemPermission(operations=["read"], paths=["/workspace\\sub\\**"])
assert rule.paths == ["/workspace\\sub\\**"]
def test_interrupt_mode_accepted(self):
rule = FilesystemPermission(operations=["write"], paths=["/secrets/**"], mode="interrupt")
assert rule.mode == "interrupt"
class _FakeReq:
"""Stand-in for ToolCallRequest; we only read `tool_call['args']` in the predicate."""
def __init__(self, args: dict) -> None:
self.tool_call = {"args": args}
class TestCheckFsPermissionInterrupt:
@pytest.mark.parametrize(
("path", "expected"),
[
("/secrets/x.txt", "interrupt"),
("/workspace/x.txt", "allow"),
],
)
def test_interrupt_rule_resolution(self, path, expected):
rules = [FilesystemPermission(operations=["write"], paths=["/secrets/**"], mode="interrupt")]
assert _check_fs_permission(rules, "write", path) == expected
def test_deny_rule_takes_precedence_when_listed_first(self):
"""First-match wins; if deny is listed first, it beats a later interrupt rule."""
rules = [
FilesystemPermission(operations=["write"], paths=["/secrets/**"], mode="deny"),
FilesystemPermission(operations=["write"], paths=["/secrets/**"], mode="interrupt"),
]
assert _check_fs_permission(rules, "write", "/secrets/x.txt") == "deny"
def test_filter_paths_does_not_drop_interrupt_paths(self):
"""Interrupt-mode paths must remain in result-filtered lists.
The pre-execution `when` predicate is scope-aware: it fires before the
tool runs for both exact-path tools and bulk-path tools (including the
pathless and parent-path cases), so by the time result-filtering runs
the user has either approved or the call was rejected. Stripping
interrupt-mode results here would silently empty out a listing the
user just approved.
"""
rules = [FilesystemPermission(operations=["read"], paths=["/secret/**"], mode="interrupt")]
kept = _filter_paths_by_permission(rules, "read", ["/secret/a.txt", "/public/b.txt"])
assert kept == ["/secret/a.txt", "/public/b.txt"]
class TestBuildInterruptOnFromPermissions:
def test_empty_when_no_rules(self):
assert _build_interrupt_on_from_permissions([]) == {}
def test_empty_when_no_interrupt_rules(self):
rules = [FilesystemPermission(operations=["write"], paths=["/secrets/**"], mode="deny")]
assert _build_interrupt_on_from_permissions(rules) == {}
def test_registers_only_tools_whose_op_could_interrupt(self):
"""A write-only interrupt rule registers only the write-op tools."""
rule = FilesystemPermission(operations=["write"], paths=["/secrets/**"], mode="interrupt")
out = _build_interrupt_on_from_permissions([rule])
assert set(out) == {"write_file", "edit_file"}
def test_registers_read_tools_for_read_interrupt(self):
rule = FilesystemPermission(operations=["read"], paths=["/secrets/**"], mode="interrupt")
out = _build_interrupt_on_from_permissions([rule])
assert set(out) == {"ls", "read_file", "glob", "grep"}
@pytest.mark.parametrize(
("file_path", "expected"),
[
# literal match inside the rule's subtree
("/secrets/key.pem", True),
# outside the rule
("/workspace/x.txt", False),
# the parent dir itself doesn't match `/secrets/**` (** requires content after the slash)
("/secrets", False),
],
)
def test_exact_predicate(self, file_path, expected):
"""Exact-scope tools fire iff the literal path arg matches the rule."""
rule = FilesystemPermission(operations=["write"], paths=["/secrets/**"], mode="interrupt")
when = _make_fs_when_predicate([rule], "write", "file_path", "exact")
assert when(_FakeReq({"file_path": file_path})) is expected
@pytest.mark.parametrize(
("args", "expected"),
[
# pathless: can't localize → fire
({"path": None}, True),
({}, True),
# current-dir aliases collapse to root → fire
({"path": "."}, True),
({"path": ""}, True),
({"path": "./"}, True),
({"path": "/."}, True),
({"path": "/"}, True),
# ancestor of the rule's anchor → fire (listing surfaces protected children)
({"path": "/secrets"}, True),
# inside the rule's subtree → fire
({"path": "/secrets/sub"}, True),
# unrelated subtree → no fire
({"path": "/workspace"}, False),
# prefix lookalike — component-aware match, not string prefix
({"path": "/secret"}, False),
# path-validation failure short-circuits to no interrupt
({"path": "/secrets/../etc/passwd"}, False),
],
)
def test_bulk_predicate(self, args, expected):
"""Bulk-scope tools fire when the call subtree could intersect a rule.
Covers the HITL-bypass regressions for pathless calls and current-dir
aliases like ``"."``/``""``/``"./"``: `validate_path` collapses those to
``/.``, which doesn't string-prefix any anchor, so the predicate
previously returned False and an agent could call e.g. ``grep(pattern,
path=".")`` to read interrupt-protected paths with no HITL prompt.
"""
rule = FilesystemPermission(operations=["read"], paths=["/secrets/**"], mode="interrupt")
when = _make_fs_when_predicate([rule], "read", "path", "bulk")
assert when(_FakeReq(args)) is expected
@pytest.mark.parametrize(
("args", "expected"),
[
# Absolute pattern: Python `glob` ignores `path` (the backend's
# `os.chdir(path)` has no effect on an absolute pattern), so gate on
# the pattern's own anchor. A benign `path` must not suppress the
# interrupt when the pattern reaches into a protected subtree.
({"pattern": "/secrets/**", "path": "/workspace"}, True),
({"pattern": "/secrets/sub/*.txt", "path": "/workspace"}, True),
# Absolute pattern anchored at root → overlaps everything → fire.
({"pattern": "/**/key.pem", "path": "/workspace"}, True),
# Absolute pattern outside any interrupt rule → no fire.
({"pattern": "/workspace/**", "path": "/workspace"}, False),
# Relative pattern climbing out of `path` via `..` can't be
# localized statically → fire conservatively.
({"pattern": "../secrets/*", "path": "/workspace"}, True),
({"pattern": "../../etc/*", "path": "/workspace/sub"}, True),
# Benign relative pattern under a non-overlapping path → no fire.
({"pattern": "*.txt", "path": "/workspace"}, False),
# Relative pattern under the protected subtree still fires via the
# existing `path` check.
({"pattern": "*.txt", "path": "/secrets"}, True),
],
)
def test_bulk_glob_pattern_arg(self, args, expected):
"""The glob bulk predicate gates on its `pattern` arg, not just `path`.
Regression for the HITL bypass where `glob(pattern="/secrets/**",
path="/workspace")` slipped past an interrupt rule on `/secrets/**`:
the predicate saw only the benign `/workspace` path while the sandbox
backend (`os.chdir(path)` then `glob.glob(pattern)`) enumerated
`/secrets` anyway, because Python's `glob` ignores the working
directory for absolute patterns.
"""
rule = FilesystemPermission(operations=["read"], paths=["/secrets/**"], mode="interrupt")
when = _build_interrupt_on_from_permissions([rule])["glob"]["when"]
assert when(_FakeReq(args)) is expected
class TestGlobAnchorAndOverlap:
@pytest.mark.parametrize(
("pattern", "expected"),
[
("/secrets/**", "/secrets"),
("/a/*/b", "/a"),
("/secrets/key.pem", "/secrets/key.pem"),
("/*/foo", "/"),
("/**/secrets", "/"),
],
)
def test_glob_anchor(self, pattern, expected):
assert _glob_anchor(pattern) == expected
@pytest.mark.parametrize(
("a", "b", "expected"),
[
# equal
("/a/b", "/a/b", True),
# call inside rule
("/a/b/c", "/a/b", True),
# rule inside call
("/a", "/a/b", True),
# root overlaps everything, either direction
("/", "/anywhere", True),
("/anywhere", "/", True),
# component-aware: prefix lookalikes don't overlap
("/secret", "/secrets", False),
("/secrets", "/secret", False),
# disjoint
("/workspace", "/secrets", False),
],
)
def test_paths_overlap(self, a, b, expected):
assert _paths_overlap(a, b) is expected
class TestFilesystemMiddlewarePermissionInit:
def _backend(self):