diff --git a/README.md b/README.md index 10a6ae0..19ccdff 100644 --- a/README.md +++ b/README.md @@ -37,7 +37,7 @@ DevMind: [searches the project] [runs tests] [reads code] - **Reliable sessions:** named saves, autosave, atomic persistence, and compacted conversation history. - **Visible operation:** streaming output, structured logs, task state, retry metrics, and per-tool latency. - **Cost awareness:** token usage and estimated USD cost are recorded for each API call. -- **Offline confidence:** 156 tests cover tools, security guards, persistence, metrics, plugins, and control flow without paid API calls. +- **Offline confidence:** 170 tests cover tools, security guards, persistence, metrics, plugins, and control flow without paid API calls. ## Architecture diff --git a/security.py b/security.py index 4c2745c..d3b8b5b 100644 --- a/security.py +++ b/security.py @@ -96,18 +96,166 @@ def validate_path( # Command validation # ============================================================ _SUSPICIOUS_CHAR_RUNS = re.compile(r"`.+`|\$\([^)]+\)") +_SHELL_OPERATORS = re.compile(r"&&|\|\||[;|\n]") +_ENV_ASSIGNMENT = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*=") +_WRAPPERS = { + "builtin", + "command", + "env", + "exec", + "nice", + "nohup", + "setsid", + "stdbuf", + "time", + "timeout", + "xargs", +} +_WRAPPER_VALUE_OPTIONS = { + "env": {"-u", "--unset", "-C", "--chdir"}, + "exec": {"-a"}, + "nice": {"-n", "--adjustment"}, + "stdbuf": {"-i", "-o", "-e", "--input", "--output", "--error"}, + "time": {"-f", "--format", "-o", "--output"}, + "timeout": {"-s", "--signal", "-k", "--kill-after"}, + "xargs": { + "-a", + "--arg-file", + "-d", + "--delimiter", + "-E", + "--eof", + "-I", + "--replace", + "-L", + "--max-lines", + "-n", + "--max-args", + "-P", + "--max-procs", + "-s", + "--max-chars", + }, +} +_PRIVILEGE_TOOLS = {"sudo", "su", "doas", "pkexec"} +_SHELLS = {"bash", "sh", "zsh", "ksh", "fish", "dash", "powershell", "pwsh", "cmd"} +_INLINE_CODE = { + "py": {"-c"}, + "python": {"-c"}, + "python3": {"-c"}, + "node": {"-e", "--eval"}, + "perl": {"-e"}, + "ruby": {"-e"}, + "php": {"-r"}, +} + + +def _segments(command: str) -> list[str]: + return [ + segment.strip() + for segment in _SHELL_OPERATORS.split(command) + if segment.strip() + ] + + +def _split_env_string(value: str) -> list[str]: + try: + return shlex.split(value, posix=True) + except ValueError: + return value.split() + + +def _executable(tokens: list[str]) -> tuple[str, list[str]]: + while tokens: + if _ENV_ASSIGNMENT.match(tokens[0]): + tokens = tokens[1:] + continue + head = Path(tokens[0]).name.lower() + if head not in _WRAPPERS: + break + tokens = tokens[1:] + + while tokens and tokens[0].startswith("-"): + option = tokens[0] + tokens = tokens[1:] + if option == "--": + break + if head == "env" and ( + option in {"-S", "--split-string"} + or option.startswith("--split-string=") + or (option.startswith("-S") and len(option) > 2) + ): + if option.startswith("--split-string="): + value = option.split("=", 1)[1] + elif option.startswith("-S") and len(option) > 2: + value = option[2:] + elif tokens: + value = tokens[0] + tokens = tokens[1:] + else: + break + tokens = _split_env_string(value) + tokens + continue + if option in _WRAPPER_VALUE_OPTIONS.get(head, set()) and tokens: + tokens = tokens[1:] + + if head == "timeout" and tokens and not tokens[0].startswith("-"): + tokens = tokens[1:] + + if not tokens: + return "", [] + return Path(tokens[0]).name.lower(), tokens[1:] + + +def _shell_has_command_flag(arguments: list[str]) -> bool: + for argument in arguments: + lowered = argument.lower() + if lowered in {"-c", "/c", "/k"}: + return True + if ( + lowered.startswith("-") + and not lowered.startswith("--") + and "c" in lowered[1:] + ): + return True + return False + + +def _has_inline_flag(arguments: list[str], flags: set[str]) -> bool: + short_flags = { + flag[1:] for flag in flags if flag.startswith("-") and not flag.startswith("--") + } + for argument in arguments: + lowered = argument.lower() + if not lowered.startswith("-"): + break + if lowered in flags: + return True + if lowered.startswith("-") and not lowered.startswith("--"): + if any(flag in lowered[1:] for flag in short_flags): + return True + return False def validate_command(command: str) -> tuple[bool, str]: """ - Validate a shell command before running it. + Best-effort shell guardrail for the bash tool. + + This is a denylist over parsed shell segments, not an OS sandbox: a + determined caller can still run arbitrary code through an allowed + interpreter. The agent-level approval flow is the real boundary. Rules: - Must be a non-empty string - Null bytes rejected - - Must not contain any blocked substring from config.tool.blocked_commands - - Tokens starting with `sudo` are blocked - - Length is capped (10 KB) to avoid abuse + - Command substitution (backticks, `$()`) rejected + - Blocked substrings from config.tool.blocked_commands rejected + - Privilege escalation (sudo/su/doas/pkexec) rejected anywhere, + including via env wrappers and absolute paths + - Nested shell `-c` and inline interpreter code (`python -c`, `node -e`) + rejected + - base64 decode piped into a shell rejected + - Length capped (10 KB) Returns: (is_valid, error_message) @@ -121,6 +269,9 @@ def validate_command(command: str) -> tuple[bool, str]: return False, "Invalid command (null byte)." if len(command) > 10_000: return False, "Command too long (max 10 KB)." + if _SUSPICIOUS_CHAR_RUNS.search(stripped): + log.warning(f"Command substitution blocked: {stripped[:80]}") + return False, "Command substitution (backticks or $()) is blocked." lower = stripped.lower() @@ -135,16 +286,30 @@ def validate_command(command: str) -> tuple[bool, str]: ), ) - # Block sudo and su outright - try: - tokens = shlex.split(stripped, posix=True) - except ValueError: - # Unbalanced quotes etc. — allow bash itself to error out - tokens = stripped.split() + if re.search(r"\bbase64\b[^|]*(?:-d|--decode)[^|]*\|\s*(?:ba|z|k)?sh\b", lower): + return False, "Decoding base64 into a shell is blocked." - if tokens and tokens[0] in ("sudo", "su", "doas"): - log.warning(f"Privilege escalation blocked: {stripped[:80]}") - return False, "Privilege escalation commands (sudo/su/doas) are blocked." + for segment in _segments(stripped): + try: + tokens = shlex.split(segment, posix=True) + except ValueError: + tokens = segment.split() + executable, arguments = _executable(tokens) + if not executable: + continue + if executable in _PRIVILEGE_TOOLS: + log.warning(f"Privilege escalation blocked: {stripped[:80]}") + return False, "Privilege escalation commands (sudo/su/doas) are blocked." + if executable == "eval": + return False, "Shell eval is blocked." + if executable in _SHELLS and _shell_has_command_flag(arguments): + return False, "Nested shell execution (-c) is blocked." + inline_flags = _INLINE_CODE.get(executable, set()) + if inline_flags and _has_inline_flag(arguments, inline_flags): + return False, ( + f"Inline code execution via {executable} is blocked; " + "write a script file instead." + ) return True, "" diff --git a/tests/test_security.py b/tests/test_security.py index c189278..c4b56c7 100644 --- a/tests/test_security.py +++ b/tests/test_security.py @@ -67,6 +67,55 @@ def test_blocks_shutdown(self): ok, err = validate_command("shutdown -h now") assert ok is False + def test_blocks_absolute_sudo(self): + ok, err = validate_command("/usr/bin/sudo id") + assert ok is False + + def test_blocks_sudo_through_env(self): + ok, err = validate_command("env FOO=bar sudo id") + assert ok is False + + def test_blocks_sudo_through_wrapper_options(self): + for command in ( + "env -u FOO sudo id", + "env -S 'sudo id'", + "time -f '%E' sudo id", + "timeout 5 sudo id", + "xargs -I {} sudo id", + "stdbuf -o 0 sudo id", + ): + ok, err = validate_command(command) + assert ok is False, (command, err) + + def test_blocks_nested_shell_command(self): + ok, err = validate_command("bash -lc 'sudo id'") + assert ok is False + ok, err = validate_command('cmd /k "sudo id"') + assert ok is False + + def test_blocks_inline_python(self): + ok, err = validate_command("python -c \"import shutil; shutil.rmtree('.')\"") + assert ok is False + ok, err = validate_command("py -c \"import os; os.system('calc')\"") + assert ok is False + + def test_blocks_base64_pipe_to_shell(self): + ok, err = validate_command("echo cm0gLXJmIC8= | base64 -d | bash") + assert ok is False + + def test_blocks_command_substitution(self): + ok, err = validate_command("echo $(sudo id)") + assert ok is False + + def test_allows_python_module_command(self): + ok, err = validate_command("python -m pytest -c pytest.ini -q") + assert ok is True + + def test_allows_regular_dev_commands(self): + for command in ("git status", "python --version", "pytest -q", "ls -la"): + ok, err = validate_command(command) + assert ok is True, (command, err) + class TestValidatePath: def test_accepts_normal_relative_path(self, tmp_path, monkeypatch): diff --git a/tests/test_tools.py b/tests/test_tools.py index 5848a6d..9ffc8e5 100644 --- a/tests/test_tools.py +++ b/tests/test_tools.py @@ -298,8 +298,9 @@ def test_skips_binary_files(self, tmp_path, monkeypatch): # ListFilesTool Tests # ============================================================ class TestListFilesTool: - def test_list_directory(self, tmp_path): + def test_list_directory(self, tmp_path, monkeypatch): """Should list all items in a directory""" + monkeypatch.chdir(tmp_path) (tmp_path / "file1.py").write_text("x=1") (tmp_path / "file2.txt").write_text("hello") (tmp_path / "subdir").mkdir() @@ -309,18 +310,21 @@ def test_list_directory(self, tmp_path): assert "file2.txt" in result assert "subdir" in result - def test_nonexistent_directory(self): + def test_nonexistent_directory(self, tmp_path, monkeypatch): """Should return an error for a missing directory""" - result = list_files_tool.invoke({"directory": "/fake/nonexistent/path"}) + monkeypatch.chdir(tmp_path) + result = list_files_tool.invoke({"directory": "missing-dir"}) assert "not found" in result.lower() - def test_empty_directory(self, tmp_path): + def test_empty_directory(self, tmp_path, monkeypatch): """Should report that an empty directory is empty""" + monkeypatch.chdir(tmp_path) result = list_files_tool.invoke({"directory": str(tmp_path)}) assert "empty" in result.lower() - def test_file_sizes_shown(self, tmp_path): + def test_file_sizes_shown(self, tmp_path, monkeypatch): """File sizes must be displayed""" + monkeypatch.chdir(tmp_path) (tmp_path / "small.txt").write_text("hi") result = list_files_tool.invoke({"directory": str(tmp_path)}) assert "B" in result or "KB" in result @@ -350,6 +354,50 @@ def test_tools_have_descriptions(self): assert t.description, f"{t.name} is missing a description!" +class TestReadToolTraversal: + def _setup(self, tmp_path, monkeypatch): + workspace = tmp_path / "workspace" + outside = tmp_path / "outside" + workspace.mkdir() + outside.mkdir() + (outside / "secret.txt").write_text("SECRET_TOKEN=abcd1234", encoding="utf-8") + (workspace / "notes.txt").write_text("a normal needle", encoding="utf-8") + monkeypatch.chdir(workspace) + return workspace, outside + + def test_grep_blocks_parent_directory(self, tmp_path, monkeypatch): + _, _ = self._setup(tmp_path, monkeypatch) + result = grep_tool.invoke( + {"pattern": "SECRET_TOKEN", "directory": "../outside"} + ) + assert "Security" in result + assert "abcd1234" not in result + + def test_grep_blocks_absolute_outside_path(self, tmp_path, monkeypatch): + _, outside = self._setup(tmp_path, monkeypatch) + result = grep_tool.invoke( + {"pattern": "SECRET_TOKEN", "directory": str(outside)} + ) + assert "Security" in result + assert "abcd1234" not in result + + def test_grep_allows_workspace(self, tmp_path, monkeypatch): + self._setup(tmp_path, monkeypatch) + result = grep_tool.invoke({"pattern": "needle", "directory": "."}) + assert "needle" in result + + def test_list_files_blocks_parent_directory(self, tmp_path, monkeypatch): + _, _ = self._setup(tmp_path, monkeypatch) + result = list_files_tool.invoke({"directory": "../outside"}) + assert "Security" in result + assert "secret.txt" not in result + + def test_list_files_allows_workspace(self, tmp_path, monkeypatch): + self._setup(tmp_path, monkeypatch) + result = list_files_tool.invoke({"directory": "."}) + assert "notes.txt" in result + + if __name__ == "__main__": import subprocess diff --git a/tools.py b/tools.py index 550ebbe..ba7fce6 100644 --- a/tools.py +++ b/tools.py @@ -341,11 +341,13 @@ def _grep_impl(pattern: str, directory: str = ".") -> str: return "Pattern is empty! Please provide a search pattern." try: - results: list[str] = [] - search_dir = Path(directory) + ok, error, search_dir = validate_path(directory, must_exist=True) + if not ok or search_dir is None: + return error + if not search_dir.is_dir(): + return f"Not a directory: {directory}" - if not search_dir.exists(): - return f"Directory not found: {directory}" + results: list[str] = [] max_results = config.tool.grep_max_results max_files = config.tool.grep_max_files @@ -361,6 +363,10 @@ def _grep_impl(pattern: str, directory: str = ".") -> str: continue if not file_path.is_file(): continue + try: + file_path.resolve().relative_to(search_dir) + except (OSError, ValueError): + continue suffix = file_path.suffix.lower() if suffix and suffix not in config.tool.text_extensions: @@ -409,9 +415,9 @@ def list_files_tool(directory: str = ".") -> str: @_timed("list_files_tool") def _list_files_impl(directory: str = ".") -> str: try: - path = Path(directory) - if not path.exists(): - return f"Directory not found: {directory}" + ok, error, path = validate_path(directory, must_exist=True) + if not ok or path is None: + return error if not path.is_dir(): return f"This is a file, not a directory: {directory}" @@ -420,6 +426,10 @@ def _list_files_impl(directory: str = ".") -> str: for item in sorted(path.iterdir()): if item.name in config.tool.skip_dirs or item.name.startswith("."): continue + try: + item.resolve().relative_to(path) + except (OSError, ValueError): + continue if item.is_dir(): items.append(f"[DIR] {item.name}/") else: