From 22c0f45e36427e27fd7a39d7925eed148cf37315 Mon Sep 17 00:00:00 2001 From: RaresKeY <158580472+RaresKeY@users.noreply.github.com> Date: Mon, 31 Aug 2026 14:58:03 +0000 Subject: [PATCH] fix(security): reject inode aliases and workspace redirects --- src/agent_tools/filesystem_tools.py | 43 ++++++++++++++-- src/app_initializer.py | 34 +++++++++++-- src/tool_execution.py | 20 +++++++- tests/test_agent_state_dir_confinement.py | 60 +++++++++++++++++++++++ 4 files changed, 148 insertions(+), 9 deletions(-) diff --git a/src/agent_tools/filesystem_tools.py b/src/agent_tools/filesystem_tools.py index c5643b629..5e93730fb 100644 --- a/src/agent_tools/filesystem_tools.py +++ b/src/agent_tools/filesystem_tools.py @@ -646,10 +646,13 @@ class GrepTool: remaining_hits = max_hits - len(lines) if remaining_hits <= 0: break + # JSON output gives us the canonical match pathname so it + # can be revalidated before any line reaches the model. + # This is required for hardlink aliases inside an allowed + # workspace; lexical/path checks alone cannot see them. cmd = [ - rg, "--no-config", "--no-follow", "--line-number", - "--no-heading", "--color=never", "--max-count", - str(remaining_hits), + rg, "--json", "--no-config", "--no-follow", + "--max-count", str(remaining_hits), ] if ignore_case: cmd.append("--ignore-case") @@ -703,8 +706,38 @@ class GrepTool: return None, "grep: timed out" if line is None: break - if line and line not in lines: - lines.append(line) + if not line: + continue + try: + event = json.loads(line) + except (TypeError, json.JSONDecodeError): + # Keep lightweight/fake runners compatible with + # the historical plain `path:line:text` stream; + # still revalidate the path before exposing it. + pieces = line.split(":", 2) + if len(pieces) >= 3: + plain_path = pieces[0] + if not _is_denied_tool_path(os.path.realpath(plain_path)): + if line not in lines: + lines.append(line) + continue + if event.get("type") != "match": + continue + match = event.get("data") or {} + path_data = match.get("path") or {} + match_path = path_data.get("text") + if not match_path: + continue + if _is_denied_tool_path(os.path.realpath(match_path)): + continue + line_text = (match.get("lines") or {}).get("text", "") + line_number = match.get("line_number", "?") + rendered = ( + f"{match_path}:{line_number}:" + f"{line_text.rstrip()[:_CODENAV_MAX_LINE]}" + ) + if rendered not in lines: + lines.append(rendered) finally: if process.poll() is None: process.terminate() diff --git a/src/app_initializer.py b/src/app_initializer.py index ba4d81d30..aebb14e1e 100644 --- a/src/app_initializer.py +++ b/src/app_initializer.py @@ -2,6 +2,7 @@ """Initialize all application components and dependencies.""" import os import logging +import stat from typing import Dict, Any from src.constants import ( @@ -28,10 +29,37 @@ logger = logging.getLogger(__name__) def create_directories(): """Create necessary directories if they don't exist.""" - for directory in (DATA_DIR, PERSONAL_DIR, RUNBOOK_DIR, UPLOAD_DIR, - AGENT_WORKSPACE_DIR): + 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(DATA_DIR) + workspace = os.path.abspath(os.path.expanduser(AGENT_WORKSPACE_DIR)) + try: + if os.path.commonpath([workspace, data_root]) != data_root or workspace == data_root: + raise RuntimeError("agent workspace must resolve inside DATA_DIR") + except ValueError as exc: + raise RuntimeError("agent workspace must resolve inside DATA_DIR") from exc + 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) + try: + inside = os.path.commonpath([resolved_workspace, data_root]) == data_root + except ValueError: + inside = False + if resolved_workspace == data_root or not inside: + raise RuntimeError("agent workspace must resolve inside 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/tool_execution.py b/src/tool_execution.py index 5057ba2c7..527956349 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 @@ -259,9 +260,22 @@ def _is_app_state_path(resolved: str) -> bool: ) +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) + 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: @@ -357,6 +371,8 @@ def _resolve_tool_path(raw_path: str) -> str: 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: @@ -396,6 +412,8 @@ def _resolve_tool_path_in_workspace(workspace: str, raw_path: str) -> str: 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 diff --git a/tests/test_agent_state_dir_confinement.py b/tests/test_agent_state_dir_confinement.py index 76042a19d..91e88d6e5 100644 --- a/tests/test_agent_state_dir_confinement.py +++ b/tests/test_agent_state_dir_confinement.py @@ -102,6 +102,39 @@ def test_blocks_app_state_reached_through_a_symlink(tmp_path): _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. @@ -120,6 +153,33 @@ def test_default_search_root_is_the_agent_workspace(): 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_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'."""