feat(dangerous-mode): implement real-filesystem access with safety ch… (#276)

* feat(dangerous-mode): implement real-filesystem access with safety checks

- Introduced a 'dangerous mode' allowing the agent to operate on the real filesystem.
- Updated command validation to bypass path confinement while enforcing a blocklist for privileged commands.
- Added warnings and guidelines for users when operating in dangerous mode.
- Enhanced configuration to support dangerous mode and ensure it implies auto-approval.
- Updated tests to verify the behavior of commands and configurations in dangerous mode.

* feat(dangerous-mode): enhance logging and environment management for dangerous mode

* feat(dangerous-mode): improve handling of dangerous mode with environment flags and enhance test isolation
This commit is contained in:
Xi Zhang
2026-06-10 18:19:13 +01:00
committed by GitHub
parent fc05b5eed2
commit c02be519f6
16 changed files with 509 additions and 47 deletions
+9
View File
@@ -274,11 +274,16 @@ def _load_mcp_tools_cached(on_progress=None) -> dict[str, list]:
def _configured_system_prompt(cfg) -> str:
memory_controls = MemoryControls.from_config(cfg)
# In dangerous mode the agent works on the real filesystem; give it the real
# cwd so it can use absolute paths instead of the virtual `/` workspace root.
real_cwd = str(_paths_mod.resolve_virtual_path("/")) if cfg.dangerous_mode else None
return get_system_prompt(
enable_observation_memory=memory_controls.observations_enabled,
enable_observation_writes=memory_controls.observation_tool_enabled(
MemoryObservationTarget.AGENT
),
dangerous=cfg.dangerous_mode,
cwd=real_cwd,
)
@@ -600,10 +605,13 @@ def _get_default_backend():
user_skills_dir = str(_paths_mod.USER_SKILLS_DIR)
global_skills_dir = str(_paths_mod.GLOBAL_SKILLS_DIR)
# Dangerous mode opens the workspace (`/`) route to the real filesystem;
# the /skills/ and /memories/ routes stay confined (virtual_mode=True).
ws_backend = CustomSandboxBackend(
root_dir=workspace_dir,
virtual_mode=True,
timeout=cfg.sandbox_execute_timeout,
dangerous=cfg.dangerous_mode,
)
sk_backend = MergedSkillsBackend(
primary_dir=user_skills_dir,
@@ -919,6 +927,7 @@ def create_cli_agent(
root_dir=workspace_dir,
virtual_mode=True,
timeout=cfg.sandbox_execute_timeout,
dangerous=cfg.dangerous_mode,
)
sk_backend = MergedSkillsBackend(
primary_dir=_usr_skills_dir,
+77 -31
View File
@@ -43,10 +43,14 @@ _SYSTEM_PATH_PREFIXES = (
"/root/",
)
# Dangerous patterns that could escape the workspace
BLOCKED_PATTERNS = [
# Path-confinement patterns: keep the agent inside the workspace. These are
# bypassed in dangerous mode (real-filesystem access).
_PATH_PATTERNS = [
r"~/", # home directory
r"\bcd\s+/", # cd to absolute path
]
# Destructive patterns: catastrophic regardless of mode — always enforced.
_DESTRUCTIVE_PATTERNS = [
r"\brm\s+-rf\s+/", # rm -rf with absolute path
]
@@ -499,6 +503,8 @@ def _extract_all_paths(
def validate_command(
command: str,
allow_prefixes: tuple[str, ...] = (),
*,
dangerous: bool = False,
) -> str | None:
"""
Validate a shell command for safety.
@@ -507,20 +513,35 @@ def validate_command(
command: Shell command string.
allow_prefixes: Absolute path prefixes exempt from the system-path
block list (matching rules in ``_is_under_allowed_prefix``).
dangerous: When True (real-filesystem mode), skip the path-confinement
checks (``..`` traversal, ``~/``/``cd /`` patterns, absolute system
paths). Privileged commands (:data:`BLOCKED_COMMANDS`) and
catastrophic patterns (:data:`_DESTRUCTIVE_PATTERNS`) are still
enforced.
Returns:
None if command is safe, error message string if blocked.
"""
# Check for '..' path traversal as a path component
if _has_traversal_component(command):
return (
"Command blocked: contains '..' path traversal. "
"All commands must operate within the workspace directory. "
"Use relative paths (e.g., './file.py') instead."
)
# Path-confinement checks — skipped in dangerous mode.
if not dangerous:
# Check for '..' path traversal as a path component
if _has_traversal_component(command):
return (
"Command blocked: contains '..' path traversal. "
"All commands must operate within the workspace directory. "
"Use relative paths (e.g., './file.py') instead."
)
# Check for dangerous patterns
for pattern in BLOCKED_PATTERNS:
for pattern in _PATH_PATTERNS:
if re.search(pattern, command):
return (
f"Command blocked: contains forbidden pattern '{pattern}'. "
f"All commands must operate within the workspace directory. "
f"Use relative paths (e.g., './file.py') instead."
)
# Catastrophic patterns (e.g. `rm -rf /`) — always enforced.
for pattern in _DESTRUCTIVE_PATTERNS:
if re.search(pattern, command):
return (
f"Command blocked: contains forbidden pattern '{pattern}'. "
@@ -528,7 +549,7 @@ def validate_command(
f"Use relative paths (e.g., './file.py') instead."
)
# Check for dangerous commands (pipeline-aware)
# Check for dangerous commands (pipeline-aware) — always enforced.
for base_cmd in _split_shell_commands(command):
if base_cmd in BLOCKED_COMMANDS:
return (
@@ -536,16 +557,17 @@ def validate_command(
f"Only standard development commands are permitted."
)
# Check for absolute system paths (including inside quoted strings).
# This catches attacks like: python -c "os.remove('/Users/foo/file')"
escaped_paths = _extract_all_paths(command, allow_prefixes=allow_prefixes)
if escaped_paths:
path_sample = escaped_paths[0]
return (
f"Command blocked: contains absolute system path '{path_sample}'. "
f"All file operations must use relative paths within the workspace. "
f"Use relative paths (e.g., './file.py') instead."
)
# Absolute-system-path check — skipped in dangerous mode.
# Catches attacks like: python -c "os.remove('/Users/foo/file')"
if not dangerous:
escaped_paths = _extract_all_paths(command, allow_prefixes=allow_prefixes)
if escaped_paths:
path_sample = escaped_paths[0]
return (
f"Command blocked: contains absolute system path '{path_sample}'. "
f"All file operations must use relative paths within the workspace. "
f"Use relative paths (e.g., './file.py') instead."
)
return None
@@ -828,7 +850,7 @@ class MergedSkillsBackend(BackendProtocol):
def prepare_sandbox_command(
command: str, cwd: str | Path, *, virtual_mode: bool = True
command: str, cwd: str | Path, *, virtual_mode: bool = True, dangerous: bool = False
) -> tuple[str, str | None]:
"""Normalize workspace paths in ``command`` and validate it for the sandbox.
@@ -849,10 +871,13 @@ def prepare_sandbox_command(
# Replace literal workspace-root absolute paths with ./ after SSH masking so
# remote paths that happen to contain the local cwd are preserved, and before
# validation so local workspace paths are sanitized before the system-path
# check fires.
ws = cwd_str + "/"
if ws in command:
command = command.replace(ws, "./")
# check fires. Skipped in dangerous mode: there is no virtual workspace, the
# agent uses real absolute paths, and rewriting would corrupt any argument
# (echo text, grep/git pattern) that merely contains the cwd string.
if not dangerous:
ws = cwd_str + "/"
if ws in command:
command = command.replace(ws, "./")
if virtual_mode:
command = convert_virtual_paths_in_command(
command=command,
@@ -866,7 +891,9 @@ def prepare_sandbox_command(
str(paths.MEMORIES_DIR),
str(_BUILTIN_SKILLS_DIR),
)
error = validate_command(command, allow_prefixes=allow_prefixes)
error = validate_command(
command, allow_prefixes=allow_prefixes, dangerous=dangerous
)
if error:
return command, error
return _restore_spans(command, ssh_replacements), None
@@ -893,6 +920,7 @@ class CustomSandboxBackend(LocalShellBackend):
max_output_bytes: int = 100_000,
env: dict[str, str] | None = None,
inherit_env: bool = True,
dangerous: bool = False,
):
"""
Initialize custom sandbox backend.
@@ -904,7 +932,16 @@ class CustomSandboxBackend(LocalShellBackend):
max_output_bytes: Max output size before truncation (default 100KB)
env: Extra environment variables for subprocess
inherit_env: Whether to inherit parent process env (default True)
dangerous: Real-filesystem mode — the agent operates on real absolute
paths anywhere on disk (no workspace confinement). Forces
``virtual_mode=False`` and relaxes path validation while keeping
the privileged-command blocklist. Defaults to False.
"""
self._dangerous = dangerous
if dangerous:
# Real paths require the legacy (non-virtual) resolution path so the
# parent backend returns absolute paths as-is.
virtual_mode = False
super().__init__(
root_dir=root_dir,
virtual_mode=virtual_mode,
@@ -927,7 +964,13 @@ class CustomSandboxBackend(LocalShellBackend):
2. /<ws_name>/file.py → /file.py
3. /Users/name/.../<ws_name>/f → /f (strip at LAST <ws_name>/)
4. /Users/name/file.py → /file.py (keep basename)
In dangerous (real-filesystem) mode, skip all rewriting and let the
parent resolve real absolute paths as-is.
"""
if self._dangerous:
return super()._resolve_path(key)
cwd_str = str(self.cwd).rstrip("/")
ws_name = Path(cwd_str).name # e.g. "workspace", "my-project"
@@ -976,7 +1019,7 @@ class CustomSandboxBackend(LocalShellBackend):
Then delegates to LocalShellBackend.execute() for actual execution.
"""
command, error = prepare_sandbox_command(
command, self.cwd, virtual_mode=self.virtual_mode
command, self.cwd, virtual_mode=self.virtual_mode, dangerous=self._dangerous
)
if error:
return ExecuteResponse(output=error, exit_code=1, truncated=False)
@@ -988,7 +1031,10 @@ class CustomSandboxBackend(LocalShellBackend):
if response.exit_code == 124:
cmd_words = command.split()
grep_hint = cmd_words[0] if cmd_words else "process"
bg_cmd = f'{command} > /output.log 2>&1 & echo "PID: $!"'
# In dangerous mode `/` is the host root; use a workspace-relative
# log path so the suggested command doesn't fail or write to `/`.
output_log = "./output.log" if self._dangerous else "/output.log"
bg_cmd = f'{command} > {output_log} 2>&1 & echo "PID: $!"'
response = ExecuteResponse(
output=(
f"{response.output}\n\n"
@@ -998,7 +1044,7 @@ class CustomSandboxBackend(LocalShellBackend):
f" 2. Runs indefinitely? Run it in the background and keep the PID:\n"
f" {bg_cmd}\n"
f" Check: ps -p <PID> (or: ps aux | grep {grep_hint}) · "
f"Read: cat /output.log · Stop: kill <PID>"
f"Read: cat {output_log} · Stop: kill <PID>"
),
exit_code=response.exit_code,
truncated=response.truncated,
+7
View File
@@ -11,6 +11,13 @@ def _agent_name() -> str:
return AGENT_NAME
# Dangerous-mode warning banner — shared by Rich CLI, Textual TUI, and serve so
# the wording never drifts. Label is rendered white-on-red, message in red.
DANGEROUS_BANNER_LABEL = "DANGEROUS MODE"
DANGEROUS_BANNER_MESSAGE = (
"Real-filesystem access • the agent can read/write/delete anywhere."
)
WELCOME_SLOGANS = [
"Ready for vibe research? What do you want cooking?",
"Science doesn't sleep. Neither do your sub-agents.",
+21
View File
@@ -1322,6 +1322,11 @@ def serve(
"--ask-user",
help="Enable agent to ask clarifying questions about your research preferences",
),
dangerous: bool = typer.Option(
False,
"--dangerous",
help="DANGEROUS: real-filesystem access (no workspace confinement); implies --auto-approve",
),
debug: bool = typer.Option(
False,
"--debug",
@@ -1344,6 +1349,8 @@ def serve(
cli_overrides["enable_ask_user"] = False
elif ask_user:
cli_overrides["enable_ask_user"] = True
if dangerous:
cli_overrides["dangerous_mode"] = True
if debug:
cli_overrides["log_level"] = "DEBUG"
cli_overrides["channel_debug_tracing"] = True
@@ -1390,6 +1397,13 @@ def serve(
# async sub-agents inherit the CLI's workspace via EVOSCIENTIST_WORKSPACE_DIR).
_ensure_async_subagent_server(config, workspace_dir=ws)
if config.dangerous_mode:
from ._constants import DANGEROUS_BANNER_LABEL, DANGEROUS_BANNER_MESSAGE
console.print(
f"[bold white on red] ⚠ {DANGEROUS_BANNER_LABEL} [/bold white on red] "
f"[bold red]{DANGEROUS_BANNER_MESSAGE}[/bold red]"
)
console.print("[dim]Loading agent...[/dim]")
agent = _load_agent(workspace_dir=ws, config=config)
from ..sessions import generate_thread_id
@@ -1964,6 +1978,11 @@ def _main_callback(
"--ask-user",
help="Enable agent to ask clarifying questions about your research preferences",
),
dangerous: bool = typer.Option(
False,
"--dangerous",
help="DANGEROUS: real-filesystem access (no workspace confinement); implies --auto-approve",
),
auth_mode: str | None = typer.Option(
None,
"--auth-mode",
@@ -2001,6 +2020,8 @@ def _main_callback(
cli_overrides["enable_ask_user"] = False
elif ask_user:
cli_overrides["enable_ask_user"] = True
if dangerous:
cli_overrides["dangerous_mode"] = True
if auth_mode:
if auth_mode not in ("api_key", "oauth"):
raise typer.BadParameter("--auth-mode must be 'api_key' or 'oauth'")
+24 -1
View File
@@ -44,7 +44,14 @@ from ..sessions import (
from ..stream.console import console
from ..stream.display import _fix_markdown_heading_spacing
from ._agent_loader import BackgroundAgentLoader, MCPProgressTracker
from ._constants import LOGO_GRADIENT, LOGO_LINES, WELCOME_SLOGANS, build_metadata
from ._constants import (
DANGEROUS_BANNER_LABEL,
DANGEROUS_BANNER_MESSAGE,
LOGO_GRADIENT,
LOGO_LINES,
WELCOME_SLOGANS,
build_metadata,
)
from .agent import _create_session_workspace, _load_agent, _shorten_path
from .channel import (
ChannelMessage,
@@ -141,6 +148,22 @@ def print_banner(
info.append(" \u2022 Ctrl+C ", style="#ffe082")
info.append("interrupt", style="#ffe082 bold")
console.print(info)
print_dangerous_warning()
def print_dangerous_warning() -> None:
"""Print an unmissable warning when dangerous (real-filesystem) mode is on."""
try:
from ..config import get_effective_config
if not get_effective_config().dangerous_mode:
return
except Exception:
return
warn = Text()
warn.append(f"\n \u26a0 {DANGEROUS_BANNER_LABEL}", style="bold white on red")
warn.append(f" {DANGEROUS_BANNER_MESSAGE}", style="bold red")
console.print(warn)
# =============================================================================
+18 -1
View File
@@ -36,7 +36,14 @@ from ..sessions import (
from ..stream.events import stream_agent_events
from ..stream.state import ResearchPhase, StreamState
from ._agent_loader import BackgroundAgentLoader, MCPProgressTracker
from ._constants import LOGO_GRADIENT, LOGO_LINES, WELCOME_SLOGANS, build_metadata
from ._constants import (
DANGEROUS_BANNER_LABEL,
DANGEROUS_BANNER_MESSAGE,
LOGO_GRADIENT,
LOGO_LINES,
WELCOME_SLOGANS,
build_metadata,
)
from .channel import (
ChannelMessage,
_auto_start_channel,
@@ -139,6 +146,16 @@ def _build_welcome_banner(
info.append("@ files", style="#ffe082 bold")
info.append(" \u2022 Ctrl+C ", style="#ffe082")
info.append("interrupt", style="#ffe082 bold")
try:
from ..config import get_effective_config
if get_effective_config().dangerous_mode:
info.append(
f"\n \u26a0 {DANGEROUS_BANNER_LABEL}", style="bold white on red"
)
info.append(f" {DANGEROUS_BANNER_MESSAGE}", style="bold red")
except Exception:
pass
banner.append_text(info)
slogan = Text(f"\n {random.choice(WELCOME_SLOGANS)}", style="dim italic")
+22
View File
@@ -332,6 +332,11 @@ class EvoScientistConfig:
auto_mode: bool = False # Run unattended: imply auto_approve and disable ask_user
shell_allow_list: str = "" # Comma-separated shell command prefixes to auto-approve
# Dangerous mode: real-filesystem access (no workspace confinement). The agent
# operates on real absolute paths anywhere on disk; the privileged-command
# blocklist (sudo/chmod/dd/...) still applies. Implies auto_approve.
dangerous_mode: bool = False
# Agent features
enable_ask_user: bool = True # Enable ask_user tool for agent-initiated questions
@@ -386,6 +391,12 @@ class EvoScientistConfig:
)
self.sandbox_execute_timeout = 300
# Dangerous mode implies auto_approve regardless of source (CLI, env,
# config file). Mirrors how auto_mode implies auto_approve — done here so
# the coupling holds even when dangerous_mode is set via `config set`.
if self.dangerous_mode:
self.auto_approve = True
try:
writer = MemoryObservationWriter(
str(self.memory_observation_writer).strip().lower()
@@ -636,6 +647,7 @@ _ENV_MAPPINGS = {
"openrouter_anthropic_prompt_cache": (
"EVOSCIENTIST_OPENROUTER_ANTHROPIC_PROMPT_CACHE"
),
"dangerous_mode": "EVOSCIENTIST_DANGEROUS_MODE",
"channel_debug_tracing": "EVOSCIENTIST_CHANNEL_DEBUG_TRACING",
"ccproxy_port": "EVOSCIENTIST_CCPROXY_PORT",
"use_responses_api": "EVOSCIENTIST_USE_RESPONSES_API",
@@ -761,6 +773,16 @@ def apply_config_to_env(config: EvoScientistConfig) -> None:
"EVOSCIENTIST_OPENROUTER_ANTHROPIC_PROMPT_CACHE"
):
os.environ["EVOSCIENTIST_OPENROUTER_ANTHROPIC_PROMPT_CACHE"] = "true"
# Round-trip dangerous_mode to env so it survives a fresh get_effective_config()
# (warning banner, run_in_background) and is inherited by the langgraph dev
# subprocess — otherwise a --dangerous CLI flag (not persisted to file/env)
# is invisible to those consumers while the backend is already unconfined.
# Bidirectional: clear it when off so a re-apply with a lower config (or a
# stale value) can't leave the process stuck in dangerous mode.
if config.dangerous_mode:
os.environ["EVOSCIENTIST_DANGEROUS_MODE"] = "true"
else:
os.environ.pop("EVOSCIENTIST_DANGEROUS_MODE", None)
if config.use_responses_api and not os.environ.get(
"EVOSCIENTIST_USE_RESPONSES_API"
):
+10
View File
@@ -144,6 +144,16 @@ def deploy(
border_style="cyan",
)
)
if config.dangerous_mode:
from ..cli._constants import (
DANGEROUS_BANNER_LABEL,
DANGEROUS_BANNER_MESSAGE,
)
console.print(
f"[bold white on red] ⚠ {DANGEROUS_BANNER_LABEL} [/bold white on red] "
f"[bold red]{DANGEROUS_BANNER_MESSAGE}[/bold red]"
)
# 6. ccproxy lifecycle (only if any provider uses OAuth)
_ccproxy_proc = None
+19 -2
View File
@@ -75,9 +75,19 @@ def run_in_background(
name: Optional short label to recognize the process later.
"""
cwd = str(paths.resolve_virtual_path("/"))
# Honor dangerous mode so background commands match `execute`'s policy
# (real-filesystem access, no virtual-path rewriting). Read the env flag that
# apply_config_to_env round-trips at startup (and the subprocess inherits) —
# cheaper than reloading the full config from disk on every launch, and uses
# the same truthy parsing as every other bool env flag.
from ..llm.models import _env_flag_enabled
dangerous = _env_flag_enabled("EVOSCIENTIST_DANGEROUS_MODE")
# Same path-rewriting + validation as execute (shared helper) so virtual paths
# resolve to the workspace and the command can't bypass the sandbox checks.
command, error = prepare_sandbox_command(command, cwd)
command, error = prepare_sandbox_command(
command, cwd, virtual_mode=not dangerous, dangerous=dangerous
)
if error:
return error
tid = _origin_thread_id(runtime)
@@ -85,9 +95,16 @@ def run_in_background(
command, cwd, name, origin_thread_id=tid, on_exit=lambda p: _notify_done(p, tid)
)
label = f" (name={name!r})" if name else ""
# In dangerous mode `/` is the real root, so advertise the real log path;
# in virtual mode `/.bg_processes/...` correctly maps to the workspace.
log_path = (
f"{cwd}/.bg_processes/{process_id}.log"
if dangerous
else f"/.bg_processes/{process_id}.log"
)
return (
f"Started background process {process_id}{label}. "
f"Output -> /.bg_processes/{process_id}.log. "
f"Output -> {log_path}. "
f"Poll with check_process('{process_id}'), stop with stop_process('{process_id}')."
)
+57 -12
View File
@@ -268,13 +268,11 @@ WRITING_GUIDELINES = """# Writing Guidelines
# cfg.sandbox_execute_timeout (CustomSandboxBackend); this number is just the
# documented default, and the per-command `timeout` override is the mechanism
# that matters to the agent.
SHELL_GUIDELINES = """# Shell Execution Guidelines
When using the `execute` tool for shell commands:
**Sandbox limits**: Commands default to a 300s timeout (a deployment may override this default) and 100 KB output. For a known long command (e.g. a download), pass `timeout` (up to 3600s): `execute(command="wget ...", timeout=600)`. For unbounded tasks, use background execution (below).
**Short commands** (< 30 seconds): Run directly
# Mode-independent core of the shell guidelines. ``{log_path}`` is the manual-
# background redirect target: virtual ``/output.log`` (sandbox) or real
# ``./output.log`` (dangerous mode, where ``/`` is the host root).
_SHELL_GUIDELINES_CORE = """**Short commands** (< 30 seconds): Run directly
```bash
python script.py
pip install pandas
@@ -284,14 +282,48 @@ pip install pandas
If you must background manually instead, you MUST redirect output to a file (otherwise the call blocks) and capture the PID:
```bash
python long_task.py > /output.log 2>&1 &
echo "PID: $!" # check: ps -p <PID> · stop: kill <PID> · read: cat /output.log
python long_task.py > {log_path} 2>&1 &
echo "PID: $!" # check: ps -p <PID> · stop: kill <PID> · read: cat {log_path}
```
**Before heavy compute**: Estimate runtime. If likely > 5 minutes, use background execution from the start. If GPU memory is uncertain, start with a small test run (1 epoch, small batch) before the full run.
This prevents blocking the conversation during long operations.
"""
This prevents blocking the conversation during long operations."""
# Sandbox (default) header: virtual `/` workspace.
_SHELL_GUIDELINES_SANDBOX_HEADER = """# Shell Execution Guidelines
When using the `execute` tool for shell commands:
**Sandbox limits**: Commands default to a 300s timeout (a deployment may override this default) and 100 KB output. For a known long command (e.g. a download), pass `timeout` (up to 3600s): `execute(command="wget ...", timeout=600)`. For unbounded tasks, use background execution (below)."""
# Dangerous header: real filesystem, no virtual `/`. ``{cwd}`` = real working dir.
_SHELL_GUIDELINES_DANGEROUS_HEADER = """# Shell Execution Guidelines (DANGEROUS MODE)
You operate on the **host filesystem with real absolute paths** — there is no virtual workspace sandbox. Your current working directory is `{cwd}`. Use real absolute paths (e.g. `/Users/you/Documents/file.txt`) or paths relative to the cwd; `..` and `~` work normally. Run `pwd` any time you are unsure where you are.
⚠ You can read, write, move, copy, and delete files **anywhere on this machine**. There is no workspace confinement and no approval prompt. Be deliberate: double-check destination paths before writing or deleting, and never operate on a path you have not confirmed.
When using the `execute` tool for shell commands:
**Limits**: Commands default to a 300s timeout (a deployment may override this default) and 100 KB output. For a known long command (e.g. a download), pass `timeout` (up to 3600s): `execute(command="wget ...", timeout=600)`. For unbounded tasks, use background execution (below)."""
_SHELL_GUIDELINES_DANGEROUS_FOOTER = """
**Still blocked even here**: privileged/system commands (`sudo`, `chmod`, `chown`, `mkfs`, `dd`, `shutdown`, `reboot`) and `rm -rf /` are rejected regardless of mode."""
def _build_shell_guidelines(*, dangerous: bool = False, cwd: str | None = None) -> str:
"""Assemble the shell guidelines from the shared core + per-mode header/footer."""
if dangerous:
header = _SHELL_GUIDELINES_DANGEROUS_HEADER.format(cwd=cwd or ".")
body = _SHELL_GUIDELINES_CORE.format(log_path="./output.log")
return f"{header}\n\n{body}{_SHELL_GUIDELINES_DANGEROUS_FOOTER}\n"
body = _SHELL_GUIDELINES_CORE.format(log_path="/output.log")
return f"{_SHELL_GUIDELINES_SANDBOX_HEADER}\n\n{body}\n"
SHELL_GUIDELINES = _build_shell_guidelines()
# =============================================================================
# Sub-agent delegation strategy
@@ -392,6 +424,8 @@ def get_system_prompt(
*,
enable_observation_memory: bool = True,
enable_observation_writes: bool = True,
dangerous: bool = False,
cwd: str | None = None,
) -> str:
"""Generate the complete static system prompt.
@@ -401,7 +435,7 @@ def get_system_prompt(
2. :data:`EXPERIMENT_WORKFLOW`
3. :data:`REPORT_TEMPLATE`
4. :data:`WRITING_GUIDELINES`
5. :data:`SHELL_GUIDELINES`
5. :data:`SHELL_GUIDELINES` (or :data:`SHELL_GUIDELINES_DANGEROUS`)
6. :data:`DELEGATION_STRATEGY`
7. :data:`ASYNC_NOTIFICATIONS`
@@ -410,6 +444,12 @@ def get_system_prompt(
similar per-turn values are not baked into this prompt. Memory-related
workflow sections can vary with the configured memory controls.
Args:
dangerous: When True, use the real-filesystem shell guidance
(no virtual workspace) instead of the sandboxed default.
cwd: Real absolute working directory shown to the agent in
dangerous mode. Falls back to ``.`` when not provided.
Returns:
Combined static system prompt string.
"""
@@ -417,12 +457,17 @@ def get_system_prompt(
enable_observation_memory=enable_observation_memory,
enable_observation_writes=enable_observation_writes,
)
shell_guidelines = (
_build_shell_guidelines(dangerous=True, cwd=cwd)
if dangerous
else SHELL_GUIDELINES
)
sections = [
EVOSCIENTIST_IDENTITY,
workflow,
REPORT_TEMPLATE,
WRITING_GUIDELINES,
SHELL_GUIDELINES,
shell_guidelines,
DELEGATION_STRATEGY,
ASYNC_NOTIFICATIONS,
]
+89
View File
@@ -12,6 +12,7 @@ from EvoScientist.backends import (
CustomSandboxBackend,
MergedSkillsBackend,
convert_virtual_paths_in_command,
prepare_sandbox_command,
validate_command,
)
@@ -78,6 +79,34 @@ class TestValidateCommand:
assert "/home/username/project" in result
class TestValidateCommandDangerous:
"""dangerous=True drops path confinement but keeps the command blocklist."""
def test_absolute_path_allowed(self):
assert validate_command("cat /etc/passwd", dangerous=True) is None
def test_traversal_allowed(self):
assert validate_command("cat ../../x", dangerous=True) is None
def test_home_tilde_allowed(self):
assert validate_command("cat ~/secrets.txt", dangerous=True) is None
def test_cd_absolute_allowed(self):
assert validate_command("cd /etc && ls", dangerous=True) is None
def test_sudo_still_blocked(self):
assert validate_command("sudo rm x", dangerous=True) is not None
def test_chmod_still_blocked(self):
assert validate_command("chmod 777 /tmp/x", dangerous=True) is not None
def test_dd_still_blocked(self):
assert validate_command("dd if=/dev/zero of=/x", dangerous=True) is not None
def test_rm_rf_root_still_blocked(self):
assert validate_command("rm -rf /", dangerous=True) is not None
# === convert_virtual_paths_in_command ===
@@ -729,6 +758,55 @@ class TestResolvePath:
assert backend._resolve_path(tmp_workspace) == backend._resolve_path("/")
class TestResolvePathDangerous:
"""Dangerous mode passes real absolute paths through unmangled."""
def test_absolute_path_unmangled(self, tmp_workspace):
backend = CustomSandboxBackend(root_dir=tmp_workspace, dangerous=True)
# OS-appropriate absolute path (drive-anchored on Windows) outside the ws.
target = Path(Path(tmp_workspace).anchor, "etc", "hosts")
resolved = Path(backend._resolve_path(str(target)))
# Dangerous mode must NOT confine/mangle it into the workspace.
assert Path(tmp_workspace) not in resolved.parents
assert resolved == target
def test_outside_workspace_not_confined(self, tmp_path):
ws = tmp_path / "ws"
ws.mkdir()
outside = tmp_path / "elsewhere" / "data.csv"
backend = CustomSandboxBackend(root_dir=str(ws), dangerous=True)
resolved = Path(backend._resolve_path(str(outside)))
assert resolved == outside # real path, not pulled into the workspace
def test_dangerous_forces_virtual_mode_off(self, tmp_workspace):
backend = CustomSandboxBackend(
root_dir=tmp_workspace, virtual_mode=True, dangerous=True
)
assert backend.virtual_mode is False
def test_dangerous_skips_cwd_literal_rewrite(self, tmp_workspace):
"""In dangerous mode the cwd->'./' rewrite must NOT mangle real args.
Regression: a non-path argument that merely contains the cwd string
(echo text, grep/git pattern) was being corrupted to './'.
"""
cmd = f'echo "backup of {tmp_workspace}/data"'
prepared, error = prepare_sandbox_command(
cmd, tmp_workspace, virtual_mode=False, dangerous=True
)
assert error is None
assert prepared == cmd # unchanged — no './' substitution
def test_non_dangerous_still_rewrites_cwd_literal(self, tmp_workspace):
"""Default mode keeps the workspace-literal -> './' rewrite."""
cmd = f"cat {tmp_workspace}/file.txt"
prepared, error = prepare_sandbox_command(
cmd, tmp_workspace, virtual_mode=True, dangerous=False
)
assert error is None
assert prepared == "cat ./file.txt"
# === CustomSandboxBackend.id ===
@@ -1352,6 +1430,17 @@ class TestExecuteTimeoutRecovery:
assert cmd in resp.output
assert "> /output.log 2>&1 &" in resp.output
def test_timeout_recovery_uses_relative_log_in_dangerous(self, tmp_workspace):
"""Dangerous mode: recovery hint must not point the log at the host root."""
backend = CustomSandboxBackend(
root_dir=tmp_workspace, timeout=1, dangerous=True
)
resp = backend.execute(_sleep_cmd(10))
assert resp.exit_code == 124
assert "> ./output.log 2>&1 &" in resp.output
assert "cat ./output.log" in resp.output
assert "> /output.log" not in resp.output
def test_timeout_recovery_captures_pid_and_offers_timeout(self, tmp_workspace):
backend = CustomSandboxBackend(root_dir=tmp_workspace, timeout=1)
resp = backend.execute(_sleep_cmd(10))
+45
View File
@@ -109,6 +109,51 @@ def test_run_applies_virtual_path_rewriting(tmp_path, monkeypatch):
assert captured["command"] == "python ./train.py"
def _force_dangerous(monkeypatch, value=True):
"""Make run_in_background see dangerous mode via the env flag it reads.
monkeypatch.setenv tracks the change and restores it on teardown, so this
cannot leak EVOSCIENTIST_DANGEROUS_MODE into other tests.
"""
monkeypatch.setenv("EVOSCIENTIST_DANGEROUS_MODE", "true" if value else "false")
def test_run_dangerous_allows_real_path_no_rewrite(tmp_path, monkeypatch):
"""In dangerous mode, background commands keep real absolute paths (parity with execute)."""
monkeypatch.setattr("EvoScientist.paths.resolve_virtual_path", lambda _vp: tmp_path)
_force_dangerous(monkeypatch)
captured = {}
def _spy(command, cwd, name=None, *, origin_thread_id=None, on_exit=None):
captured["command"] = command
return "pidX"
monkeypatch.setattr(bg, "launch", _spy)
# Absolute path + traversal would be BLOCKED in normal mode; allowed here.
out = run_in_background.invoke({"command": "cat /etc/hosts && cat ../x"})
assert "blocked" not in out.lower()
assert captured["command"] == "cat /etc/hosts && cat ../x" # no ./ rewrite
# Advertised log path is the real path, not the virtual /.bg_processes/.
assert f"{tmp_path}/.bg_processes/" in out
assert "Output -> /.bg_processes/" not in out
def test_run_dangerous_still_blocks_privileged_command(tmp_path, monkeypatch):
"""Dangerous mode must NOT relax the privileged-command blocklist."""
monkeypatch.setattr("EvoScientist.paths.resolve_virtual_path", lambda _vp: tmp_path)
_force_dangerous(monkeypatch)
launched = {"called": False}
def _spy(*args, **kwargs):
launched["called"] = True
return "should-not-happen"
monkeypatch.setattr(bg, "launch", _spy)
out = run_in_background.invoke({"command": "sudo rm x"})
assert launched["called"] is False
assert "blocked" in out.lower()
def test_run_enqueues_completion_notification(tmp_path, monkeypatch):
"""A finished background process enqueues a shell completion notification."""
from EvoScientist.cli import async_notifier
+2
View File
@@ -29,6 +29,7 @@ def _make_config(
log_level: str = "warning",
langgraph_dev_jobs_per_worker: int = 10,
langgraph_dev_file_persistence: bool = True,
dangerous_mode: bool = False,
):
return SimpleNamespace(
default_workdir=default_workdir,
@@ -38,6 +39,7 @@ def _make_config(
log_level=log_level,
langgraph_dev_jobs_per_worker=langgraph_dev_jobs_per_worker,
langgraph_dev_file_persistence=langgraph_dev_file_persistence,
dangerous_mode=dangerous_mode,
)
+18
View File
@@ -18,6 +18,7 @@ def _make_config(
auto_approve: bool = False,
auto_mode: bool = False,
enable_ask_user: bool = True,
dangerous_mode: bool = False,
):
return SimpleNamespace(
channel_enabled="telegram",
@@ -28,6 +29,7 @@ def _make_config(
auto_approve=auto_approve,
auto_mode=auto_mode,
enable_ask_user=enable_ask_user,
dangerous_mode=dangerous_mode,
enable_async_subagents=False,
memory_profile_enabled=True,
memory_observations_enabled=True,
@@ -50,6 +52,7 @@ def _run_serve_once(
auto_approve: bool = False,
auto_mode: bool = False,
ask_user: bool = False,
dangerous: bool = False,
):
import EvoScientist.config as config_mod
@@ -108,6 +111,7 @@ def _run_serve_once(
auto_approve=auto_approve,
auto_mode=auto_mode,
ask_user=ask_user,
dangerous=dangerous,
)
return order, captured
@@ -246,3 +250,17 @@ def test_serve_auto_mode_implies_auto_approve_and_disables_ask_user(
"auto_approve": True,
"enable_ask_user": False,
}
def test_serve_dangerous_sets_dangerous_mode(monkeypatch, tmp_path):
ws = str((tmp_path / "ws").resolve())
config = _make_config(default_workdir=ws)
_, captured = _run_serve_once(
monkeypatch,
config,
workdir=ws,
dangerous=True,
)
assert captured["cli_overrides"] == {"dangerous_mode": True}
+70
View File
@@ -28,6 +28,24 @@ from EvoScientist.config import (
# =============================================================================
@pytest.fixture(autouse=True)
def _restore_dangerous_env():
"""Snapshot/restore EVOSCIENTIST_DANGEROUS_MODE around every test.
apply_config_to_env writes this var via direct ``os.environ`` assignment, and
monkeypatch's ``delenv`` of an originally-absent key records no undo — so
without this, a test that turns dangerous mode on would leak the env var into
later tests (an order-dependent landmine, e.g. under pytest-randomly).
"""
_sentinel = object()
_prev = os.environ.get("EVOSCIENTIST_DANGEROUS_MODE", _sentinel)
yield
if _prev is _sentinel:
os.environ.pop("EVOSCIENTIST_DANGEROUS_MODE", None)
else:
os.environ["EVOSCIENTIST_DANGEROUS_MODE"] = _prev
@pytest.fixture
def temp_config_dir(tmp_path, monkeypatch):
"""Use a temporary directory for config during tests."""
@@ -53,6 +71,7 @@ def temp_config_dir(tmp_path, monkeypatch):
"EVOSCIENTIST_AUXILIARY_MODEL",
"EVOSCIENTIST_AUXILIARY_PROVIDER",
"EVOSCIENTIST_OPENROUTER_ANTHROPIC_PROMPT_CACHE",
"EVOSCIENTIST_DANGEROUS_MODE",
]:
monkeypatch.delenv(key, raising=False)
return config_dir
@@ -75,6 +94,7 @@ def clean_env(monkeypatch):
"EVOSCIENTIST_AUXILIARY_MODEL",
"EVOSCIENTIST_AUXILIARY_PROVIDER",
"EVOSCIENTIST_OPENROUTER_ANTHROPIC_PROMPT_CACHE",
"EVOSCIENTIST_DANGEROUS_MODE",
]:
monkeypatch.delenv(key, raising=False)
@@ -144,6 +164,18 @@ class TestEvoScientistConfig:
assert config.model == "gpt-4o"
assert config.default_mode == "run"
def test_dangerous_mode_default(self):
"""dangerous_mode defaults off and does not force auto_approve."""
config = EvoScientistConfig()
assert config.dangerous_mode is False
assert config.auto_approve is False
def test_dangerous_mode_implies_auto_approve(self):
"""Enabling dangerous_mode forces auto_approve via __post_init__."""
config = EvoScientistConfig(dangerous_mode=True)
assert config.dangerous_mode is True
assert config.auto_approve is True
# =============================================================================
# Test config path functions
@@ -612,6 +644,44 @@ class TestApplyConfigToEnv:
"true"
)
def test_dangerous_mode_round_trips_to_env(self, clean_env, monkeypatch):
"""dangerous_mode set via CLI override must survive a fresh re-read.
Regression: a --dangerous CLI flag is never persisted to file/env, so a
fresh get_effective_config() (warning banner, run_in_background, the
langgraph dev subprocess) would read it back as False while the backend
is already unconfined. apply_config_to_env round-trips it to env.
"""
from EvoScientist.config import get_effective_config
monkeypatch.delenv("EVOSCIENTIST_DANGEROUS_MODE", raising=False)
cfg = get_effective_config({"dangerous_mode": True})
assert cfg.dangerous_mode is True
apply_config_to_env(cfg)
assert os.environ.get("EVOSCIENTIST_DANGEROUS_MODE") == "true"
# The fresh, no-override read now agrees with the backend.
assert get_effective_config().dangerous_mode is True
def test_dangerous_mode_not_applied_when_off(self, clean_env, monkeypatch):
"""dangerous_mode=False must not write the env var."""
monkeypatch.delenv("EVOSCIENTIST_DANGEROUS_MODE", raising=False)
apply_config_to_env(EvoScientistConfig())
assert os.environ.get("EVOSCIENTIST_DANGEROUS_MODE") is None
def test_dangerous_mode_off_clears_stale_env(self, clean_env, monkeypatch):
"""Re-applying a non-dangerous config clears a previously-set env var.
Regression: the round-trip used to be set-only, so once dangerous mode
was applied in a process it could never be lowered — leaking unconfined
access into later non-dangerous reads.
"""
monkeypatch.delenv("EVOSCIENTIST_DANGEROUS_MODE", raising=False)
apply_config_to_env(EvoScientistConfig(dangerous_mode=True))
assert os.environ.get("EVOSCIENTIST_DANGEROUS_MODE") == "true"
apply_config_to_env(EvoScientistConfig()) # dangerous off
assert os.environ.get("EVOSCIENTIST_DANGEROUS_MODE") is None
def test_ollama_base_url_applied(self, clean_env, monkeypatch):
"""Test that ollama_base_url is applied to OLLAMA_BASE_URL env var."""
monkeypatch.delenv("OLLAMA_BASE_URL", raising=False)
+21
View File
@@ -182,3 +182,24 @@ class TestShellGuidelines:
"""SHELL_GUIDELINES content should live ONLY in its own constant."""
# Sentinel phrase unique to SHELL_GUIDELINES
assert "Sandbox limits" not in EXPERIMENT_WORKFLOW
class TestDangerousShellGuidelines:
def test_default_uses_virtual_paths(self):
result = get_system_prompt()
assert "> /output.log" in result
assert "DANGEROUS MODE" not in result
def test_dangerous_swaps_guidelines(self):
result = get_system_prompt(dangerous=True, cwd="/Users/me/ws/demo")
assert "DANGEROUS MODE" in result
assert "/Users/me/ws/demo" in result
# virtual-path example is gone
assert "> /output.log" not in result
# privileged-command blocklist still advertised
assert "sudo" in result
assert "rm -rf /" in result
def test_dangerous_without_cwd_falls_back(self):
result = get_system_prompt(dangerous=True)
assert "DANGEROUS MODE" in result