From f7063b536443498295550ed36c4d5214f665182f Mon Sep 17 00:00:00 2001 From: RaresKeY <158580472+RaresKeY@users.noreply.github.com> Date: Mon, 31 Aug 2026 14:19:57 +0000 Subject: [PATCH] fix: bound protected filesystem searches --- src/agent_tools/filesystem_tools.py | 156 ++++++++++++++----- src/tool_execution.py | 84 +++++++---- tests/test_agent_state_dir_confinement.py | 173 +++++++++++++++++++++- tests/test_code_nav_tools.py | 11 ++ 4 files changed, 362 insertions(+), 62 deletions(-) diff --git a/src/agent_tools/filesystem_tools.py b/src/agent_tools/filesystem_tools.py index 84b85becb..c5643b629 100644 --- a/src/agent_tools/filesystem_tools.py +++ b/src/agent_tools/filesystem_tools.py @@ -3,8 +3,8 @@ import json import os import re import difflib -import fnmatch import shutil +import time from typing import Optional, Dict, Any, Tuple, List from src.constants import MAX_READ_CHARS, MAX_DIFF_LINES, MAX_OUTPUT_CHARS @@ -407,7 +407,11 @@ def _apply_patch_hunks(original: str, hunks: List[List[str]], label: str) -> str class LsTool: async def execute(self, content: str, ctx: dict) -> dict: - from src.tool_execution import _resolve_tool_path, _resolve_search_root, _truncate + from src.tool_execution import ( + _is_denied_tool_path, + _resolve_search_root, + _truncate, + ) raw_path = "" _s = (content or "").strip() if _s.startswith("{"): @@ -431,6 +435,8 @@ class LsTool: for entry in it: if entry.name.startswith("."): continue + if _is_denied_tool_path(os.path.realpath(entry.path)): + continue try: is_dir = entry.is_dir(follow_symlinks=False) size = entry.stat(follow_symlinks=False).st_size if not is_dir else 0 @@ -565,6 +571,7 @@ class GrepTool: async def execute(self, content: str, ctx: dict) -> dict: from src.tool_execution import ( _SENSITIVE_FILE_PATTERNS, + _agent_readable_data_subdirs, _can_traverse_tool_path, _is_denied_tool_path, _path_within, @@ -600,44 +607,118 @@ 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 + real_root = os.path.realpath(root) + data_dir = os.path.realpath(DATA_DIR) + spans_state = _path_within(data_dir, real_root) + if spans_state and not rg: + return None, "grep: ripgrep is required when the search root contains application state" if rg: - 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: - cmd += ["--glob", glob_pat] - # --iglob (not --glob) so the exclusion is case-insensitive: - # on a case-insensitive filesystem "ID_RSA"/"Known_Hosts" - # resolve to the same secret as their lowercase forms, and the - # Python fallback below already folds case via _is_sensitive_path. - for _pat in _SENSITIVE_FILE_PATTERNS: - cmd += ["--iglob", f"!*{_pat}*"] - for _d in _CODENAV_SKIP_DIRS: - cmd += ["--glob", f"!**/{_d}/**"] - cmd += ["--regexp", pattern, root] - try: - import subprocess - p = subprocess.run(cmd, capture_output=True, text=True, timeout=20) - lines = [ln for ln in (p.stdout or "").splitlines() if ln][:max_hits] - return lines, None - except subprocess.TimeoutExpired: - return None, "grep: timed out" - except Exception as _e: - return None, f"grep: {_e}" + searches: list[tuple[str, list[str]]] = [(real_root, [])] + if spans_state: + searches = [] + # Search everything outside DATA_DIR with a native rg glob + # exclusion. Search validated carve-outs separately so + # their contents remain available without exposing state + # siblings. --no-follow prevents a symlink from bypassing + # the excluded canonical subtree. + if real_root != data_dir: + rel_data = os.path.relpath(data_dir, real_root).replace( + os.sep, "/" + ) + searches.append( + (real_root, [f"!{rel_data}", f"!{rel_data}/**"]) + ) + seen_roots: set[str] = set() + for readable in _agent_readable_data_subdirs(): + if not _path_within( + readable, real_root + ) or not os.path.exists(readable): + continue + canonical = os.path.realpath(readable) + if canonical not in seen_roots: + seen_roots.add(canonical) + searches.append((canonical, [])) + + lines: list[str] = [] + deadline = time.monotonic() + 20 + for search_root, state_excludes in searches: + remaining_hits = max_hits - len(lines) + if remaining_hits <= 0: + break + cmd = [ + rg, "--no-config", "--no-follow", "--line-number", + "--no-heading", "--color=never", "--max-count", + str(remaining_hits), + ] + if ignore_case: + cmd.append("--ignore-case") + if glob_pat: + cmd += ["--glob", glob_pat] + # --iglob (not --glob) so the exclusion is case-insensitive: + # on a case-insensitive filesystem "ID_RSA"/"Known_Hosts" + # resolve to the same secret as their lowercase forms. + for _pat in _SENSITIVE_FILE_PATTERNS: + cmd += ["--iglob", f"!*{_pat}*"] + for _d in _CODENAV_SKIP_DIRS: + cmd += ["--glob", f"!**/{_d}/**"] + for exclusion in state_excludes: + cmd += ["--glob", exclusion] + cmd += ["--regexp", pattern, search_root] + timeout = deadline - time.monotonic() + if timeout <= 0: + return None, "grep: timed out" + try: + import queue + import subprocess + import threading + process = subprocess.Popen( + cmd, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + text=True, + bufsize=1, + ) + except Exception as _e: + return None, f"grep: {_e}" + output: queue.Queue[Optional[str]] = queue.Queue() + + def _read_stdout() -> None: + assert process.stdout is not None + try: + for line in process.stdout: + output.put(line.rstrip("\n")) + finally: + output.put(None) + + threading.Thread(target=_read_stdout, daemon=True).start() + try: + while len(lines) < max_hits: + remaining = deadline - time.monotonic() + if remaining <= 0: + return None, "grep: timed out" + try: + line = output.get(timeout=remaining) + except queue.Empty: + return None, "grep: timed out" + if line is None: + break + if line and line not in lines: + lines.append(line) + finally: + if process.poll() is None: + process.terminate() + try: + process.wait(timeout=1) + except subprocess.TimeoutExpired: + process.kill() + process.wait() + return lines, None try: rx = _re.compile(pattern, _re.IGNORECASE if ignore_case else 0) except _re.error as _e: return None, f"grep: bad pattern: {_e}" + glob_rx = _glob_to_regex(glob_pat.replace("\\", "/")) if glob_pat else None hits = [] if os.path.isfile(root): file_iter = [root] @@ -653,7 +734,12 @@ class GrepTool: 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): + rel = os.path.relpath(os.path.join(dp, fn), root).replace( + os.sep, "/" + ) + if glob_rx and not ( + glob_rx.fullmatch(rel) or glob_rx.fullmatch(fn) + ): continue file_iter.append(os.path.join(dp, fn)) for fp in file_iter: diff --git a/src/tool_execution.py b/src/tool_execution.py index 9d37c9bda..5057ba2c7 100644 --- a/src/tool_execution.py +++ b/src/tool_execution.py @@ -129,14 +129,39 @@ def _is_sensitive_path(resolved: str) -> bool: def _path_within(resolved: str, root: str) -> bool: """True when *resolved* is *root* itself or sits underneath it. - Folds case before comparing, because the caller is a deny rule: a missed - match here ALLOWS the path, where a missed match in the allowlist below - only rejects it. On default macOS a case-variant name opens the same file - and realpath does not canonicalise case, so an unfolded compare would let - the state directory through. casefold rather than os.path.normcase for the - reason given above _is_sensitive_path: normcase is a no-op on POSIX, which - is exactly where that gap lives. + Use the platform's path-case rules. This helper participates in allow + decisions, so unconditional case-folding would let a distinct ``/DATA`` + tree masquerade as a descendant of ``/data`` on case-sensitive systems. """ + resolved, root = os.path.normcase(resolved), os.path.normcase(root) + if resolved == root: + return True + try: + if os.path.commonpath([resolved, root]) == root: + return True + except ValueError: + return False + # normcase is intentionally conservative about assumptions (notably on + # POSIX), so consult the filesystem when paths exist. This recognizes a + # case alias on a case-insensitive volume without treating distinct + # case-sensitive paths as the same allow root. + if os.path.exists(root): + candidate = resolved + while True: + try: + if os.path.exists(candidate) and os.path.samefile(candidate, root): + return True + except OSError: + pass + parent = os.path.dirname(candidate) + if parent == candidate: + break + candidate = parent + return False + + +def _path_within_conservative(resolved: str, root: str) -> bool: + """Containment for deny decisions, folding case to fail closed.""" resolved, root = resolved.casefold(), root.casefold() if resolved == root: return True @@ -151,7 +176,8 @@ def _agent_readable_data_subdirs() -> tuple[str, ...]: The agent's own scratch folder, plus the directories of user content whose paths the application itself gives to the model, which it would then be - unable to open: + unable to open. These normally live under DATA_DIR; the documented mail + attachment override may instead name a disjoint external directory: UPLOAD_DIR the chat upload manifest renders "path=

" and says to read it with read_file (agent_loop.py) @@ -174,29 +200,37 @@ def _agent_readable_data_subdirs() -> tuple[str, ...]: UPLOAD_DIR, ) configured = ( - AGENT_WORKSPACE_DIR, - UPLOAD_DIR, - MAIL_ATTACHMENTS_DIR, - PERSONAL_DIR, - PERSONAL_UPLOADS_DIR, + (AGENT_WORKSPACE_DIR, False), + (UPLOAD_DIR, False), + # This has a documented environment override and may legitimately + # live outside DATA_DIR, but it must never equal/contain DATA_DIR. + (MAIL_ATTACHMENTS_DIR, True), + (PERSONAL_DIR, False), + (PERSONAL_UPLOADS_DIR, False), ) data_dir = os.path.realpath(DATA_DIR) safe: list[str] = [] - for raw in configured: + for raw, external_ok 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. + # These paths are security-policy roots, not ordinary allowlist + # entries. Accept only explicit absolute directory paths. State + # carve-outs must be strict DATA_DIR descendants; the documented mail + # override may also be disjoint. Empty/dot, filesystem-root, ancestor, + # equality, file, or symlink-equivalent settings fail closed. 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) - ): + if os.path.exists(resolved) and not os.path.isdir(resolved): + continue + inside_data = resolved != data_dir and _path_within(resolved, data_dir) + external_safe = ( + external_ok + and resolved != data_dir + and os.path.dirname(resolved) != resolved + and not _path_within(data_dir, resolved) + and not _path_within(resolved, data_dir) + ) + if not (inside_data or external_safe) or _is_sensitive_path(resolved): continue safe.append(resolved) return tuple(safe) @@ -217,7 +251,7 @@ def _is_app_state_path(resolved: str) -> bool: own settings.json or app.db inside a real workspace is not caught. """ from src.constants import DATA_DIR - if not _path_within(resolved, os.path.realpath(DATA_DIR)): + if not _path_within_conservative(resolved, os.path.realpath(DATA_DIR)): return False return not any( _path_within(resolved, d) diff --git a/tests/test_agent_state_dir_confinement.py b/tests/test_agent_state_dir_confinement.py index 715842b73..76042a19d 100644 --- a/tests/test_agent_state_dir_confinement.py +++ b/tests/test_agent_state_dir_confinement.py @@ -20,7 +20,9 @@ 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 +import shutil +import time +from contextlib import contextmanager, nullcontext import pytest @@ -40,7 +42,7 @@ from src.tool_execution import ( agent_cwd, vet_workspace, ) -from src.agent_tools.filesystem_tools import GlobTool, GrepTool +from src.agent_tools.filesystem_tools import GlobTool, GrepTool, LsTool APP_STATE_FILES = [ "sessions.json", # session token -> username, cleartext @@ -343,3 +345,170 @@ def test_recursive_glob_and_grep_hide_state_from_extra_root(tmp_path, monkeypatc assert "auth.json" not in glob_result["output"] assert "public.txt" in grep_result["output"] assert "auth.json" not in grep_result["output"] + + +def test_existing_file_cannot_become_a_readable_directory_carveout( + tmp_path, monkeypatch +): + data_dir = tmp_path / "data" + data_dir.mkdir() + _configure_test_data_tree(monkeypatch, data_dir) + secret = data_dir / "auth.json" + secret.write_text("STATE_SECRET\n", encoding="utf-8") + current_constants = importlib.import_module("src.constants") + monkeypatch.setattr(current_constants, "UPLOAD_DIR", str(secret)) + current_execution = importlib.import_module("src.tool_execution") + + assert ( + os.path.realpath(secret) + not in current_execution._agent_readable_data_subdirs() + ) + with pytest.raises(ValueError, match="application state"): + current_execution._resolve_tool_path(str(secret)) + + +def test_external_mail_attachment_directory_remains_readable(tmp_path, monkeypatch): + data_dir = tmp_path / "data" + data_dir.mkdir() + _configure_test_data_tree(monkeypatch, data_dir) + external = tmp_path / "external-mail" + external.mkdir() + attachment = external / "message.txt" + attachment.write_text("mail body\n", encoding="utf-8") + current_constants = importlib.import_module("src.constants") + monkeypatch.setattr(current_constants, "MAIL_ATTACHMENTS_DIR", str(external)) + current_execution = importlib.import_module("src.tool_execution") + + assert current_execution._resolve_tool_path(str(attachment)) == os.path.realpath( + attachment + ) + + +@pytest.mark.skipif( + os.path.normcase("DATA") == os.path.normcase("data"), + reason="requires a platform with case-sensitive path comparison", +) +def test_case_distinct_path_cannot_masquerade_as_readable_descendant( + tmp_path, monkeypatch +): + data_dir = tmp_path / "data" + data_dir.mkdir() + _configure_test_data_tree(monkeypatch, data_dir) + distinct = tmp_path / "DATA" / "agent_workspace" + distinct.mkdir(parents=True) + current_execution = importlib.import_module("src.tool_execution") + monkeypatch.setattr(current_execution, "AGENT_WORKSPACE_DIR", str(distinct)) + + assert ( + os.path.realpath(distinct) + not in current_execution._agent_readable_data_subdirs() + ) + + +@pytest.mark.parametrize("use_workspace", [True, False]) +def test_ls_hides_protected_entries_when_root_contains_data( + tmp_path, monkeypatch, use_workspace +): + data_dir = tmp_path / "data" + data_dir.mkdir() + _configure_test_data_tree(monkeypatch, data_dir) + (tmp_path / "visible.txt").write_text("visible\n", encoding="utf-8") + secret = data_dir / "settings.json" + secret.write_text("SECRET_WITH_SIZE\n", encoding="utf-8") + if use_workspace: + context = current_workspace_at(tmp_path) + content = '{"path": ""}' + else: + monkeypatch.setattr("src.settings.get_setting", lambda *_a, **_k: [str(tmp_path)]) + context = nullcontext() + content = f'{{"path": "{tmp_path}"}}' + + with context: + result = asyncio.run(LsTool().execute(content, {})) + + assert "visible.txt" in result["output"] + assert "settings.json" not in result["output"] + assert "SECRET_WITH_SIZE" not in result["output"] + assert "data/" not in result["output"] + + +@pytest.mark.skipif(shutil.which("rg") is None, reason="requires ripgrep") +def test_state_spanning_grep_bounds_dangerous_regex(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 / "long.txt").write_text("a" * 250_000 + "!\n", encoding="utf-8") + (data_dir / "auth.txt").write_text("a" * 250_000 + "!\n", encoding="utf-8") + + started = time.monotonic() + with current_workspace_at(tmp_path): + result = asyncio.run(GrepTool().execute( + '{"pattern": "(a+)+$", "path": "", "max_results": 1}', {} + )) + elapsed = time.monotonic() - started + + assert result["exit_code"] == 0 + assert "auth.txt" not in result["output"] + assert elapsed < 5 + + +def test_state_spanning_grep_stops_process_at_max_results(tmp_path, monkeypatch): + import subprocess + + data_dir = tmp_path / "data" + data_dir.mkdir() + _configure_test_data_tree(monkeypatch, data_dir) + instances = [] + + class FakeProcess: + def __init__(self, *args, **kwargs): + self.stdout = iter( + f"{tmp_path}/visible-{index}.txt:1:MATCH\n" for index in range(100) + ) + self.terminated = False + instances.append(self) + + def poll(self): + return 0 if self.terminated else None + + def terminate(self): + self.terminated = True + + def wait(self, timeout=None): + return 0 + + def kill(self): + self.terminated = True + + monkeypatch.setattr(shutil, "which", lambda _name: "/usr/bin/rg") + monkeypatch.setattr(subprocess, "Popen", FakeProcess) + with current_workspace_at(tmp_path): + result = asyncio.run(GrepTool().execute( + '{"pattern": "MATCH", "path": "", "max_results": 1}', {} + )) + + assert len(instances) == 1 + assert instances[0].terminated is True + assert result["output"].count(":1:MATCH") == 1 + + +@pytest.mark.skipif(shutil.which("rg") is None, reason="requires ripgrep") +def test_state_spanning_grep_keeps_relative_glob_semantics(tmp_path, monkeypatch): + data_dir = tmp_path / "data" + data_dir.mkdir() + readable = _configure_test_data_tree(monkeypatch, data_dir) + nested = readable["AGENT_WORKSPACE_DIR"] / "nested" + nested.mkdir(parents=True) + (nested / "readable.py").write_text("PATH_GLOB_MARKER\n", encoding="utf-8") + (data_dir / "protected.py").write_text("PATH_GLOB_MARKER\n", encoding="utf-8") + + with current_workspace_at(tmp_path): + result = asyncio.run(GrepTool().execute( + '{"pattern": "PATH_GLOB_MARKER", "path": "", "glob": "**/*.py"}', + {}, + )) + + assert "readable.py" in result["output"] + assert "protected.py" not in result["output"] diff --git a/tests/test_code_nav_tools.py b/tests/test_code_nav_tools.py index 5be50220a..98598c291 100644 --- a/tests/test_code_nav_tools.py +++ b/tests/test_code_nav_tools.py @@ -91,6 +91,17 @@ def test_grep_python_fallback_when_no_rg(repo, monkeypatch): assert ".git/config" not in r["output"] +def test_grep_python_fallback_uses_relative_glob_paths(repo, monkeypatch): + monkeypatch.setattr(shutil, "which", lambda name: None) + r = _run( + "grep", + f'{{"pattern": "needle|python", "glob": "**/*.py", "path": "{repo}"}}', + ) + assert r["exit_code"] == 0 + assert "a.py" in r["output"] + assert "sub/deep/c.py" in r["output"] + + @pytest.mark.skipif(shutil.which("rg") is None, reason="targets the ripgrep fast-path") def test_grep_skips_case_variant_sensitive_files_rg(repo): """The rg fast-path must exclude deny-listed key files case-insensitively.