Compare commits

...
Author SHA1 Message Date
RaresKeY 22c0f45e36 fix(security): reject inode aliases and workspace redirects 2026-08-31 14:58:03 +00:00
RaresKeY f7063b5364 fix: bound protected filesystem searches 2026-08-31 14:19:57 +00:00
RaresKeY 51a518a061 fix: enforce state deny during recursive file search 2026-08-31 13:55:24 +00:00
nopoz 76aa8d7feb fix(security): keep agent file tools out of the app state directory
The agent's read tools (read_file, grep, glob, ls) resolved model-supplied
paths against a root list whose first entry was the whole data directory.
That directory holds the session store, the auth database, the app
encryption key and the settings file, so prompt-injected content could ask
for any of them. No approval prompt stood in the way: reads are classified
read_workspace and pass the untrusted-context gate untouched, which is
correct for reading a workspace and wrong for reading the app's own state.

The agent gets data/agent_workspace/ instead, and the subprocess cwd and
HOME move with it so bash and read_file agree on where scratch files live.

The deny itself is a property of the path, not of the root it arrived
through, because three routes reach the same bytes and closing only the
first leaves the other two working:

  - the default root list
  - a workspace bound at or above the data directory, which vet_workspace
    accepted and chat_routes auto-binds from a path named in the message
  - a tool_path_extra_roots setting covering the data directory

_resolve_search_root also returned the workspace root unchecked when the
path was empty, so a bare ls enumerated the directory whatever the deny
list said. It now resolves that case through the same guards.

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.

Four directories of user content stay readable, because the application
hands their paths to the model and tells it to open them: the chat upload
manifest, downloaded mail attachments, personal docs (which covers the
runbook) and personal uploads.
2026-08-28 11:26:49 -07:00
8 changed files with 1020 additions and 59 deletions
+2 -1
View File
@@ -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"),
]
+177 -34
View File
@@ -3,8 +3,8 @@ import json
import os
import re
import difflib
import fnmatch
import shutil
import time
from typing import Optional, Dict, Any, Tuple, List
from src.constants import MAX_READ_CHARS, MAX_DIFF_LINES, MAX_OUTPUT_CHARS
@@ -407,7 +407,11 @@ def _apply_patch_hunks(original: str, hunks: List[List[str]], label: str) -> str
class LsTool:
async def execute(self, content: str, ctx: dict) -> dict:
from src.tool_execution import _resolve_tool_path, _resolve_search_root, _truncate
from src.tool_execution import (
_is_denied_tool_path,
_resolve_search_root,
_truncate,
)
raw_path = ""
_s = (content or "").strip()
if _s.startswith("{"):
@@ -431,6 +435,8 @@ class LsTool:
for entry in it:
if entry.name.startswith("."):
continue
if _is_denied_tool_path(os.path.realpath(entry.path)):
continue
try:
is_dir = entry.is_dir(follow_symlinks=False)
size = entry.stat(follow_symlinks=False).st_size if not is_dir else 0
@@ -458,7 +464,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 +514,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 +524,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 +543,7 @@ class GlobTool:
if regex.fullmatch(rel) or regex.fullmatch(name):
# Skip deny-listed sensitive files (.env, id_rsa,
# known_hosts, …) the same way grep does.
if _is_sensitive_path(os.path.realpath(full)):
if _is_denied_tool_path(os.path.realpath(full)):
continue
try:
mtime = os.stat(full).st_mtime
@@ -559,7 +571,10 @@ class GrepTool:
async def execute(self, content: str, ctx: dict) -> dict:
from src.tool_execution import (
_SENSITIVE_FILE_PATTERNS,
_is_sensitive_path,
_agent_readable_data_subdirs,
_can_traverse_tool_path,
_is_denied_tool_path,
_path_within,
_resolve_tool_path,
_resolve_search_root,
_truncate,
@@ -592,50 +607,178 @@ class GrepTool:
import re as _re
import shutil
rg = shutil.which("rg")
from src.constants import DATA_DIR
real_root = os.path.realpath(root)
data_dir = os.path.realpath(DATA_DIR)
spans_state = _path_within(data_dir, real_root)
if spans_state and not rg:
return None, "grep: ripgrep is required when the search root contains application state"
if rg:
cmd = [rg, "--line-number", "--no-heading", "--color=never",
"--max-count", str(max_hits)]
if ignore_case:
cmd.append("--ignore-case")
if glob_pat:
cmd += ["--glob", glob_pat]
# --iglob (not --glob) so the exclusion is case-insensitive:
# on a case-insensitive filesystem "ID_RSA"/"Known_Hosts"
# resolve to the same secret as their lowercase forms, and the
# Python fallback below already folds case via _is_sensitive_path.
for _pat in _SENSITIVE_FILE_PATTERNS:
cmd += ["--iglob", f"!*{_pat}*"]
for _d in _CODENAV_SKIP_DIRS:
cmd += ["--glob", f"!**/{_d}/**"]
cmd += ["--regexp", pattern, root]
try:
import subprocess
p = subprocess.run(cmd, capture_output=True, text=True, timeout=20)
lines = [ln for ln in (p.stdout or "").splitlines() if ln][:max_hits]
return lines, None
except subprocess.TimeoutExpired:
return None, "grep: timed out"
except Exception as _e:
return None, f"grep: {_e}"
searches: list[tuple[str, list[str]]] = [(real_root, [])]
if spans_state:
searches = []
# Search everything outside DATA_DIR with a native rg glob
# exclusion. Search validated carve-outs separately so
# their contents remain available without exposing state
# siblings. --no-follow prevents a symlink from bypassing
# the excluded canonical subtree.
if real_root != data_dir:
rel_data = os.path.relpath(data_dir, real_root).replace(
os.sep, "/"
)
searches.append(
(real_root, [f"!{rel_data}", f"!{rel_data}/**"])
)
seen_roots: set[str] = set()
for readable in _agent_readable_data_subdirs():
if not _path_within(
readable, real_root
) or not os.path.exists(readable):
continue
canonical = os.path.realpath(readable)
if canonical not in seen_roots:
seen_roots.add(canonical)
searches.append((canonical, []))
lines: list[str] = []
deadline = time.monotonic() + 20
for search_root, state_excludes in searches:
remaining_hits = max_hits - len(lines)
if remaining_hits <= 0:
break
# 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, "--json", "--no-config", "--no-follow",
"--max-count", str(remaining_hits),
]
if ignore_case:
cmd.append("--ignore-case")
if glob_pat:
cmd += ["--glob", glob_pat]
# --iglob (not --glob) so the exclusion is case-insensitive:
# on a case-insensitive filesystem "ID_RSA"/"Known_Hosts"
# resolve to the same secret as their lowercase forms.
for _pat in _SENSITIVE_FILE_PATTERNS:
cmd += ["--iglob", f"!*{_pat}*"]
for _d in _CODENAV_SKIP_DIRS:
cmd += ["--glob", f"!**/{_d}/**"]
for exclusion in state_excludes:
cmd += ["--glob", exclusion]
cmd += ["--regexp", pattern, search_root]
timeout = deadline - time.monotonic()
if timeout <= 0:
return None, "grep: timed out"
try:
import queue
import subprocess
import threading
process = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
text=True,
bufsize=1,
)
except Exception as _e:
return None, f"grep: {_e}"
output: queue.Queue[Optional[str]] = queue.Queue()
def _read_stdout() -> None:
assert process.stdout is not None
try:
for line in process.stdout:
output.put(line.rstrip("\n"))
finally:
output.put(None)
threading.Thread(target=_read_stdout, daemon=True).start()
try:
while len(lines) < max_hits:
remaining = deadline - time.monotonic()
if remaining <= 0:
return None, "grep: timed out"
try:
line = output.get(timeout=remaining)
except queue.Empty:
return None, "grep: timed out"
if line is None:
break
if 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()
try:
process.wait(timeout=1)
except subprocess.TimeoutExpired:
process.kill()
process.wait()
return lines, None
try:
rx = _re.compile(pattern, _re.IGNORECASE if ignore_case else 0)
except _re.error as _e:
return None, f"grep: bad pattern: {_e}"
glob_rx = _glob_to_regex(glob_pat.replace("\\", "/")) if glob_pat else None
hits = []
if os.path.isfile(root):
file_iter = [root]
else:
file_iter = []
for dp, dns, fns in os.walk(root):
dns[:] = [d for d in dns if d not in _CODENAV_SKIP_DIRS]
if not _can_traverse_tool_path(os.path.realpath(dp)):
dns[:] = []
continue
dns[:] = [
d for d in dns
if d not in _CODENAV_SKIP_DIRS
and _can_traverse_tool_path(os.path.realpath(os.path.join(dp, d)))
]
for fn in fns:
if glob_pat and not fnmatch.fnmatch(fn, glob_pat):
rel = os.path.relpath(os.path.join(dp, fn), root).replace(
os.sep, "/"
)
if glob_rx and not (
glob_rx.fullmatch(rel) or glob_rx.fullmatch(fn)
):
continue
file_iter.append(os.path.join(dp, fn))
for fp in file_iter:
if len(hits) >= max_hits:
break
if _is_sensitive_path(os.path.realpath(fp)):
if _is_denied_tool_path(os.path.realpath(fp)):
continue
try:
with open(fp, "r", encoding="utf-8", errors="strict") as f:
+30 -1
View File
@@ -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
@@ -31,6 +32,34 @@ def create_directories():
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.
+5
View File
@@ -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
+213 -17
View File
@@ -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 <repo_root>/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 <repo_root>/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,169 @@ 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=<p>" 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, False),
(UPLOAD_DIR, False),
# This has a documented environment override and may legitimately
# live outside DATA_DIR, but it must never equal/contain DATA_DIR.
(MAIL_ATTACHMENTS_DIR, True),
(PERSONAL_DIR, False),
(PERSONAL_UPLOADS_DIR, False),
)
data_dir = os.path.realpath(DATA_DIR)
safe: list[str] = []
for raw, external_ok in configured:
value = str(raw or "").strip()
# These paths are security-policy roots, not ordinary allowlist
# entries. Accept only explicit absolute directory paths. State
# carve-outs must be strict DATA_DIR descendants; the documented mail
# override may also be disjoint. Empty/dot, filesystem-root, ancestor,
# equality, file, or symlink-equivalent settings fail closed.
if not value or not os.path.isabs(os.path.expanduser(value)):
continue
resolved = os.path.realpath(os.path.expanduser(value))
if os.path.exists(resolved) and not os.path.isdir(resolved):
continue
inside_data = resolved != data_dir and _path_within(resolved, data_dir)
external_safe = (
external_ok
and resolved != data_dir
and os.path.dirname(resolved) != resolved
and not _path_within(data_dir, resolved)
and not _path_within(resolved, data_dir)
)
if not (inside_data or external_safe) or _is_sensitive_path(resolved):
continue
safe.append(resolved)
return tuple(safe)
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 +297,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 +367,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 +408,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 +463,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
@@ -304,16 +494,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__)
+574
View File
@@ -0,0 +1,574 @@
"""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 os
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_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(
"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
)
@pytest.mark.skipif(
os.path.normcase("DATA") == os.path.normcase("data"),
reason="requires a platform with case-sensitive path comparison",
)
def test_case_distinct_path_cannot_masquerade_as_readable_descendant(
tmp_path, monkeypatch
):
data_dir = tmp_path / "data"
data_dir.mkdir()
_configure_test_data_tree(monkeypatch, data_dir)
distinct = tmp_path / "DATA" / "agent_workspace"
distinct.mkdir(parents=True)
current_execution = importlib.import_module("src.tool_execution")
monkeypatch.setattr(current_execution, "AGENT_WORKSPACE_DIR", str(distinct))
assert (
os.path.realpath(distinct)
not in current_execution._agent_readable_data_subdirs()
)
@pytest.mark.parametrize("use_workspace", [True, False])
def test_ls_hides_protected_entries_when_root_contains_data(
tmp_path, monkeypatch, use_workspace
):
data_dir = tmp_path / "data"
data_dir.mkdir()
_configure_test_data_tree(monkeypatch, data_dir)
(tmp_path / "visible.txt").write_text("visible\n", encoding="utf-8")
secret = data_dir / "settings.json"
secret.write_text("SECRET_WITH_SIZE\n", encoding="utf-8")
if use_workspace:
context = current_workspace_at(tmp_path)
content = '{"path": ""}'
else:
monkeypatch.setattr("src.settings.get_setting", lambda *_a, **_k: [str(tmp_path)])
context = nullcontext()
content = f'{{"path": "{tmp_path}"}}'
with context:
result = asyncio.run(LsTool().execute(content, {}))
assert "visible.txt" in result["output"]
assert "settings.json" not in result["output"]
assert "SECRET_WITH_SIZE" not in result["output"]
assert "data/" not in result["output"]
@pytest.mark.skipif(shutil.which("rg") is None, reason="requires ripgrep")
def test_state_spanning_grep_bounds_dangerous_regex(tmp_path, monkeypatch):
data_dir = tmp_path / "data"
data_dir.mkdir()
readable = _configure_test_data_tree(monkeypatch, data_dir)
workspace = readable["AGENT_WORKSPACE_DIR"]
workspace.mkdir()
(workspace / "long.txt").write_text("a" * 250_000 + "!\n", encoding="utf-8")
(data_dir / "auth.txt").write_text("a" * 250_000 + "!\n", encoding="utf-8")
started = time.monotonic()
with current_workspace_at(tmp_path):
result = asyncio.run(GrepTool().execute(
'{"pattern": "(a+)+$", "path": "", "max_results": 1}', {}
))
elapsed = time.monotonic() - started
assert result["exit_code"] == 0
assert "auth.txt" not in result["output"]
assert elapsed < 5
def test_state_spanning_grep_stops_process_at_max_results(tmp_path, monkeypatch):
import subprocess
data_dir = tmp_path / "data"
data_dir.mkdir()
_configure_test_data_tree(monkeypatch, data_dir)
instances = []
class FakeProcess:
def __init__(self, *args, **kwargs):
self.stdout = iter(
f"{tmp_path}/visible-{index}.txt:1:MATCH\n" for index in range(100)
)
self.terminated = False
instances.append(self)
def poll(self):
return 0 if self.terminated else None
def terminate(self):
self.terminated = True
def wait(self, timeout=None):
return 0
def kill(self):
self.terminated = True
monkeypatch.setattr(shutil, "which", lambda _name: "/usr/bin/rg")
monkeypatch.setattr(subprocess, "Popen", FakeProcess)
with current_workspace_at(tmp_path):
result = asyncio.run(GrepTool().execute(
'{"pattern": "MATCH", "path": "", "max_results": 1}', {}
))
assert len(instances) == 1
assert instances[0].terminated is True
assert result["output"].count(":1:MATCH") == 1
@pytest.mark.skipif(shutil.which("rg") is None, reason="requires ripgrep")
def test_state_spanning_grep_keeps_relative_glob_semantics(tmp_path, monkeypatch):
data_dir = tmp_path / "data"
data_dir.mkdir()
readable = _configure_test_data_tree(monkeypatch, data_dir)
nested = readable["AGENT_WORKSPACE_DIR"] / "nested"
nested.mkdir(parents=True)
(nested / "readable.py").write_text("PATH_GLOB_MARKER\n", encoding="utf-8")
(data_dir / "protected.py").write_text("PATH_GLOB_MARKER\n", encoding="utf-8")
with current_workspace_at(tmp_path):
result = asyncio.run(GrepTool().execute(
'{"pattern": "PATH_GLOB_MARKER", "path": "", "glob": "**/*.py"}',
{},
))
assert "readable.py" in result["output"]
assert "protected.py" not in result["output"]
+11
View File
@@ -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.
+7 -5
View File
@@ -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: