diff --git a/src/agent_tools/filesystem_tools.py b/src/agent_tools/filesystem_tools.py index f2fa20c54..84b85becb 100644 --- a/src/agent_tools/filesystem_tools.py +++ b/src/agent_tools/filesystem_tools.py @@ -458,7 +458,8 @@ class GlobTool: async def execute(self, content: str, ctx: dict) -> dict: from src.tool_execution import ( _SENSITIVE_BASENAMES, - _is_sensitive_path, + _can_traverse_tool_path, + _is_denied_tool_path, _resolve_tool_path, _resolve_search_root, _truncate, @@ -507,7 +508,7 @@ class GlobTool: # .ssh/id_rsa, …) falls through to the walk, which skips it — # otherwise glob would surface secret paths that read_file / # grep already refuse to touch. - if inside and os.path.exists(cand) and not _is_sensitive_path(cand): + if inside and os.path.exists(cand) and not _is_denied_tool_path(cand): return [cand], None # Literal not at exact path — fall through to walk so # e.g. "foo.py" still matches at any depth (like rglob). @@ -517,13 +518,18 @@ class GlobTool: cap = _CODENAV_MAX_HITS * 5 try: for dp, dns, fns in os.walk(base): + if not _can_traverse_tool_path(os.path.realpath(dp)): + dns[:] = [] + continue # Prune skipped dirs before descending (unlike rglob which # descends first then filters — fatal on large node_modules). # Sensitive dirs (.ssh, .gnupg, …) are pruned too so glob # never enumerates the keys/tokens inside them. dns[:] = [ d for d in dns - if d not in _CODENAV_SKIP_DIRS and d not in _SENSITIVE_BASENAMES + if d not in _CODENAV_SKIP_DIRS + and d not in _SENSITIVE_BASENAMES + and _can_traverse_tool_path(os.path.realpath(os.path.join(dp, d))) ] for name in fns + dns: full = os.path.join(dp, name) @@ -531,7 +537,7 @@ class GlobTool: if regex.fullmatch(rel) or regex.fullmatch(name): # Skip deny-listed sensitive files (.env, id_rsa, # known_hosts, …) the same way grep does. - if _is_sensitive_path(os.path.realpath(full)): + if _is_denied_tool_path(os.path.realpath(full)): continue try: mtime = os.stat(full).st_mtime @@ -559,7 +565,9 @@ class GrepTool: async def execute(self, content: str, ctx: dict) -> dict: from src.tool_execution import ( _SENSITIVE_FILE_PATTERNS, - _is_sensitive_path, + _can_traverse_tool_path, + _is_denied_tool_path, + _path_within, _resolve_tool_path, _resolve_search_root, _truncate, @@ -592,9 +600,18 @@ class GrepTool: import re as _re import shutil rg = shutil.which("rg") + # ripgrep does not offer a policy callback for each traversed + # result. When the search root contains DATA_DIR, use the Python + # walker so protected state can be pruned while legitimate data + # carve-outs remain searchable. + from src.constants import DATA_DIR + if _path_within(os.path.realpath(DATA_DIR), os.path.realpath(root)): + rg = None if rg: - cmd = [rg, "--line-number", "--no-heading", "--color=never", - "--max-count", str(max_hits)] + cmd = [ + rg, "--no-config", "--no-follow", "--line-number", + "--no-heading", "--color=never", "--max-count", str(max_hits), + ] if ignore_case: cmd.append("--ignore-case") if glob_pat: @@ -627,7 +644,14 @@ class GrepTool: else: file_iter = [] for dp, dns, fns in os.walk(root): - dns[:] = [d for d in dns if d not in _CODENAV_SKIP_DIRS] + if not _can_traverse_tool_path(os.path.realpath(dp)): + dns[:] = [] + continue + dns[:] = [ + d for d in dns + if d not in _CODENAV_SKIP_DIRS + and _can_traverse_tool_path(os.path.realpath(os.path.join(dp, d))) + ] for fn in fns: if glob_pat and not fnmatch.fnmatch(fn, glob_pat): continue @@ -635,7 +659,7 @@ class GrepTool: for fp in file_iter: if len(hits) >= max_hits: break - if _is_sensitive_path(os.path.realpath(fp)): + if _is_denied_tool_path(os.path.realpath(fp)): continue try: with open(fp, "r", encoding="utf-8", errors="strict") as f: diff --git a/src/tool_execution.py b/src/tool_execution.py index 219ebe632..9d37c9bda 100644 --- a/src/tool_execution.py +++ b/src/tool_execution.py @@ -167,18 +167,39 @@ def _agent_readable_data_subdirs() -> tuple[str, ...]: when grep/glob/ls are called with no path. """ from src.constants import ( + DATA_DIR, MAIL_ATTACHMENTS_DIR, PERSONAL_DIR, PERSONAL_UPLOADS_DIR, UPLOAD_DIR, ) - return ( + configured = ( AGENT_WORKSPACE_DIR, UPLOAD_DIR, MAIL_ATTACHMENTS_DIR, PERSONAL_DIR, PERSONAL_UPLOADS_DIR, ) + data_dir = os.path.realpath(DATA_DIR) + safe: list[str] = [] + for raw in configured: + value = str(raw or "").strip() + # These paths are security-policy carve-outs, not ordinary allowlist + # entries. Accept only explicit absolute paths whose canonical target + # is a strict descendant of DATA_DIR. In particular, an empty/dot, + # filesystem-root, ancestor, equality, or symlink-equivalent setting + # must not turn the whole state directory into readable content. + if not value or not os.path.isabs(os.path.expanduser(value)): + continue + resolved = os.path.realpath(os.path.expanduser(value)) + if ( + resolved == data_dir + or not _path_within(resolved, data_dir) + or _is_sensitive_path(resolved) + ): + continue + safe.append(resolved) + return tuple(safe) def _is_app_state_path(resolved: str) -> bool: @@ -199,11 +220,28 @@ def _is_app_state_path(resolved: str) -> bool: if not _path_within(resolved, os.path.realpath(DATA_DIR)): return False return not any( - _path_within(resolved, os.path.realpath(d)) + _path_within(resolved, d) for d in _agent_readable_data_subdirs() ) +def _is_denied_tool_path(resolved: str) -> bool: + """Apply every path deny to a canonical traversal result.""" + return _is_sensitive_path(resolved) or _is_app_state_path(resolved) + + +def _can_traverse_tool_path(resolved: str) -> bool: + """Allow walking a denied state parent only to reach safe carve-outs.""" + if _is_sensitive_path(resolved): + return False + if not _is_app_state_path(resolved): + return True + return any( + _path_within(readable, resolved) + for readable in _agent_readable_data_subdirs() + ) + + def _tool_path_roots() -> list[str]: """Return the list of directory roots that read_file / write_file may touch. Default: project data/ + system temp dirs. Extra roots @@ -416,7 +454,10 @@ def _resolve_search_root(raw_path: str) -> str: return _resolve_tool_path_in_workspace(ws, raw or ws) if not raw: roots = _tool_path_roots() - return roots[0] if roots else os.path.realpath(".") + default_root = os.path.realpath(AGENT_WORKSPACE_DIR) + if default_root in roots and not _is_denied_tool_path(default_root): + return default_root + raise ValueError("default agent workspace is not a safe readable data subdirectory") return _resolve_tool_path(raw) logger = logging.getLogger(__name__) diff --git a/tests/test_agent_state_dir_confinement.py b/tests/test_agent_state_dir_confinement.py index 5bf093779..715842b73 100644 --- a/tests/test_agent_state_dir_confinement.py +++ b/tests/test_agent_state_dir_confinement.py @@ -17,6 +17,8 @@ together, because closing only the first leaves the other two working: so the guard is a property of the path, not of the root it arrived through. """ +import asyncio +import importlib import os from contextlib import contextmanager @@ -38,6 +40,7 @@ from src.tool_execution import ( agent_cwd, vet_workspace, ) +from src.agent_tools.filesystem_tools import GlobTool, GrepTool APP_STATE_FILES = [ "sessions.json", # session token -> username, cleartext @@ -223,3 +226,120 @@ def test_sensitive_deny_list_still_fires_inside_the_agent_workspace(): """The new guard is layered on the existing one, not a replacement.""" with pytest.raises(ValueError, match="sensitive directory"): _resolve_tool_path(os.path.join(AGENT_WORKSPACE_DIR, "id_rsa")) + + +# ── Misconfigured carve-outs and recursive traversal ──────────────── + +def _configure_test_data_tree(monkeypatch, data_dir): + current_constants = importlib.import_module("src.constants") + current_execution = importlib.import_module("src.tool_execution") + monkeypatch.setattr(current_constants, "DATA_DIR", str(data_dir), raising=False) + readable = { + "AGENT_WORKSPACE_DIR": data_dir / "agent_workspace", + "UPLOAD_DIR": data_dir / "uploads", + "MAIL_ATTACHMENTS_DIR": data_dir / "mail-attachments", + "PERSONAL_DIR": data_dir / "personal_docs", + "PERSONAL_UPLOADS_DIR": data_dir / "personal_uploads", + } + for name, path in readable.items(): + monkeypatch.setattr(current_constants, name, str(path), raising=False) + monkeypatch.setattr( + current_execution, + "AGENT_WORKSPACE_DIR", + str(readable["AGENT_WORKSPACE_DIR"]), + ) + return readable + + +@contextmanager +def current_workspace_at(path): + current_execution = importlib.import_module("src.tool_execution") + token = current_execution._active_workspace.set(os.path.realpath(path)) + try: + yield + finally: + current_execution._active_workspace.reset(token) + + +@pytest.mark.parametrize( + "bad_kind", ["equal", "ancestor", "root", "empty", "dot", "symlink"] +) +def test_invalid_readable_carveout_cannot_cancel_state_deny( + tmp_path, monkeypatch, bad_kind +): + data_dir = tmp_path / "data" + data_dir.mkdir() + readable = _configure_test_data_tree(monkeypatch, data_dir) + bad = { + "equal": str(data_dir), + "ancestor": str(tmp_path), + "root": os.path.abspath(os.sep), + "empty": "", + "dot": ".", + }.get(bad_kind) + if bad_kind == "symlink": + link = tmp_path / "data-link" + try: + link.symlink_to(data_dir, target_is_directory=True) + except OSError: + pytest.skip("cannot create symlink") + bad = str(link) + current_execution = importlib.import_module("src.tool_execution") + monkeypatch.setattr(current_execution, "AGENT_WORKSPACE_DIR", bad) + secret = data_dir / "settings.json" + secret.write_text("STATE_SECRET\n", encoding="utf-8") + + with pytest.raises(ValueError, match="application state"): + current_execution._resolve_tool_path(str(secret)) + with pytest.raises(ValueError, match="default agent workspace"): + current_execution._resolve_search_root("") + assert os.path.realpath(readable["UPLOAD_DIR"]) in current_execution._tool_path_roots() + + +def test_recursive_glob_and_grep_hide_state_but_keep_readable_descendants( + tmp_path, monkeypatch +): + data_dir = tmp_path / "data" + data_dir.mkdir() + readable = _configure_test_data_tree(monkeypatch, data_dir) + workspace = readable["AGENT_WORKSPACE_DIR"] + workspace.mkdir() + (workspace / "notes.json").write_text("SHARED_MARKER readable\n", encoding="utf-8") + (data_dir / "settings.json").write_text("SHARED_MARKER secret\n", encoding="utf-8") + + with current_workspace_at(tmp_path): + glob_result = asyncio.run(GlobTool().execute( + '{"pattern": "**/*.json", "path": ""}', {} + )) + grep_result = asyncio.run(GrepTool().execute( + '{"pattern": "SHARED_MARKER", "path": ""}', {} + )) + + assert "notes.json" in glob_result["output"] + assert "settings.json" not in glob_result["output"] + assert "notes.json" in grep_result["output"] + assert "settings.json" not in grep_result["output"] + + +def test_recursive_glob_and_grep_hide_state_from_extra_root(tmp_path, monkeypatch): + data_dir = tmp_path / "data" + data_dir.mkdir() + readable = _configure_test_data_tree(monkeypatch, data_dir) + readable["AGENT_WORKSPACE_DIR"].mkdir() + (readable["AGENT_WORKSPACE_DIR"] / "public.txt").write_text( + "TOKEN visible\n", encoding="utf-8" + ) + (data_dir / "auth.json").write_text("TOKEN hidden\n", encoding="utf-8") + monkeypatch.setattr("src.settings.get_setting", lambda *_a, **_k: [str(tmp_path)]) + + glob_result = asyncio.run(GlobTool().execute( + f'{{"pattern": "**/*", "path": "{tmp_path}"}}', {} + )) + grep_result = asyncio.run(GrepTool().execute( + f'{{"pattern": "TOKEN", "path": "{tmp_path}"}}', {} + )) + + assert "public.txt" in glob_result["output"] + assert "auth.json" not in glob_result["output"] + assert "public.txt" in grep_result["output"] + assert "auth.json" not in grep_result["output"]