From ccec204bc74e50248c7bd1caf84347309dadd225 Mon Sep 17 00:00:00 2001 From: ciregenz Date: Wed, 13 May 2026 11:49:33 -0700 Subject: [PATCH] [eric] security: gate Write/Edit on sensitive paths + restore HOME/PATH for force-mode Outputs --- backend/apps/agents/agent_manager.py | 64 ++++++++++++++++++++++++++-- backend/apps/outputs/executor.py | 48 ++++++++++++++++++--- 2 files changed, 102 insertions(+), 10 deletions(-) diff --git a/backend/apps/agents/agent_manager.py b/backend/apps/agents/agent_manager.py index f9d302ce..39a9bdb7 100644 --- a/backend/apps/agents/agent_manager.py +++ b/backend/apps/agents/agent_manager.py @@ -1124,6 +1124,60 @@ class AgentManager: def _default_for(tool_name: str) -> str: return _DEFAULTS.get(tool_name, "always_allow") + # Path patterns that flip Write/Edit/NotebookEdit from always_allow + # to ask regardless of the user's base permission. Targets the + # narrow set of files a prompt-injected agent would use to exfil + # or persist (SSH keys, shell rc files, env files, cloud creds, + # system dirs). Normal in-project / in-workspace / in-Downloads + # edits never match — keeping the prompt-fatigue surface tiny. + import fnmatch as _fnmatch + + _SENSITIVE_PATH_PATTERNS = ( + "*/.ssh", "*/.ssh/*", + "*/.aws/*", "*/.config/gcloud/*", "*/.kube/*", + "*/.gnupg/*", "*/.docker/config*", + "*/.zshrc", "*/.bashrc", "*/.bash_profile", + "*/.profile", "*/.zprofile", "*/.zshenv", + "*/.gitconfig", "*/.npmrc", "*/.pypirc", "*/.netrc", + "*/.env", "*/.env.*", "*.env", + "*/Library/Application Support/*", # macOS credential stores + "*/Library/Keychains/*", + "/etc/*", "/private/etc/*", "/System/*", + "/usr/local/etc/*", + ) + + def _is_sensitive_write_path(file_path: str) -> bool: + if not file_path or not isinstance(file_path, str): + return False + try: + norm = os.path.normpath(os.path.expanduser(file_path)) + except Exception: + return False + for pat in _SENSITIVE_PATH_PATTERNS: + if _fnmatch.fnmatch(norm, pat): + return True + return False + + _PATH_GATED_TOOLS = ("Write", "Edit", "NotebookEdit") + + def _extract_target_path(tool_name: str, tool_input) -> str: + if not isinstance(tool_input, dict): + return "" + if tool_name == "NotebookEdit": + return str(tool_input.get("notebook_path") or "") + return str(tool_input.get("file_path") or "") + + def _maybe_override_policy(policy: str, tool_name: str, tool_input) -> str: + """Flip a permissive policy to 'ask' when the target path is + sensitive. Defense in depth: even if the user has Write set to + always_allow for productivity, a prompt-injected agent writing + to ~/.ssh/authorized_keys or ~/.zshrc gets surfaced for review.""" + if policy != "always_allow" or tool_name not in _PATH_GATED_TOOLS: + return policy + if _is_sensitive_write_path(_extract_target_path(tool_name, tool_input)): + return "ask" + return policy + def _get_effective_policy(tool_name: str) -> str: """Return 'always_allow', 'deny', or 'ask' for any tool.""" if tool_name in _builtin_perms: @@ -1196,7 +1250,9 @@ class AgentManager: async def can_use_tool(tool_name, input_data, context): if tool_name != "AskUserQuestion": - policy = _get_effective_policy(tool_name) + policy = _maybe_override_policy( + _get_effective_policy(tool_name), tool_name, input_data + ) if policy == "always_allow": return PermissionResultAllow(updated_input=input_data) if policy == "deny": @@ -1218,7 +1274,10 @@ class AgentManager: hook_event = input_data.get("hook_event_name", "PreToolUse") if tool_name and tool_name != "AskUserQuestion": - policy = _get_effective_policy(tool_name) + tool_input = input_data.get("tool_input", {}) + policy = _maybe_override_policy( + _get_effective_policy(tool_name), tool_name, tool_input + ) if policy == "deny": return { @@ -1230,7 +1289,6 @@ class AgentManager: } if policy == "ask": - tool_input = input_data.get("tool_input", {}) decision = await _request_user_approval(tool_name, tool_input) if decision.get("behavior") == "allow": diff --git a/backend/apps/outputs/executor.py b/backend/apps/outputs/executor.py index eca24320..01d62341 100644 --- a/backend/apps/outputs/executor.py +++ b/backend/apps/outputs/executor.py @@ -96,14 +96,48 @@ def _validate_code_safety(code: str) -> None: raise UnsafeCodeError(warnings[0]) -def _minimal_env() -> dict: - """Build a stripped-down env for the executor subprocess. +# Env vars we always scrub from the subprocess, regardless of strict-vs-force. +# These are the keys an attacker would actually want — install token, provider +# API keys, cloud credentials. Everything else is local-machine convenience. +_SCRUBBED_ENV_KEYS = frozenset({ + "OPENSWARM_AUTH_TOKEN", + "ANTHROPIC_API_KEY", + "OPENAI_API_KEY", + "GOOGLE_API_KEY", + "GEMINI_API_KEY", + "OPENROUTER_API_KEY", + "AWS_ACCESS_KEY_ID", + "AWS_SECRET_ACCESS_KEY", + "AWS_SESSION_TOKEN", + "GOOGLE_APPLICATION_CREDENTIALS", + "STRIPE_API_KEY", + "STRIPE_SECRET_KEY", + "GITHUB_TOKEN", +}) - Drops PATH, OPENSWARM_AUTH_TOKEN, OPENAI_API_KEY, ANTHROPIC_API_KEY, and - every other inherited credential. Keeps only what Python itself needs to - boot on each platform — on Windows that's SYSTEMROOT et al, on POSIX - nothing is strictly required. + +def _minimal_env(force: bool = False) -> dict: + """Build the env for the executor subprocess. + + Strict mode (force=False): only language essentials. AST-validated code + is data-shaping only — `import os` and `open()` are blocked, so the + subprocess can't read env vars or expand `~` anyway. Minimal env is + correct here. + + Force mode (force=True): user has explicitly approved unsafe imports + via the HITL preview. They expect the code to behave like a normal + Python process — read HOME, find files, etc. Inherit the real env + minus credentials, so an `open(os.path.expanduser("~/data.csv"))` + actually works instead of silently misbehaving. + + Both modes scrub _SCRUBBED_ENV_KEYS so even force-mode code never + sees the install token or provider API keys. """ + if force: + env = {k: v for k, v in os.environ.items() if k not in _SCRUBBED_ENV_KEYS} + env["PYTHONDONTWRITEBYTECODE"] = "1" + return env + env = { "PYTHONDONTWRITEBYTECODE": "1", "LANG": os.environ.get("LANG", "C.UTF-8"), @@ -176,7 +210,7 @@ async def execute_backend_code( stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=workdir, - env=_minimal_env(), + env=_minimal_env(force=skip_validation), ) try: