diff --git a/launcher.py b/launcher.py index ba158444f..192ba83c6 100644 --- a/launcher.py +++ b/launcher.py @@ -14,6 +14,13 @@ import threading import time import webbrowser +# PyInstaller multiprocessing children re-enter this executable with a private +# bootstrap argument. Consume it before splash/UI or application imports so a +# spawn-based worker does not relaunch the full desktop application. +if __name__ == "__main__": + import multiprocessing + multiprocessing.freeze_support() + # Define a dummy NullWriter to suppress standard stream crashes (isatty etc.) in GUI mode class NullWriter: def write(self, text): diff --git a/setup.py b/setup.py index 5b4eadcb5..8c4934a82 100644 --- a/setup.py +++ b/setup.py @@ -16,7 +16,7 @@ sys.path.insert(0, BASE_DIR) from src.constants import ( DATA_DIR, AUTH_FILE, UPLOAD_DIR, PERSONAL_DIR, PERSONAL_UPLOADS_DIR, TTS_CACHE_DIR, GENERATED_IMAGES_DIR, DEEP_RESEARCH_DIR, CHROMA_DIR, - RAG_DIR, MEMORY_VECTORS_DIR, PASSWORD_MIN_LENGTH, + RAG_DIR, MEMORY_VECTORS_DIR, AGENT_WORKSPACE_DIR, PASSWORD_MIN_LENGTH, ) from core.auth import RESERVED_USERNAMES @@ -31,6 +31,7 @@ DIRS = [ CHROMA_DIR, RAG_DIR, MEMORY_VECTORS_DIR, + AGENT_WORKSPACE_DIR, os.path.join(BASE_DIR, "logs"), ] diff --git a/src/agent_tools/filesystem_tools.py b/src/agent_tools/filesystem_tools.py index f2fa20c54..6a5361ab5 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 @@ -16,6 +16,8 @@ _CODENAV_SKIP_DIRS = frozenset({ }) _CODENAV_MAX_HITS = 200 _CODENAV_MAX_LINE = 400 +_GREP_TIMEOUT_SECONDS = 20 +_GREP_STDERR_PREFIX = 20_000 def _glob_to_regex(pat: str) -> "re.Pattern": @@ -42,6 +44,113 @@ def _glob_to_regex(pat: str) -> "re.Pattern": i += 1 return re.compile("".join(out)) + +def _python_grep_worker(payload: dict, output_queue) -> None: + """Spawn-safe fallback grep worker used when ripgrep is unavailable. + + Keep this at module scope: a frozen Windows executable cannot safely be + relaunched as ``sys.executable -c ...``, while multiprocessing can invoke a + top-level target through its frozen-process bootstrap. + """ + try: + flags = re.IGNORECASE if payload["ignore_case"] else 0 + try: + regex = re.compile(payload["pattern"], flags) + glob_regex = ( + _glob_to_regex(payload["glob"].replace("\\", "/")) + if payload["glob"] + else None + ) + except re.error as exc: + output_queue.put(("error", f"grep: bad pattern: {exc}")) + return + + requested_root = payload["root"] + skip_dirs = set(payload["skip_dirs"]) + sensitive = {name.casefold() for name in payload["sensitive_names"]} + max_hits = payload["max_hits"] + hits = 0 + + def within(path: str, root: str) -> bool: + try: + return os.path.commonpath( + [os.path.normcase(path), os.path.normcase(root)] + ) == os.path.normcase(root) + except ValueError: + return False + + def safe_file(path: str, target: str) -> Optional[str]: + if os.path.islink(path): + return None + canonical = os.path.realpath(path) + if not within(canonical, requested_root) or not within(canonical, target): + return None + parts = [part.casefold() for part in canonical.split(os.sep)] + if any(part in sensitive for part in parts): + return None + try: + if not os.path.isfile(canonical) or os.stat(canonical).st_nlink > 1: + return None + except OSError: + return None + return canonical + + for target in payload["targets"]: + if hits >= max_hits: + break + if os.path.isfile(target): + file_iter = iter((target,)) + else: + def walk_files(): + for directory, dirnames, filenames in os.walk( + target, followlinks=False + ): + dirnames[:] = [ + name + for name in dirnames + if name not in skip_dirs + and name.casefold() not in sensitive + and not os.path.islink(os.path.join(directory, name)) + ] + for name in filenames: + yield os.path.join(directory, name) + + file_iter = walk_files() + + for candidate in file_iter: + path = safe_file(candidate, target) + if path is None: + continue + relative = os.path.relpath(path, requested_root).replace(os.sep, "/") + if glob_regex and not ( + glob_regex.fullmatch(relative) + or glob_regex.fullmatch(os.path.basename(path)) + ): + continue + try: + with open(path, "r", encoding="utf-8", errors="strict") as handle: + for number, line in enumerate(handle, 1): + if regex.search(line): + output_queue.put(( + "match", + path, + number, + line.rstrip()[:_CODENAV_MAX_LINE], + )) + hits += 1 + if hits >= max_hits: + break + except (UnicodeDecodeError, OSError): + continue + if hits >= max_hits: + break + output_queue.put(("done",)) + except BaseException as exc: + try: + output_queue.put(("error", f"grep: fallback worker failed: {exc}")) + except BaseException: + pass + def _unified_diff(old: str, new: str, path: str) -> Optional[Dict[str, Any]]: if old == new: return None @@ -407,7 +516,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 +544,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 @@ -458,7 +573,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 +623,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 +633,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 +652,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 @@ -558,9 +679,12 @@ class GlobTool: class GrepTool: async def execute(self, content: str, ctx: dict) -> dict: from src.tool_execution import ( + _SENSITIVE_BASENAMES, _SENSITIVE_FILE_PATTERNS, + _agent_readable_data_subdirs, + _is_denied_tool_path, _is_sensitive_path, - _resolve_tool_path, + _path_within, _resolve_search_root, _truncate, ) @@ -589,64 +713,307 @@ class GrepTool: return {"error": f"grep: {e}", "exit_code": 1} def _grep(): - import re as _re - import shutil + import multiprocessing + import queue + import subprocess + import threading + + from src.constants import DATA_DIR + rg = shutil.which("rg") - if rg: - cmd = [rg, "--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] + real_root = os.path.realpath(root) + data_dir = os.path.realpath(DATA_DIR) + spans_state = _path_within(data_dir, real_root) + + def is_top_level_safe(path: str, *, partition_generated: bool) -> bool: + lexical = os.path.abspath(path) + if os.path.islink(lexical): + return False + canonical = os.path.realpath(lexical) + if not _path_within(canonical, real_root): + return False + if partition_generated and os.path.basename(lexical) in _CODENAV_SKIP_DIRS: + return False + if _is_sensitive_path(canonical) or _is_denied_tool_path(canonical): + return False + return True + + def safe_targets() -> tuple[list[str], Optional[str]]: + candidates: list[tuple[str, bool]] = [] + if not spans_state: + # Preserve direct-root compatibility: skip-directory policy + # prunes descendants, but an explicitly requested allowed + # root named node_modules remains searchable. + candidates.append((real_root, False)) + else: + current = real_root + if current != data_dir: + for part in os.path.relpath(data_dir, current).split(os.sep): + try: + with os.scandir(current) as entries: + for entry in entries: + if entry.name != part: + # Reject a sibling link lexically before + # canonicalizing or treating it as a target. + if entry.is_symlink(): + continue + candidates.append((entry.path, True)) + except OSError as exc: + return [], f"grep: {exc}" + current = os.path.join(current, part) + for readable in _agent_readable_data_subdirs(): + if ( + _path_within(readable, data_dir) + and _path_within(readable, real_root) + and os.path.exists(readable) + ): + candidates.append((readable, True)) + + targets: list[str] = [] + seen: set[str] = set() + for candidate, partition_generated in candidates: + if not is_top_level_safe( + candidate, partition_generated=partition_generated + ): + continue + canonical = os.path.realpath(candidate) + if canonical not in seen: + seen.add(canonical) + targets.append(canonical) + return targets, None + + targets, target_error = safe_targets() + if target_error: + return None, target_error + + base = real_root if os.path.isdir(real_root) else os.path.dirname(real_root) + deadline = time.monotonic() + _GREP_TIMEOUT_SECONDS + lines: list[str] = [] + + def parse_rg_result(raw: str) -> Optional[str]: 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}" - try: - rx = _re.compile(pattern, _re.IGNORECASE if ignore_case else 0) - except _re.error as _e: - return None, f"grep: bad pattern: {_e}" - hits = [] - if os.path.isfile(root): - file_iter = [root] - else: - file_iter = [] - for dp, dns, fns in os.walk(root): - dns[:] = [d for d in dns if d not in _CODENAV_SKIP_DIRS] - for fn in fns: - if glob_pat and not fnmatch.fnmatch(fn, glob_pat): + record = json.loads(raw) + except (TypeError, json.JSONDecodeError): + return None + if record.get("type") != "match": + return None + data = record.get("data") or {} + path = (data.get("path") or {}).get("text") + text_value = (data.get("lines") or {}).get("text") + number = data.get("line_number") + if not isinstance(path, str) or not isinstance(text_value, str): + return None + absolute = path if os.path.isabs(path) else os.path.join(base, path) + canonical = os.path.realpath(absolute) + if not _path_within(canonical, real_root) or _is_denied_tool_path(canonical): + return None + return f"{os.path.abspath(absolute)}:{number}:{text_value.rstrip()[:_CODENAV_MAX_LINE]}" + + def run_rg(cmd: list[str]) -> Optional[str]: + try: + process = subprocess.Popen( + cmd, + cwd=base, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + bufsize=1, + ) + except Exception as exc: + return f"grep: {exc}" + output: queue.Queue[Optional[str]] = queue.Queue(maxsize=max_hits + 2) + stderr_prefix: list[str] = [] + stderr_size = 0 + stop_reader = threading.Event() + + def enqueue_stdout(value: Optional[str]) -> bool: + # The consumer stops at the result cap or deadline. Never + # leave a producer blocked on its bounded queue afterward. + while not stop_reader.is_set(): + try: + output.put(value, timeout=0.05) + return True + except queue.Full: continue - file_iter.append(os.path.join(dp, fn)) - for fp in file_iter: - if len(hits) >= max_hits: - break - if _is_sensitive_path(os.path.realpath(fp)): - continue + return False + + def read_stdout() -> None: + assert process.stdout is not None + try: + for line in process.stdout: + if not enqueue_stdout(line.rstrip("\n")): + break + finally: + enqueue_stdout(None) + + def read_stderr() -> None: + nonlocal stderr_size + assert process.stderr is not None + while True: + chunk = process.stderr.read(4096) + if not chunk: + break + if stderr_size < _GREP_STDERR_PREFIX: + kept = chunk[:_GREP_STDERR_PREFIX - stderr_size] + stderr_prefix.append(kept) + stderr_size += len(kept) + + stdout_thread = threading.Thread(target=read_stdout, daemon=True) + stderr_thread = threading.Thread(target=read_stderr, daemon=True) + stdout_thread.start() + stderr_thread.start() + timed_out = False + capped = False try: - with open(fp, "r", encoding="utf-8", errors="strict") as f: - for i, line in enumerate(f, 1): - if rx.search(line): - hits.append(f"{fp}:{i}:{line.rstrip()[:_CODENAV_MAX_LINE]}") - if len(hits) >= max_hits: - break - except (UnicodeDecodeError, OSError): - continue - return hits, None + while len(lines) < max_hits: + remaining = deadline - time.monotonic() + if remaining <= 0: + timed_out = True + break + try: + raw = output.get(timeout=remaining) + except queue.Empty: + timed_out = True + break + if raw is None: + break + parsed = parse_rg_result(raw) + if parsed and parsed not in lines: + lines.append(parsed) + capped = len(lines) >= max_hits + finally: + stop_reader.set() + if (timed_out or capped) and process.poll() is None: + process.terminate() + try: + remaining = max(0.01, deadline - time.monotonic()) + return_code = process.wait(timeout=min(1, remaining)) + except subprocess.TimeoutExpired: + process.kill() + return_code = process.wait() + stdout_thread.join() + stderr_thread.join() + if timed_out: + return "grep: timed out" + if not capped and return_code not in (0, 1): + detail = "".join(stderr_prefix).strip() + return f"grep: {detail or f'process exited {return_code}'}" + return None + + if rg: + # Validate even when policy filtering leaves no search targets. + if not targets: + error = run_rg([rg, "--json", "--no-config", "--regexp", pattern]) + return (None, error) if error else ([], None) + relative_targets = [os.path.relpath(target, base) for target in targets] + for offset in range(0, len(relative_targets), 128): + if len(lines) >= max_hits: + break + cmd = [ + rg, "--json", "--no-config", "--no-follow", + "--max-count", str(max_hits - len(lines)), + "--max-columns", str(_CODENAV_MAX_LINE), + "--max-columns-preview", + ] + if ignore_case: + cmd.append("--ignore-case") + if glob_pat: + cmd += ["--glob", glob_pat] + for sensitive_pattern in _SENSITIVE_FILE_PATTERNS: + cmd += ["--iglob", f"!{sensitive_pattern}"] + for skipped_dir in _CODENAV_SKIP_DIRS: + cmd += ["--glob", f"!**/{skipped_dir}/**"] + cmd += ["--regexp", pattern, "--", *relative_targets[offset:offset + 128]] + error = run_rg(cmd) + if error: + return None, error + return lines, None + + # This runs inside asyncio.to_thread(), so forking would clone a + # multithreaded process and can deadlock. Spawn is platform-safe and + # PyInstaller-compatible via launcher's early freeze_support(). + payload = { + "root": real_root, + "targets": targets, + "pattern": pattern, + "ignore_case": ignore_case, + "glob": glob_pat, + "max_hits": max_hits, + "skip_dirs": tuple(_CODENAV_SKIP_DIRS), + "sensitive_names": tuple( + set(_SENSITIVE_BASENAMES) | set(_SENSITIVE_FILE_PATTERNS) + ), + } + try: + context = multiprocessing.get_context("spawn") + output_queue = context.Queue(maxsize=max_hits + 2) + worker = context.Process( + target=_python_grep_worker, args=(payload, output_queue) + ) + worker.start() + except Exception as exc: + try: + output_queue.close() + except (NameError, OSError, ValueError): + pass + return None, f"grep: could not start fallback worker: {exc}" + error = None + completed = False + try: + while len(lines) < max_hits: + remaining = deadline - time.monotonic() + if remaining <= 0: + error = "grep: timed out" + break + try: + # Keep queue waits short enough to observe a spawn + # worker that dies during bootstrap/import before it + # can enqueue either an error or the done sentinel. + record = output_queue.get(timeout=min(0.05, remaining)) + except queue.Empty: + if worker.is_alive(): + continue + worker.join(timeout=0) + try: + # A multiprocessing queue's feeder can make the + # final record visible at process-exit time. Give + # that record precedence over the exit status. + remaining = deadline - time.monotonic() + record = output_queue.get( + timeout=min(0.05, max(0, remaining)) + ) + except queue.Empty: + error = f"grep: fallback worker exited {worker.exitcode}" + break + if record[0] == "done": + completed = True + break + if record[0] == "error": + error = record[1] + break + _, path, number, text_value = record + canonical = os.path.realpath(path) + if not _path_within(canonical, real_root) or _is_denied_tool_path(canonical): + continue + rendered = f"{path}:{number}:{text_value}" + if rendered not in lines: + lines.append(rendered) + finally: + if completed: + worker.join(timeout=min(1, max(0.01, deadline - time.monotonic()))) + if worker.is_alive(): + worker.terminate() + worker.join(timeout=1) + if worker.is_alive(): + worker.kill() + worker.join() + output_queue.close() + if error: + return None, error + if worker.exitcode not in (0, None) and len(lines) < max_hits: + return None, f"grep: fallback worker exited {worker.exitcode}" + return lines, None lines, err = await asyncio.to_thread(_grep) if err: diff --git a/src/app_initializer.py b/src/app_initializer.py index 1b29f06d2..23fdc68ad 100644 --- a/src/app_initializer.py +++ b/src/app_initializer.py @@ -2,10 +2,11 @@ """Initialize all application components and dependencies.""" import os import logging +import stat from typing import Dict, Any from src.constants import ( - DATA_DIR, PERSONAL_DIR, RUNBOOK_DIR, UPLOAD_DIR, + DATA_DIR, PERSONAL_DIR, RUNBOOK_DIR, UPLOAD_DIR, AGENT_WORKSPACE_DIR, SESSIONS_FILE, DEFAULT_HOST, OPENAI_API_KEY ) from src.memory import MemoryManager @@ -30,7 +31,35 @@ def create_directories(): """Create necessary directories if they don't exist.""" for directory in (DATA_DIR, PERSONAL_DIR, RUNBOOK_DIR, UPLOAD_DIR): os.makedirs(directory, exist_ok=True) - + + # The model-controlled workspace must be a real child of DATA_DIR. Never + # follow a pre-existing symlink here: it would silently move the default + # native-file root outside the application volume before any resolver runs. + data_root = os.path.realpath(os.path.abspath(os.path.expanduser(DATA_DIR))) + workspace = os.path.abspath(os.path.expanduser(AGENT_WORKSPACE_DIR)) + expected_workspace = os.path.join(data_root, "agent_workspace") + # Validate the real parent so a supported DATA_DIR bind/symlink works, but + # require the fixed internal carve-out name and reject a link at the model- + # controlled workspace entry itself. + if ( + os.path.basename(workspace) != "agent_workspace" + or os.path.realpath(os.path.dirname(workspace)) != data_root + ): + raise RuntimeError("agent workspace must be the canonical child of DATA_DIR") + if os.path.lexists(workspace): + mode = os.lstat(workspace).st_mode + if stat.S_ISLNK(mode) or not stat.S_ISDIR(mode): + raise RuntimeError("agent workspace must be a real directory") + else: + os.mkdir(workspace, 0o700) + resolved_workspace = os.path.realpath(workspace) + if resolved_workspace != expected_workspace: + raise RuntimeError("agent workspace must be the canonical child of DATA_DIR") + try: + os.chmod(workspace, 0o700) + except OSError: + pass + def initialize_managers(base_dir: str, rag_manager=None) -> Dict[str, Any]: """ Initialize all manager and handler instances. diff --git a/src/constants.py b/src/constants.py index 28d47efa0..d4f8ba63f 100644 --- a/src/constants.py +++ b/src/constants.py @@ -54,6 +54,11 @@ GALLERY_DIR = os.path.join(DATA_DIR, "gallery") GALLERY_UPLOADS_DIR = os.path.join(DATA_DIR, "gallery_uploads") MEMORY_VECTORS_DIR = os.path.join(DATA_DIR, "memory_vectors") +# The only part of DATA_DIR the agent's file tools and subprocesses may touch. +# Everything else under DATA_DIR is application state (session store, auth +# database, encryption key, settings), and the agent has no business reading it. +AGENT_WORKSPACE_DIR = os.path.join(DATA_DIR, "agent_workspace") + # Paths with an intentional dedicated env override, defaulting under DATA_DIR. MAIL_ATTACHMENTS_DIR = os.getenv("ODYSSEUS_MAIL_ATTACHMENTS_DIR", os.path.join(DATA_DIR, "mail-attachments")) # `or` (not os.getenv's default arg) so a PRESENT-but-EMPTY value falls back to diff --git a/src/tool_execution.py b/src/tool_execution.py index 8c0c83032..230c41a46 100644 --- a/src/tool_execution.py +++ b/src/tool_execution.py @@ -15,6 +15,7 @@ import logging import os import pathlib import re +import stat import sys import time from typing import Any, Awaitable, Callable, Dict, Optional, Tuple @@ -30,7 +31,12 @@ from src.tool_security import ( from src.tool_capabilities import ToolRunSecurityContext, blocked_tool_result from src.tool_approvals import ExactToolApproval from src.tool_policy import ToolPolicy -from src.constants import MAX_OUTPUT_CHARS, MAX_READ_CHARS, MAX_DIFF_LINES, DATA_DIR +from src.constants import ( + MAX_OUTPUT_CHARS, + MAX_READ_CHARS, + MAX_DIFF_LINES, + AGENT_WORKSPACE_DIR, +) from src.tool_utils import _truncate, get_mcp_manager @@ -46,11 +52,11 @@ _MISSING_TOOL_SECURITY_CONTEXT = _MissingToolSecurityContext() NO_TOOL_SECURITY_CONTEXT = _NoToolSecurityContext() # Persistent working directory for agent subprocesses. -# Resolves to /data, which is the bind-mounted volume in Docker -# (/app/data) and the local data directory for manual installs. -# Using this as cwd and HOME prevents the agent from silently creating files -# in ephemeral container layers that are lost on the next rebuild. -_AGENT_WORKDIR = DATA_DIR +# Resolves to /data/agent_workspace, inside the bind-mounted volume +# in Docker (/app/data), so files survive a rebuild as before. The subdirectory +# rather than data/ itself keeps agent scratch files and dotfiles out of the +# directory holding the session store and the auth database. +_AGENT_WORKDIR = AGENT_WORKSPACE_DIR @@ -66,10 +72,15 @@ _AGENT_WORKDIR = DATA_DIR # 1. Sensitive-subpath deny list — checked FIRST. Blocks .ssh, # .gnupg, shell rc files, token/env files even if the root above # them is on the allowlist. -# 2. Allowlist — only the directories the agent legitimately needs -# (project data/, system tmp). $HOME is NOT on the default list. -# 3. Opt-in extra roots — admin can add broader roots via the -# "tool_path_extra_roots" setting (list of path strings). +# 2. Application-state deny (_is_app_state_path) - DATA_DIR holds the +# session store, auth database, app key and settings, so only +# _agent_readable_data_subdirs() is readable inside it. +# 3. Allowlist - only the directories the agent legitimately needs +# (its data/ workspace, user content, system tmp). $HOME is NOT on +# the default list. +# 4. Opt-in extra roots - admin can add broader roots via the +# "tool_path_extra_roots" setting. These cannot re-open DATA_DIR; +# rule 2 is independent of which root a path arrived through. # --------------------------------------------------------------------------- _SENSITIVE_BASENAMES: set[str] = { @@ -116,6 +127,184 @@ def _is_sensitive_path(resolved: str) -> bool: return filename in _SENSITIVE_FILE_PATTERNS_CF +def _path_within(resolved: str, root: str) -> bool: + """True when *resolved* is *root* itself or sits underneath it. + + 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 + try: + return os.path.commonpath([resolved, root]) == root + except ValueError: + return False + + +def _agent_readable_data_subdirs() -> tuple[str, ...]: + """The only parts of DATA_DIR the agent's file tools may reach. + + 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. 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) + MAIL_ATTACHMENTS_DIR download_attachment returns the path and its own + description tells the model to read it + PERSONAL_DIR GET /api/personal returns a path per file and is + reachable through the app_api tool; RUNBOOK_DIR + nests under it + PERSONAL_UPLOADS_DIR indexed as a personal-docs directory, which + manage_rag lists as an absolute path + + Order matters: the first entry is roots[0], which _resolve_search_root uses + 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, + ) + configured = ( + (AGENT_WORKSPACE_DIR, "agent_workspace", False), + (UPLOAD_DIR, "uploads", 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, "mail-attachments", True), + (PERSONAL_DIR, "personal_docs", False), + (PERSONAL_UPLOADS_DIR, "personal_uploads", False), + ) + configured_data_dir = os.path.abspath(os.path.expanduser(str(DATA_DIR))) + data_dir = os.path.realpath(configured_data_dir) + safe: list[str] = [] + for raw, internal_name, external_ok in configured: + value = str(raw or "").strip() + # These paths are security-policy roots, not ordinary allowlist + # entries. Internal roles may inherit a relative DATA_DIR, but must + # still resolve to their exact canonical child below. External mail + # overrides require an absolute, disjoint directory. + if not value: + continue + expanded = os.path.abspath(os.path.expanduser(value)) + # A policy root must not acquire an exemption by redirecting its final + # path component to protected state or to an unrelated external tree. + if os.path.islink(expanded): + continue + resolved = os.path.realpath(expanded) + if os.path.exists(resolved) and not os.path.isdir(resolved): + continue + expected_internal = os.path.join(data_dir, internal_name) + expected_configured = os.path.join(configured_data_dir, internal_name) + inside_data = ( + os.path.normcase(expanded) + in { + os.path.normcase(expected_configured), + os.path.normcase(expected_internal), + } + and resolved == expected_internal + ) + external_safe = ( + external_ok + and os.path.isabs(os.path.expanduser(value)) + 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) + + +def _is_app_state_path(resolved: str) -> bool: + """True for anything under DATA_DIR that is not agent-readable. + + DATA_DIR holds the session store, the auth database, the app encryption key + and the settings file. A model-supplied path must not reach those through + any root, so this is checked in both resolvers rather than expressed as an + absence from the allowlist: a workspace bound at or above the data + directory, or an opt-in tool_path_extra_roots entry covering it, would + otherwise put them back in reach. + + A containment rule rather than a filename deny list, so state files added + later are covered without anyone remembering to list them, and so a user's + own settings.json or app.db inside a real workspace is not caught. + """ + from src.constants import DATA_DIR + if not _path_within_conservative(resolved, os.path.realpath(DATA_DIR)): + return False + return not any( + _path_within(resolved, d) + for d in _agent_readable_data_subdirs() + ) + + +def _is_hardlinked_regular_file(resolved: str) -> bool: + """Reject inode aliases that can smuggle DATA_DIR state into an allow root.""" + try: + target = os.stat(resolved, follow_symlinks=False) + except OSError: + return False + return stat.S_ISREG(target.st_mode) and getattr(target, "st_nlink", 1) > 1 + + +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) + or _is_hardlinked_regular_file(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 @@ -123,9 +312,9 @@ def _tool_path_roots() -> list[str]: """ roots: list[str] = [] - # Project data directory — the agent's primary workspace. - from src.constants import DATA_DIR - roots.append(DATA_DIR) + # The agent's workspace plus the user-content directories inside data/. + # The rest of DATA_DIR is denied by _is_app_state_path. + roots.extend(_agent_readable_data_subdirs()) # /tmp (and its macOS realpath /private/tmp). roots.append("/tmp") @@ -193,6 +382,12 @@ def _resolve_tool_path(raw_path: str) -> str: f"path '{raw_path}' is inside a sensitive directory " f"(e.g. .ssh, .gnupg) or matches a sensitive filename" ) + if _is_app_state_path(resolved): + raise ValueError( + f"path '{raw_path}' is inside the application state directory" + ) + if _is_hardlinked_regular_file(resolved): + raise ValueError(f"path '{raw_path}' is a hard-linked file") for root in _tool_path_roots(): if resolved == root: @@ -228,6 +423,12 @@ def _resolve_tool_path_in_workspace(workspace: str, raw_path: str) -> str: f"path '{raw_path}' is inside a sensitive directory " f"(e.g. .ssh, .gnupg) or matches a sensitive filename" ) + if _is_app_state_path(resolved): + raise ValueError( + f"path '{raw_path}' is inside the application state directory" + ) + if _is_hardlinked_regular_file(resolved): + raise ValueError(f"path '{raw_path}' is a hard-linked file") if resolved != base: # normcase so containment holds on case-insensitive filesystems # (Windows, default macOS): it lowercases on Windows and is a no-op on @@ -277,6 +478,10 @@ def vet_workspace(raw: str) -> Optional[str]: resolved = os.path.realpath(os.path.expanduser(raw)) if not os.path.isdir(resolved) or _is_sensitive_path(resolved): return None + # Refuse the bind rather than binding a workspace where every subsequent + # tool call would fail on the same deny list. + if _is_app_state_path(resolved): + return None # Reject filesystem roots: binding / (or a Windows drive/UNC root) as the # workspace would make every absolute path "inside" it, collapsing the # confinement into host-wide file access. A root is its own dirname, which @@ -289,7 +494,13 @@ def vet_workspace(raw: str) -> Optional[str]: def agent_cwd() -> str: """Working directory for agent subprocesses (bash/python/background jobs): the active workspace when set, else the persistent data dir.""" - return get_active_workspace() or _AGENT_WORKDIR + workspace = get_active_workspace() + if workspace: + return workspace + resolved = os.path.realpath(_AGENT_WORKDIR) + if resolved not in _agent_readable_data_subdirs(): + raise RuntimeError("agent workspace is not a safe real directory") + return resolved def get_mcp_manager(): @@ -304,16 +515,22 @@ def _resolve_search_root(raw_path: str) -> str: With a workspace active, the workspace folder is the root and a supplied path is confined inside it. Otherwise an empty path defaults to the agent's - primary root (project data dir) and a supplied path is confined by the - global allowlist + sensitive-file policy. + primary root (its workspace under the project data dir) and a supplied path + is confined by the global allowlist + sensitive-file policy. """ raw = (raw_path or "").strip() ws = get_active_workspace() if ws: - return os.path.realpath(ws) if not raw else _resolve_tool_path_in_workspace(ws, raw) + # Resolve the empty case as the workspace path rather than returning + # it directly: returned unchecked it skipped both deny lists, so a + # bare ls listed whatever the workspace was bound to. + 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 new file mode 100644 index 000000000..f719b7686 --- /dev/null +++ b/tests/test_agent_state_dir_confinement.py @@ -0,0 +1,1044 @@ +"""The agent's file tools must not reach the application's own state. + +read_file / grep / glob / ls resolve model-supplied paths against +_tool_path_roots(), and the data directory holds the session store, the +credential database, the encryption key and the settings file. A read tool +pointed at those is a credential disclosure, and no approval prompt stands in +the way because reads are classified read_workspace and pass the untrusted- +context gate untouched. + +The agent gets its own subdirectory instead. Three routes have to close +together, because closing only the first leaves the other two working: + + - the default roots, which put DATA_DIR first + - an active workspace bound at (or above) the data directory + - a tool_path_extra_roots setting that covers the data directory + +so the guard is a property of the path, not of the root it arrived through. +""" + +import asyncio +import importlib +import json +import multiprocessing +import os +import queue +import shutil +import time +from contextlib import contextmanager, nullcontext + +import pytest + +from src.constants import ( + AGENT_WORKSPACE_DIR, + DATA_DIR, + MAIL_ATTACHMENTS_DIR, + PERSONAL_DIR, + PERSONAL_UPLOADS_DIR, + RUNBOOK_DIR, + UPLOAD_DIR, +) +from src.tool_execution import ( + _active_workspace, + _resolve_search_root, + _resolve_tool_path, + agent_cwd, + vet_workspace, +) +from src.agent_tools.filesystem_tools import GlobTool, GrepTool, LsTool + +APP_STATE_FILES = [ + "sessions.json", # session token -> username, cleartext + "auth.json", # bcrypt hashes, admin flags, privileges + "app.db", # every user's notes, documents, mail rows + ".app_key", # Fernet key for secret_storage + "settings.json", # provider API keys +] + + +@contextmanager +def workspace_at(path): + """Bind an active workspace for the body of a test. + + Set and reset in the same context; a ContextVar token cannot be reset from + fixture teardown, which runs in a different one. + """ + token = _active_workspace.set(os.path.realpath(path)) + try: + yield + finally: + _active_workspace.reset(token) + + +# ── The default roots ──────────────────────────────────────────────── + +@pytest.mark.parametrize("name", APP_STATE_FILES) +def test_blocks_app_state_file(name): + with pytest.raises(ValueError, match="application state"): + _resolve_tool_path(os.path.join(DATA_DIR, name)) + + +def test_blocks_listing_the_data_directory_itself(): + """`ls data` enumerated the state files, which is how an attacker who + does not know the install path finds them.""" + with pytest.raises(ValueError, match="application state"): + _resolve_tool_path(DATA_DIR) + + +def test_blocks_app_state_reached_by_relative_path(monkeypatch): + """The data directory is a relative hop from the checkout root, so + confinement cannot depend on the model supplying an absolute path.""" + monkeypatch.chdir(os.path.dirname(DATA_DIR)) + with pytest.raises(ValueError, match="application state"): + _resolve_tool_path(os.path.join(os.path.basename(DATA_DIR), "sessions.json")) + + +def test_blocks_app_state_reached_through_a_symlink(tmp_path): + """/tmp is an allowed root and the agent can create links there in an + un-armed turn, so containment has to survive one.""" + link = tmp_path / "shortcut" + try: + link.symlink_to(DATA_DIR) + except OSError: + pytest.skip("cannot create symlink") + with pytest.raises(ValueError, match="application state"): + _resolve_tool_path(str(link / "sessions.json")) + + +def test_native_file_tools_hide_control_plane_hardlink_alias(tmp_path, monkeypatch): + """A pathname inside an allowed root must not alias a protected state inode.""" + data_dir = tmp_path / "data" + data_dir.mkdir() + readable = _configure_test_data_tree(monkeypatch, data_dir) + workspace = readable["AGENT_WORKSPACE_DIR"] + workspace.mkdir() + secret = data_dir / "sessions.json" + secret.write_text("LIVE_ADMIN_SESSION\n", encoding="utf-8") + alias = workspace / "notes.txt" + try: + os.link(secret, alias) + except OSError: + pytest.skip("cannot create hardlink") + + with pytest.raises(ValueError, match="hard-linked"): + importlib.import_module("src.tool_execution")._resolve_tool_path(str(alias)) + + ls_result = asyncio.run(LsTool().execute( + f'{{"path": "{workspace}"}}', {} + )) + glob_result = asyncio.run(GlobTool().execute( + f'{{"pattern": "**/*", "path": "{workspace}"}}', {} + )) + grep_result = asyncio.run(GrepTool().execute( + f'{{"pattern": "LIVE_ADMIN_SESSION", "path": "{workspace}"}}', {} + )) + assert "notes.txt" not in ls_result["output"] + assert "notes.txt" not in glob_result["output"] + assert "notes.txt" not in grep_result["output"] + assert ":1:LIVE_ADMIN_SESSION" not in grep_result["output"] + + +def test_blocks_app_state_on_a_case_insensitive_filesystem(): + """On default macOS a case-variant path opens the same file, and realpath + does not canonicalise case there the way it does on Windows. + + This deny rule fails OPEN when containment misses, unlike the allowlist + beside it, which fails closed. So it folds case, for the same reason + _is_sensitive_path does and not with normcase, which is a no-op on POSIX. + """ + shouty = os.path.join(DATA_DIR.upper(), "SESSIONS.JSON") + with pytest.raises(ValueError, match="application state"): + _resolve_tool_path(shouty) + + +def test_default_search_root_is_the_agent_workspace(): + """grep/glob/ls with no path fall back to roots[0]. That was DATA_DIR.""" + assert _resolve_search_root("") == os.path.realpath(AGENT_WORKSPACE_DIR) + + +def test_startup_rejects_agent_workspace_symlink_escape(tmp_path, monkeypatch): + """Startup must not accept a dedicated workspace redirected outside DATA_DIR.""" + import src.app_initializer as app_initializer + import src.tool_execution as tool_execution + + data_dir = tmp_path / "data" + outside = tmp_path / "outside" + data_dir.mkdir() + outside.mkdir() + workspace = data_dir / "agent_workspace" + try: + workspace.symlink_to(outside, target_is_directory=True) + except OSError: + pytest.skip("cannot create symlink") + + personal = data_dir / "personal_docs" + monkeypatch.setattr(app_initializer, "DATA_DIR", str(data_dir)) + monkeypatch.setattr(app_initializer, "PERSONAL_DIR", str(personal)) + monkeypatch.setattr(app_initializer, "RUNBOOK_DIR", str(personal / "runbook")) + monkeypatch.setattr(app_initializer, "UPLOAD_DIR", str(data_dir / "uploads")) + monkeypatch.setattr(app_initializer, "AGENT_WORKSPACE_DIR", str(workspace)) + monkeypatch.setattr(tool_execution, "AGENT_WORKSPACE_DIR", str(workspace)) + + with pytest.raises(RuntimeError, match="real directory"): + app_initializer.create_directories() + + +def test_startup_allows_workspace_below_symlinked_data_dir(tmp_path, monkeypatch): + import src.app_initializer as app_initializer + + real_data = tmp_path / "real-data" + real_data.mkdir() + data_link = tmp_path / "mounted-data" + try: + data_link.symlink_to(real_data, target_is_directory=True) + except OSError: + pytest.skip("cannot create symlink") + workspace = data_link / "agent_workspace" + personal = data_link / "personal_docs" + monkeypatch.setattr(app_initializer, "DATA_DIR", str(data_link)) + monkeypatch.setattr(app_initializer, "PERSONAL_DIR", str(personal)) + monkeypatch.setattr(app_initializer, "RUNBOOK_DIR", str(personal / "runbook")) + monkeypatch.setattr(app_initializer, "UPLOAD_DIR", str(data_link / "uploads")) + monkeypatch.setattr(app_initializer, "AGENT_WORKSPACE_DIR", str(workspace)) + + app_initializer.create_directories() + + assert workspace.is_dir() + assert not workspace.is_symlink() + assert os.path.realpath(workspace) == str(real_data / "agent_workspace") + readable = _configure_test_data_tree(monkeypatch, data_link) + note = readable["AGENT_WORKSPACE_DIR"] / "note.txt" + note.write_text("visible\n", encoding="utf-8") + assert importlib.import_module("src.tool_execution")._resolve_tool_path( + str(note) + ) == os.path.realpath(note) + + +def test_agent_workspace_is_inside_the_data_directory(): + """It has to stay under data/ to be covered by the Docker bind mount, + so the guard cannot simply be 'anything under DATA_DIR is denied'.""" + assert os.path.realpath(AGENT_WORKSPACE_DIR).startswith( + os.path.realpath(DATA_DIR) + os.sep + ) + + +# ── An active workspace ────────────────────────────────────────────── + +def test_workspace_bound_at_the_data_directory_still_blocks_app_state(): + """vet_workspace() accepts the data directory, and chat_routes auto-binds + a workspace from a path named in the message, so this is reachable.""" + with workspace_at(DATA_DIR): + with pytest.raises(ValueError, match="application state"): + _resolve_tool_path("sessions.json") + + +def test_workspace_bound_above_the_data_directory_still_blocks_app_state(): + with workspace_at(os.path.dirname(DATA_DIR)): + with pytest.raises(ValueError, match="application state"): + _resolve_tool_path(os.path.join(DATA_DIR, "sessions.json")) + + +def test_workspace_bound_at_the_data_directory_refuses_the_empty_search_root(): + """grep/glob/ls with no path take the workspace itself as the root, which + skipped the in-workspace resolver and enumerated the state directory.""" + with workspace_at(DATA_DIR): + with pytest.raises(ValueError, match="application state"): + _resolve_search_root("") + + +def test_vet_workspace_refuses_the_data_directory(): + """Rejecting the bind is the cleaner failure: the client is told the + workspace was refused instead of every tool call erroring separately.""" + assert vet_workspace(DATA_DIR) is None + + +def test_vet_workspace_accepts_the_agent_workspace(): + os.makedirs(AGENT_WORKSPACE_DIR, exist_ok=True) + assert vet_workspace(AGENT_WORKSPACE_DIR) == os.path.realpath(AGENT_WORKSPACE_DIR) + + +def test_workspace_bound_at_the_data_directory_still_allows_the_agent_workspace(): + with workspace_at(DATA_DIR): + resolved = _resolve_tool_path(os.path.join("agent_workspace", "notes.txt")) + assert resolved == os.path.realpath(os.path.join(AGENT_WORKSPACE_DIR, "notes.txt")) + + +# ── An opt-in extra root ───────────────────────────────────────────── + +def test_extra_root_covering_the_data_directory_still_blocks_app_state(monkeypatch): + monkeypatch.setattr( + "src.settings.get_setting", lambda *_a, **_k: [os.path.dirname(DATA_DIR)] + ) + with pytest.raises(ValueError, match="application state"): + _resolve_tool_path(os.path.join(DATA_DIR, "sessions.json")) + + +# ── What the agent keeps ───────────────────────────────────────────── + +def test_allows_files_in_the_agent_workspace(): + resolved = _resolve_tool_path(os.path.join(AGENT_WORKSPACE_DIR, "scratch.txt")) + assert resolved == os.path.realpath(os.path.join(AGENT_WORKSPACE_DIR, "scratch.txt")) + + +@pytest.mark.parametrize("directory, why", [ + (UPLOAD_DIR, + "_uploaded_files_context_message emits path= and tells the model to " + "read it with read_file (src/agent_loop.py)"), + (MAIL_ATTACHMENTS_DIR, + "download_attachment returns the path and its description says to read " + "it with read_file (mcp_servers/email_server.py)"), + (PERSONAL_DIR, + "GET /api/personal returns a path per file and is reachable through the " + "app_api tool, which does not block that prefix"), + (PERSONAL_UPLOADS_DIR, + "indexed into personal docs by routes/personal_routes.py, and listed as " + "an absolute path by manage_rag"), +]) +def test_allows_user_content_the_app_hands_to_the_model(directory, why): + """Carving these out is not convenience. The app gives the model these + paths and tells it to read them, so denying them breaks the feature.""" + target = os.path.join(directory, "example.txt") + assert _resolve_tool_path(target) == os.path.realpath(target), why + + +def test_runbook_is_covered_by_the_personal_docs_carve_out(): + """RUNBOOK_DIR nests under PERSONAL_DIR, so it needs no entry of its own.""" + target = os.path.join(RUNBOOK_DIR, "notes.md") + assert _resolve_tool_path(target) == os.path.realpath(target) + + +def test_allows_tmp(): + """Unchanged: /tmp is still a root and holds no application state.""" + assert _resolve_tool_path("/tmp/scratch.txt") == os.path.realpath("/tmp/scratch.txt") + + +def test_subprocess_cwd_is_the_agent_workspace(): + """bash/python cwd has to move with the file root, or the agent writes + where read_file can no longer look.""" + assert agent_cwd() == os.path.realpath(AGENT_WORKSPACE_DIR) + + +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("relative_data", ["data", "./data"]) +def test_relative_data_dir_preserves_only_canonical_roles( + tmp_path, monkeypatch, relative_data +): + from pathlib import Path + + current_constants = importlib.import_module("src.constants") + current_execution = importlib.import_module("src.tool_execution") + monkeypatch.chdir(tmp_path) + data_dir = Path(relative_data) + data_dir.mkdir() + readable = _configure_test_data_tree(monkeypatch, data_dir) + monkeypatch.setattr(current_constants, "DATA_DIR", relative_data) + workspace = readable["AGENT_WORKSPACE_DIR"] + workspace.mkdir() + visible = workspace / "visible.txt" + visible.write_text("readable", encoding="utf-8") + protected = data_dir / "settings.json" + protected.write_text("protected", encoding="utf-8") + monkeypatch.setattr(current_execution, "_AGENT_WORKDIR", str(workspace)) + token = current_execution._active_workspace.set(None) + try: + assert set(current_execution._agent_readable_data_subdirs()) == { + os.path.realpath(path) for path in readable.values() + } + assert current_execution._resolve_search_root("") == os.path.realpath(workspace) + assert current_execution.agent_cwd() == os.path.realpath(workspace) + assert current_execution._resolve_tool_path(str(visible.resolve())) == str(visible.resolve()) + with pytest.raises(ValueError, match="application state"): + current_execution._resolve_tool_path(str(protected.resolve())) + + external_mail = tmp_path / "outside-mail" + external_mail.mkdir() + monkeypatch.setattr(current_constants, "MAIL_ATTACHMENTS_DIR", "outside-mail") + assert str(external_mail) not in current_execution._agent_readable_data_subdirs() + + # A relative override pointing to a different state role remains denied. + monkeypatch.setattr(current_constants, "MAIL_ATTACHMENTS_DIR", "data/mcp_oauth") + assert os.path.realpath("data/mcp_oauth") not in current_execution._agent_readable_data_subdirs() + 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"] + + +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 + ) + + +def test_canonical_internal_mail_attachment_directory_remains_readable( + tmp_path, monkeypatch +): + data_dir = tmp_path / "data" + data_dir.mkdir() + readable = _configure_test_data_tree(monkeypatch, data_dir) + mail_dir = readable["MAIL_ATTACHMENTS_DIR"] + mail_dir.mkdir() + attachment = mail_dir / "message.txt" + attachment.write_text("mail body\n", encoding="utf-8") + current_execution = importlib.import_module("src.tool_execution") + + assert current_execution._resolve_tool_path(str(attachment)) == os.path.realpath( + attachment + ) + + +@pytest.mark.parametrize("alias_kind", ["direct", "symlink"]) +def test_mail_attachment_root_cannot_alias_protected_state( + tmp_path, monkeypatch, alias_kind +): + data_dir = tmp_path / "data" + data_dir.mkdir() + readable = _configure_test_data_tree(monkeypatch, data_dir) + protected = data_dir / "mcp_oauth" + protected.mkdir() + secret = protected / "tokens.json" + secret.write_text("OAUTH_SECRET\n", encoding="utf-8") + current_constants = importlib.import_module("src.constants") + if alias_kind == "direct": + monkeypatch.setattr(current_constants, "MAIL_ATTACHMENTS_DIR", str(protected)) + else: + alias = readable["MAIL_ATTACHMENTS_DIR"] + try: + alias.symlink_to(protected, target_is_directory=True) + except OSError: + pytest.skip("cannot create symlink") + monkeypatch.setattr(current_constants, "MAIL_ATTACHMENTS_DIR", str(alias)) + current_execution = importlib.import_module("src.tool_execution") + + assert os.path.realpath(protected) not in current_execution._agent_readable_data_subdirs() + with pytest.raises(ValueError, match="application state"): + current_execution._resolve_tool_path(str(secret)) + + +@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, result + 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 + import threading + + readers = [] + original_thread = threading.Thread + + class TrackedThread(original_thread): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + if getattr(kwargs.get("target"), "__name__", "") == "read_stdout": + readers.append(self) + + monkeypatch.setattr(threading, "Thread", TrackedThread) + + data_dir = tmp_path / "data" + data_dir.mkdir() + _configure_test_data_tree(monkeypatch, data_dir) + (tmp_path / "visible.txt").write_text("MATCH\n", encoding="utf-8") + instances = [] + + class FakeProcess: + def __init__(self, *args, **kwargs): + self.stdout = iter(json.dumps({ + "type": "match", + "data": { + "path": {"text": "visible.txt"}, + "lines": {"text": "MATCH\n"}, + "line_number": 1, + }, + }) + "\n" for index in range(100)) + self.stderr = type("EmptyStderr", (), {"read": lambda self, _size: ""})() + 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 + assert len(readers) == 1 + assert not readers[0].is_alive(), "capped grep must release its stdout reader" + + +@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"] + + +@pytest.mark.parametrize("use_rg", [True, False]) +def test_state_spanning_grep_hides_sibling_symlink(tmp_path, monkeypatch, use_rg): + if use_rg and shutil.which("rg") is None: + pytest.skip("requires ripgrep") + data_dir = tmp_path / "data" + data_dir.mkdir() + readable = _configure_test_data_tree(monkeypatch, data_dir) + readable["AGENT_WORKSPACE_DIR"].mkdir() + protected = data_dir / "auth.txt" + protected.write_text("SIBLING_LINK_SECRET\n", encoding="utf-8") + alias = tmp_path / "public-link" + try: + alias.symlink_to(data_dir, target_is_directory=True) + except OSError: + pytest.skip("cannot create symlink") + if not use_rg: + monkeypatch.setattr(shutil, "which", lambda _name: None) + + with current_workspace_at(tmp_path): + result = asyncio.run(GrepTool().execute( + '{"pattern": "SIBLING_LINK_SECRET", "path": ""}', {} + )) + + assert result["exit_code"] == 0, result + assert "auth.txt" not in result["output"] + + +@pytest.mark.parametrize("use_rg", [True, False]) +def test_state_spanning_grep_keeps_relative_glob_semantics_in_both_modes( + tmp_path, monkeypatch, use_rg +): + if use_rg and shutil.which("rg") is None: + pytest.skip("requires ripgrep") + data_dir = tmp_path / "data[secret]" + 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("FALLBACK_MARKER public\n", encoding="utf-8") + (data_dir / "protected.py").write_text("FALLBACK_MARKER secret\n", encoding="utf-8") + if not use_rg: + monkeypatch.setattr(shutil, "which", lambda _name: None) + + with current_workspace_at(tmp_path): + result = asyncio.run(GrepTool().execute( + '{"pattern": "FALLBACK_MARKER", "path": "", "glob": "**/*.py"}', {} + )) + + assert result["exit_code"] == 0, result + assert "readable.py" in result["output"] + assert "protected.py" not in result["output"] + + +@pytest.mark.parametrize("use_rg", [True, False]) +def test_grep_reports_invalid_regex_as_error(tmp_path, monkeypatch, use_rg): + if use_rg and shutil.which("rg") is None: + pytest.skip("requires ripgrep") + data_dir = tmp_path / "data" + data_dir.mkdir() + readable = _configure_test_data_tree(monkeypatch, data_dir) + readable["AGENT_WORKSPACE_DIR"].mkdir() + if not use_rg: + monkeypatch.setattr(shutil, "which", lambda _name: None) + + with current_workspace_at(tmp_path): + result = asyncio.run(GrepTool().execute('{"pattern": "[", "path": ""}', {})) + + assert result["exit_code"] == 1 + assert any(word in result["error"].lower() for word in ("pattern", "regex")) + + +def test_no_rg_uses_top_level_spawn_worker(tmp_path, monkeypatch): + import multiprocessing + import src.agent_tools.filesystem_tools as filesystem_tools + + data_dir = tmp_path / "data" + data_dir.mkdir() + readable = _configure_test_data_tree(monkeypatch, data_dir) + workspace = readable["AGENT_WORKSPACE_DIR"] + workspace.mkdir() + (workspace / "visible.txt").write_text("FROZEN_MARKER\n", encoding="utf-8") + monkeypatch.setattr(shutil, "which", lambda _name: None) + seen = {} + + class InlineQueue(queue.Queue): + def close(self): + pass + + class InlineProcess: + exitcode = 0 + + def __init__(self, target, args): + seen["target"] = target + self.target = target + self.args = args + + def start(self): + self.target(*self.args) + + def is_alive(self): + return False + + def join(self, timeout=None): + pass + + class InlineContext: + def Queue(self, maxsize): + return InlineQueue(maxsize=maxsize) + + def Process(self, target, args): + return InlineProcess(target, args) + + def fake_get_context(method): + seen["method"] = method + return InlineContext() + + monkeypatch.setattr(multiprocessing, "get_context", fake_get_context) + + with current_workspace_at(tmp_path): + result = asyncio.run(GrepTool().execute( + '{"pattern": "FROZEN_MARKER", "path": ""}', {} + )) + + assert result["exit_code"] == 0 + assert "visible.txt" in result["output"] + assert seen["method"] == "spawn" + assert seen["target"] is filesystem_tools._python_grep_worker + + +def test_benign_hardlink_is_intentionally_rejected(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() + original = workspace / "original.txt" + alias = workspace / "copy.txt" + original.write_text("benign\n", encoding="utf-8") + try: + os.link(original, alias) + except OSError: + pytest.skip("cannot create hardlink") + + with pytest.raises(ValueError, match="hard-linked"): + importlib.import_module("src.tool_execution")._resolve_tool_path(str(alias)) + + +def test_partition_filters_skip_directories_but_explicit_root_remains_searchable( + tmp_path, monkeypatch +): + data_dir = tmp_path / "data" + data_dir.mkdir() + readable = _configure_test_data_tree(monkeypatch, data_dir) + readable["AGENT_WORKSPACE_DIR"].mkdir() + skipped = tmp_path / "node_modules" + skipped.mkdir() + (skipped / "package.txt").write_text("SKIP_POLICY_MARKER\n", encoding="utf-8") + + with current_workspace_at(tmp_path): + partitioned = asyncio.run(GrepTool().execute( + '{"pattern": "SKIP_POLICY_MARKER", "path": ""}', {} + )) + with current_workspace_at(skipped): + explicit = asyncio.run(GrepTool().execute( + '{"pattern": "SKIP_POLICY_MARKER", "path": ""}', {} + )) + + assert "package.txt" not in partitioned["output"] + assert "package.txt" in explicit["output"] + + +@pytest.mark.skipif(shutil.which("rg") is None, reason="requires ripgrep") +def test_empty_partition_still_reports_invalid_rg_regex(tmp_path, monkeypatch): + data_dir = tmp_path / "data" + data_dir.mkdir() + _configure_test_data_tree(monkeypatch, data_dir) + + with current_workspace_at(tmp_path): + result = asyncio.run(GrepTool().execute('{"pattern": "[", "path": ""}', {})) + + assert result["exit_code"] == 1 + assert "regex" in result["error"].lower() + + +def test_rg_stderr_is_fully_drained_but_only_prefix_is_reported(tmp_path, monkeypatch): + import subprocess + + target = tmp_path / "visible.txt" + target.write_text("text\n", encoding="utf-8") + chunks = ["PREFIX" + "x" * 12_000, "y" * 12_000, "TAIL"] + + class TrackingStderr: + def __init__(self): + self.reads = 0 + + def read(self, _size): + self.reads += 1 + return chunks.pop(0) if chunks else "" + + stderr = TrackingStderr() + + class FakeProcess: + stdout = iter(()) + + def __init__(self, *args, **kwargs): + self.stderr = stderr + + def poll(self): + return 2 + + def wait(self, timeout=None): + return 2 + + def terminate(self): + pass + + def kill(self): + pass + + 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": "text", "path": ""}', {})) + + assert result["exit_code"] == 1 + assert "PREFIX" in result["error"] + assert "TAIL" not in result["error"] + assert len(result["error"]) < 20_100 + assert stderr.reads == 4 + + +def test_no_rg_worker_stops_at_bounded_result_queue(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 / "many.txt").write_text("\n".join(["QUEUE_MARKER"] * 1_000)) + monkeypatch.setattr(shutil, "which", lambda _name: None) + + with current_workspace_at(tmp_path): + result = asyncio.run(GrepTool().execute( + '{"pattern": "QUEUE_MARKER", "path": "", "max_results": 3}', {} + )) + + assert result["exit_code"] == 0, result + assert result["output"].count(":QUEUE_MARKER") == 3 + assert "capped at 3 matches" in result["output"] + + +def test_no_rg_worker_is_terminated_at_deadline(tmp_path, monkeypatch): + import src.agent_tools.filesystem_tools as filesystem_tools + + 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") + monkeypatch.setattr(shutil, "which", lambda _name: None) + monkeypatch.setattr(filesystem_tools, "_GREP_TIMEOUT_SECONDS", 0.2) + + started = time.monotonic() + with current_workspace_at(tmp_path): + result = asyncio.run(GrepTool().execute( + '{"pattern": "(a+)+$", "path": ""}', {} + )) + + assert result == {"error": "grep: timed out", "exit_code": 1} + assert time.monotonic() - started < 3 + + +def test_no_rg_worker_exit_before_first_record_is_reported_promptly(tmp_path, monkeypatch): + import src.agent_tools.filesystem_tools as filesystem_tools + + data_dir = tmp_path / "data" + data_dir.mkdir() + readable = _configure_test_data_tree(monkeypatch, data_dir) + workspace = readable["AGENT_WORKSPACE_DIR"] + workspace.mkdir() + (workspace / "visible.txt").write_text("EXIT_MARKER\n", encoding="utf-8") + monkeypatch.setattr(shutil, "which", lambda _name: None) + monkeypatch.setattr(filesystem_tools, "_GREP_TIMEOUT_SECONDS", 20) + + class EmptyQueue(queue.Queue): + def close(self): + pass + + class DeadProcess: + exitcode = 71 + + def __init__(self, target, args): + self.target = target + self.args = args + + def start(self): + pass + + def is_alive(self): + return False + + def join(self, timeout=None): + pass + + class DeadContext: + def Queue(self, maxsize): + return EmptyQueue(maxsize=maxsize) + + def Process(self, target, args): + return DeadProcess(target, args) + + monkeypatch.setattr( + multiprocessing, + "get_context", + lambda method: DeadContext(), + ) + + started = time.monotonic() + with current_workspace_at(tmp_path): + result = asyncio.run(GrepTool().execute( + '{"pattern": "EXIT_MARKER", "path": ""}', {} + )) + + assert result == {"error": "grep: fallback worker exited 71", "exit_code": 1} + assert time.monotonic() - started < 1 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. diff --git a/tests/test_launcher.py b/tests/test_launcher.py index 309ad35a4..d6ab85f7c 100644 --- a/tests/test_launcher.py +++ b/tests/test_launcher.py @@ -1,12 +1,22 @@ # tests/test_launcher.py import sys import os +from pathlib import Path from unittest import mock import pytest from launcher import NullWriter, create_tray_image, on_open_browser, on_exit, open_browser +def test_frozen_multiprocessing_bootstrap_precedes_gui_and_app_imports(): + source = Path("launcher.py").read_text(encoding="utf-8") + + freeze = source.index("multiprocessing.freeze_support()") + splash = source.index("if getattr(sys, 'frozen', False):") + app_import = source.index("from app import app") + assert freeze < splash < app_import + + def test_null_writer(): writer = NullWriter() # writing and flushing should not raise any exceptions diff --git a/tests/test_tool_path_confinement.py b/tests/test_tool_path_confinement.py index c23c99750..410081fbf 100644 --- a/tests/test_tool_path_confinement.py +++ b/tests/test_tool_path_confinement.py @@ -161,12 +161,14 @@ def test_blocks_netrc(): _resolve_tool_path("~/.netrc") -def test_allows_project_data(tmp_path): - """Paths under project data/ must resolve cleanly.""" +def test_allows_agent_workspace(tmp_path): + """Paths under the agent's workspace in project data/ must resolve + cleanly. The rest of data/ is application state and is rejected; + tests/test_agent_state_dir_confinement.py covers that side.""" from src.tool_execution import _resolve_tool_path - from src.constants import DATA_DIR - target = os.path.join(DATA_DIR, "test-confinement-ok.txt") - os.makedirs(DATA_DIR, exist_ok=True) + from src.constants import AGENT_WORKSPACE_DIR + target = os.path.join(AGENT_WORKSPACE_DIR, "test-confinement-ok.txt") + os.makedirs(AGENT_WORKSPACE_DIR, exist_ok=True) with open(target, "w") as f: f.write("ok") try: