Add sensitive file access validation to bash tool

agent/tools/bash.py

@@ -1,6 +1,7 @@
 """Bash/shell command execution tool."""
 
 import os
+import re
 import subprocess
 import shlex
 import platform
@@ -24,6 +25,136 @@
     "git clean -f", "git checkout .",
 ]
 
+# Commands whose argv[0] reads file contents (checked against SENSITIVE_FILE_PATTERNS)
+FILE_READ_COMMANDS = {
+    "cat", "less", "more", "head", "tail", "vim", "vi", "nano", "strings",
+    "grep", "sort", "od", "xxd", "base64",
+}
+
+# Path fragments that identify secret-bearing files, matched against the
+# normalized/expanded path argument rather than the raw command text
+SENSITIVE_FILE_PATTERNS = [
+    re.compile(r"/etc/passwd$"),
+    re.compile(r"/etc/shadow$"),
+    re.compile(r"\.ssh(/|$)"),
+    re.compile(r"\.aws(/|$)"),
+    re.compile(r"\.netrc$"),
+    re.compile(r"id_rsa$"),
+    re.compile(r"id_ed25519$"),
+    re.compile(r"/proc/(?:self|\d+)/environ$"),
+]
+
+# Commands that print environment variables; sensitive only when dumping
+# everything (no specific name given) or when the requested name looks secret
+ENV_DUMP_COMMANDS = {"env", "printenv", "set", "declare", "export"}
+
+# argv[0] for commands that can print a variable's value directly
+ENV_PRINT_COMMANDS = {"echo", "printf"}
+
+SECRET_NAME_RE = re.compile(r"(API[_-]?KEY|SECRET|TOKEN|PASSWORD|CREDENTIAL)", re.IGNORECASE)
+VAR_REF_RE = re.compile(r"\$\{?([A-Za-z_][A-Za-z0-9_]*)\}?")
+
+# Shells whose -x flag traces commands with variables already expanded
+TRACING_SHELLS = {"bash", "sh", "zsh", "dash", "ksh"}
+
+
+def _enables_xtrace(argv0: str, args: list) -> bool:
+    """True if this command turns on xtrace (`set -x`, `set -o xtrace`,
+    `bash -x ...`), which prints every later command with its variables
+    expanded, so a secret referenced anywhere in the line reaches the
+    output."""
+    if argv0 != "set" and argv0 not in TRACING_SHELLS:
+        return False
+    for i, arg in enumerate(args):
+        if arg.startswith("-") and not arg.startswith("--") and "x" in arg[1:]:
+            return True
+        if arg == "-o" and i + 1 < len(args) and args[i + 1] == "xtrace":
+            return True
+    return False
+
+
+# env's own options, GNU and BSD/macOS. Short options in ENV_SHORT_WITH_ARG
+# take an argument (the rest of the cluster, or the next token).
+ENV_SHORT_FLAGS = set("iv0")
+ENV_SHORT_WITH_ARG = set("uCSaPLU")
+ENV_LONG_FLAGS = {"--ignore-environment", "--null", "--debug", "--list-signal-handling"}
+ENV_LONG_WITH_ARG = {"--unset", "--chdir", "--split-string", "--argv0"}
+# Signal options take an optional argument, only in the attached form.
+ENV_LONG_OPTIONAL_ARG = {"--block-signal", "--default-signal", "--ignore-signal"}
+
+
+def _unwrap_env(tokens: list) -> tuple:
+    """Parse `env [opts] [NAME=value ...] [cmd ...]` the way getopt does.
+
+    Returns ("cmd", tokens) for the wrapped command, ("dump", None) when no
+    command is left (env then prints every variable), or ("unknown", None)
+    for an option this parser does not recognise. Callers block both of
+    the latter: failing closed is the point of the check.
+    """
+    rest = list(tokens[1:])
+    while rest:
+        tok = rest.pop(0)
+        if tok == "--":
+            break
+        if tok == "-":
+            continue  # same as -i
+        if tok.startswith("--"):
+            name, eq, _ = tok.partition("=")
+            if name == "--split-string":
+                arg = tok[len(name) + 1:] if eq else (rest.pop(0) if rest else "")
+                try:
+                    rest = shlex.split(arg) + rest
+                except ValueError:
+                    return ("unknown", None)
+            elif name in ENV_LONG_WITH_ARG:
+                if not eq:
+                    if not rest:
+                        return ("unknown", None)
+                    rest.pop(0)
+            elif name in ENV_LONG_FLAGS or name in ENV_LONG_OPTIONAL_ARG:
+                pass
+            else:
+                return ("unknown", None)
+            continue
+        if tok.startswith("-"):
+            for pos, ch in enumerate(tok[1:], start=1):
+                if ch in ENV_SHORT_FLAGS:
+                    continue
+                if ch in ENV_SHORT_WITH_ARG:
+                    attached = tok[pos + 1:]
+                    if attached:
+                        arg = attached
+                    elif rest:
+                        arg = rest.pop(0)
+                    else:
+                        return ("unknown", None)
+                    if ch == "S":
+                        # -S splits its string into further env arguments.
+                        try:
+                            rest = shlex.split(arg) + rest
+                        except ValueError:
+                            return ("unknown", None)
+                    break
+                return ("unknown", None)
+            continue
+        if "=" in tok:
+            continue
+        return ("cmd", [tok] + rest)
+    # Arguments after "--" are the command, except leading assignments.
+    while rest and "=" in rest[0]:
+        rest.pop(0)
+    return ("cmd", rest) if rest else ("dump", None)
+
+
+# Commands that run another command given later in their argv. Their own
+# options vary too much to parse, so the guard checks from the first token
+# that is itself a checked command.
+WRAPPER_COMMANDS = {
+    "command", "exec", "builtin", "nice", "nohup", "time", "timeout", "stdbuf",
+    "sudo", "doas", "xargs", "ionice", "setsid", "chrt", "taskset", "unbuffer",
+}
+
+
 # Minimum timeout to prevent accidental no-timeout execution
 MIN_TIMEOUT = 1
 DEFAULT_TIMEOUT = 120
@@ -71,6 +202,109 @@ def _is_dangerous(self, command: str) -> bool:
         cmd_lower = command.lower()
         return any(p in cmd_lower for p in DANGEROUS_PATTERNS)
 
+    def _is_sensitive(self, command: str) -> Optional[str]:
+        """Check if command reads a sensitive file or prints a secret-looking
+        environment variable. Operates on parsed argv per shell segment so
+        that commands merely mentioning a secret's name (e.g. grepping for
+        it in source) are not flagged, only commands structured to read or
+        print one.
+
+        This is a best-effort guard against accidental or injected leakage,
+        not a sandbox: only top-level segments are inspected, so command
+        substitution, backticks, and interpreter wrappers such as
+        ``bash -c`` or ``python -c`` are not examined.
+        """
+        secret_refs = [
+            name for name in VAR_REF_RE.findall(command) if SECRET_NAME_RE.search(name)
+        ]
+        for segment in re.split(r"\|\||&&|[;|\n]", command):
+            segment = segment.strip()
+            if not segment:
+                continue
+            try:
+                tokens = shlex.split(segment)
+            except ValueError:
+                # Unbalanced quotes (often from the split above cutting
+                # inside a quoted "|"). Fall back to whitespace tokens so
+                # the segment is still checked rather than skipped.
+                tokens = segment.split()
+            if not tokens:
+                continue
+
+            # `env [opts] [NAME=value ...] cmd` runs cmd, so check cmd
+            # itself; with no cmd it prints every variable.
+            checked = (
+                FILE_READ_COMMANDS | ENV_DUMP_COMMANDS | ENV_PRINT_COMMANDS
+                | TRACING_SHELLS | WRAPPER_COMMANDS | {"env"}
+            )
+            while True:
+                head = os.path.basename(tokens[0])
+                if head in WRAPPER_COMMANDS:
+                    inner = next(
+                        (i for i, t in enumerate(tokens[1:], start=1)
+                         if os.path.basename(t) in checked),
+                        None,
+                    )
+                    if inner is None:
+                        break
+                    tokens = tokens[inner:]
+                    continue
+                if head != "env":
+                    break
+                kind, wrapped = _unwrap_env(tokens)
+                if kind == "dump":
+                    return f"Blocked: '{segment}' dumps environment variables"
+                if kind == "unknown":
+                    return (
+                        f"Blocked: '{segment}' uses env options this guard "
+                        "cannot parse, so it may dump environment variables"
+                    )
+                tokens = wrapped
+
+            argv0 = os.path.basename(tokens[0])
+            args = tokens[1:]
+
+            if secret_refs and _enables_xtrace(argv0, args):
+                return (
+                    f"Blocked: '{segment}' turns on command tracing, which "
+                    f"would print the value of '{secret_refs[0]}'"
+                )
+
+            if argv0 in FILE_READ_COMMANDS:
+                for arg in args:
+                    if arg.startswith("-"):
+                        continue
+                    expanded = os.path.normpath(
+                        os.path.expanduser(os.path.expandvars(arg))
+                    )
+                    for pat in SENSITIVE_FILE_PATTERNS:
+                        if pat.search(expanded):
+                            return f"Blocked: '{segment}' reads sensitive file '{arg}'"
+
+            if argv0 in ENV_DUMP_COMMANDS:
+                named = [a for a in args if not a.startswith("-")]
+                # NAME=value sets a variable (`export TOKEN=...`,
+                # `env TOKEN=x ./run.sh`); only bare names read one.
+                reads = [a for a in named if "=" not in a]
+                # Only bare `set` prints variables; `set -e`, `set -x` and
+                # `set -o ...` toggle shell options and print nothing.
+                if argv0 == "set":
+                    dumps = not args
+                else:
+                    dumps = not named
+                if dumps or any(SECRET_NAME_RE.search(a) for a in reads):
+                    return f"Blocked: '{segment}' dumps environment variables"
+
+            if argv0 in ENV_PRINT_COMMANDS:
+                for arg in args:
+                    for var_name in VAR_REF_RE.findall(arg):
+                        if SECRET_NAME_RE.search(var_name):
+                            return (
+                                f"Blocked: '{segment}' prints the value of "
+                                f"'{var_name}'"
+                            )
+        return None
+
     def _sanitize_timeout(self, timeout: Optional[int]) -> int:
         """Validate and enforce timeout bounds."""
         if timeout is None:
@@ -85,6 +319,16 @@ def execute(self, command: str, timeout: Optional[int] = None, **kw) -> ToolResu
         if blocked:
             return ToolResult(False, "", blocked)
 
+        sensitive = self._is_sensitive(command)
+        if sensitive and not self.auto_approve:
+            return ToolResult(
+                False, "",
+                f"{sensitive}\n"
+                "Interactive confirmation does not lift this check; only "
+                "auto_approve_bash=true in config skips it, and that disables "
+                "confirmation for every bash command."
+            )
+
         if self._is_dangerous(command) and not self.auto_approve:
             return ToolResult(
                 False, "",

tests/test_tools.py

@@ -239,6 +239,105 @@ def test_timeout(self):
         self.assertFalse(result.success)
         self.assertIn("timed out", result.error)
 
+    def test_sensitive_file_reads_blocked(self):
+        guarded_tool = BashTool(self.tmpdir, auto_approve=False)
+        for cmd in (
+            "cat /etc/passwd",
+            "cat /etc/shadow",
+            "head /etc/shadow",
+            "less /etc/shadow",
+            "cat /etc//shadow",
+            "cat ~/.ssh/id_rsa",
+            "cat $HOME/.ssh/id_rsa",
+            "cd ~/.ssh && cat id_rsa",
+            "cat /proc/self/environ",
+            'cat /etc/shadow "|"',
+            "grep . /etc/shadow",
+            "base64 ~/.ssh/id_rsa",
+            "env FOO=1 cat ~/.ssh/id_rsa",
+            "nohup cat ~/.ssh/id_rsa",
+            "sudo -u root cat /etc/shadow",
+        ):
+            result = guarded_tool.execute(command=cmd)
+            self.assertFalse(result.success, cmd)
+            self.assertIn("sensitive", result.error.lower(), cmd)
+
+    def test_env_dumps_blocked(self):
+        guarded_tool = BashTool(self.tmpdir, auto_approve=False)
+        for cmd in (
+            "env",
+            "env|grep key",
+            "printenv",
+            "set",
+            "env FOO=1",
+            "env -i FOO=1",
+            "env FOO=1 printenv",
+            "env -u HOME printenv ANTHROPIC_API_KEY",
+            "env -P /bin printenv",
+            "env -a sh printenv",
+            "env -iu FOO printenv",
+            "env -uFOO printenv",
+            "env --unset FOO printenv",
+            "env --unset=FOO printenv",
+            "env -S 'printenv ANTHROPIC_API_KEY'",
+            "env -S'FOO=1 printenv'",
+            "env --split-string='printenv'",
+            "env -- FOO=1 printenv",
+            "env --made-up-option printenv",
+            "env -Z printenv",
+            "command printenv",
+            "exec env",
+            "timeout 5 printenv ANTHROPIC_API_KEY",
+            "nice -n 5 env FOO=1 printenv",
+            "declare -p",
+            "printenv ANTHROPIC_API_KEY",
+            "echo $ANTHROPIC_API_KEY",
+        ):
+            result = guarded_tool.execute(command=cmd)
+            self.assertFalse(result.success, cmd)
+
+    def test_xtrace_with_secret_reference_blocked(self):
+        # Tracing prints each command with variables expanded, so allowing
+        # `set -x` must not open a route around the echo/printenv checks.
+        guarded_tool = BashTool(self.tmpdir, auto_approve=False)
+        for cmd in (
+            'set -x; curl -s -H "Authorization: $ANTHROPIC_API_KEY" localhost',
+            'set -eux; curl -s -H "x: ${OPENAI_API_KEY}" localhost',
+            'set -o xtrace; curl -s -u "u:$GITHUB_TOKEN" localhost',
+            "bash -x -c 'curl -s -H \"k: $API_TOKEN\" localhost'",
+        ):
+            result = guarded_tool.execute(command=cmd)
+            self.assertFalse(result.success, cmd)
+            self.assertIn("command tracing", result.error, cmd)
+
+    def test_sensitive_checks_overridable_with_auto_approve(self):
+        auto_tool = BashTool(self.tmpdir, auto_approve=True)
+        result = auto_tool.execute(command="echo $ANTHROPIC_API_KEY")
+        self.assertTrue(result.success)
+
+    def test_legitimate_commands_not_blocked(self):
+        guarded_tool = BashTool(self.tmpdir, auto_approve=False)
+        for cmd in (
+            "grep -rn ANTHROPIC_API_KEY scripts/",
+            "printenv PATH",
+            "cat ~/.config/nvim/init.lua",
+            "echo hello",
+            "ls -la",
+            "set -e; echo ok",
+            "set -x",
+            "set -eu",
+            "set -o pipefail",
+            "export ANTHROPIC_API_KEY=sk-test",
+            "env GITHUB_TOKEN=xxx true",
+            "env -i PATH=/usr/bin true",
+            "env -u FOO true",
+            "timeout 5 echo ok",
+            "time ls -la",
+        ):
+            result = guarded_tool.execute(command=cmd)
+            self.assertNotIn("sensitive", (result.error or "").lower(), cmd)
+            self.assertNotIn("dumps environment", (result.error or ""), cmd)
+
 
 class TestConversation(unittest.TestCase):
     def test_basic_conversation(self):