fix(security): reject inode aliases and workspace redirects

This commit is contained in:
RaresKeY
2026-08-31 14:58:03 +00:00
parent f7063b5364
commit 22c0f45e36
4 changed files with 148 additions and 9 deletions
+38 -5
View File
@@ -646,10 +646,13 @@ class GrepTool:
remaining_hits = max_hits - len(lines) remaining_hits = max_hits - len(lines)
if remaining_hits <= 0: if remaining_hits <= 0:
break 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 = [ cmd = [
rg, "--no-config", "--no-follow", "--line-number", rg, "--json", "--no-config", "--no-follow",
"--no-heading", "--color=never", "--max-count", "--max-count", str(remaining_hits),
str(remaining_hits),
] ]
if ignore_case: if ignore_case:
cmd.append("--ignore-case") cmd.append("--ignore-case")
@@ -703,8 +706,38 @@ class GrepTool:
return None, "grep: timed out" return None, "grep: timed out"
if line is None: if line is None:
break break
if line and line not in lines: if not line:
lines.append(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: finally:
if process.poll() is None: if process.poll() is None:
process.terminate() process.terminate()
+31 -3
View File
@@ -2,6 +2,7 @@
"""Initialize all application components and dependencies.""" """Initialize all application components and dependencies."""
import os import os
import logging import logging
import stat
from typing import Dict, Any from typing import Dict, Any
from src.constants import ( from src.constants import (
@@ -28,10 +29,37 @@ logger = logging.getLogger(__name__)
def create_directories(): def create_directories():
"""Create necessary directories if they don't exist.""" """Create necessary directories if they don't exist."""
for directory in (DATA_DIR, PERSONAL_DIR, RUNBOOK_DIR, UPLOAD_DIR, for directory in (DATA_DIR, PERSONAL_DIR, RUNBOOK_DIR, UPLOAD_DIR):
AGENT_WORKSPACE_DIR):
os.makedirs(directory, exist_ok=True) 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]: def initialize_managers(base_dir: str, rag_manager=None) -> Dict[str, Any]:
""" """
Initialize all manager and handler instances. Initialize all manager and handler instances.
+19 -1
View File
@@ -15,6 +15,7 @@ import logging
import os import os
import pathlib import pathlib
import re import re
import stat
import sys import sys
import time import time
from typing import Any, Awaitable, Callable, Dict, Optional, Tuple 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: def _is_denied_tool_path(resolved: str) -> bool:
"""Apply every path deny to a canonical traversal result.""" """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: def _can_traverse_tool_path(resolved: str) -> bool:
@@ -357,6 +371,8 @@ def _resolve_tool_path(raw_path: str) -> str:
raise ValueError( raise ValueError(
f"path '{raw_path}' is inside the application state directory" 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(): for root in _tool_path_roots():
if resolved == root: if resolved == root:
@@ -396,6 +412,8 @@ def _resolve_tool_path_in_workspace(workspace: str, raw_path: str) -> str:
raise ValueError( raise ValueError(
f"path '{raw_path}' is inside the application state directory" 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: if resolved != base:
# normcase so containment holds on case-insensitive filesystems # normcase so containment holds on case-insensitive filesystems
# (Windows, default macOS): it lowercases on Windows and is a no-op on # (Windows, default macOS): it lowercases on Windows and is a no-op on
+60
View File
@@ -102,6 +102,39 @@ def test_blocks_app_state_reached_through_a_symlink(tmp_path):
_resolve_tool_path(str(link / "sessions.json")) _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(): def test_blocks_app_state_on_a_case_insensitive_filesystem():
"""On default macOS a case-variant path opens the same file, and realpath """On default macOS a case-variant path opens the same file, and realpath
does not canonicalise case there the way it does on Windows. 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) 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(): def test_agent_workspace_is_inside_the_data_directory():
"""It has to stay under data/ to be covered by the Docker bind mount, """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'.""" so the guard cannot simply be 'anything under DATA_DIR is denied'."""