diff --git a/nanobot/agent/tools/shell.py b/nanobot/agent/tools/shell.py index 4451b93b4..b19c0de86 100644 --- a/nanobot/agent/tools/shell.py +++ b/nanobot/agent/tools/shell.py @@ -5,6 +5,7 @@ from __future__ import annotations import asyncio import os import re +import shlex import shutil import signal import subprocess @@ -13,6 +14,7 @@ from contextlib import suppress from dataclasses import dataclass from pathlib import Path, PureWindowsPath from typing import Any, Protocol, cast +from urllib.parse import unquote from loguru import logger from pydantic import Field @@ -1010,18 +1012,113 @@ class ExecTool(Tool): r"(?<;]*|\\\\[^\s\"'|><;]+(?:\\[^\s\"'|><;]+)*)", command ) - posix_paths = [ - p.rstrip(");},") - for p in re.findall( - r"(?:^|[\s|><='\"({,]|:(?!//))(/[^\"'>;|<()\s]+)", - command, + try: + lexer = shlex.shlex(command, posix=True, punctuation_chars="();<>|&") + lexer.whitespace_split = True + lexer.commenters = "" + tokens = list(lexer) + except ValueError: + # Keep malformed quoting fail-closed. The shell will normally reject + # it too, but a conservative raw scan must not turn it into a bypass. + tokens = [command] + + paths = [*win_paths] + seen = set(win_paths) + for index, token in enumerate(tokens): + for path in ExecTool._extract_posix_paths_from_token(token): + if path not in seen: + paths.append(path) + seen.add(path) + if index > 0 and tokens[index - 1] in {"-c", "-lc", "--command"}: + for path in ExecTool._extract_absolute_paths(token): + if path not in seen: + paths.append(path) + seen.add(path) + return paths + + @staticmethod + def _extract_posix_paths_from_token(token: str) -> list[str]: + """Extract local POSIX/home paths from one shell-decoded token. + + ``shlex`` separates real grouping/redirection operators while preserving + parentheses and spaces that were quoted or escaped as part of a path. + Embedded scripts (for example ``sh -c \"cat /tmp/x\"``) still need a + small boundary scan. Colons are deliberately not boundaries: treating + them as such misclassifies URLs, ``host:/remote`` and ``C:/Windows``. + """ + paths: list[str] = [] + for match in re.finditer( + r"file://(?:[^/\s\"']+)?(/[^\s\"'<>|;&]*)", + token, + flags=re.IGNORECASE, + ): + uri_prefix = token[: match.start()] + raw_path = match.group(1) + if uri_prefix.count("(") > uri_prefix.count(")"): + raw_path = raw_path.split(")", 1)[0] + if uri_prefix.count("{") > uri_prefix.count("}"): + raw_path = raw_path.split(",", 1)[0].split("}", 1)[0] + raw_path = raw_path.split("?", 1)[0].split("#", 1)[0] + if raw_path: + paths.append(unquote(raw_path)) + boundary_chars = frozenset(" \t\r\n=({,<>|;&\"'") + i = 0 + while i < len(token): + is_posix = token[i] == "/" + is_home = token.startswith("~/", i) or token.startswith("~+/", i) + if not is_posix and not is_home: + i += 1 + continue + + prefix = token[:i] + at_boundary = i == 0 or token[i - 1] in boundary_chars + parameter_default = ( + i >= 2 and token[i - 2] == ":" and token[i - 1] in "-+?=" ) - ] - home_paths = [ - p.rstrip(");},") - for p in re.findall(r"(?:^|[\s|><='\"({,:])(~[/+][^\"'>;|<()\s]+)", command) - ] - return win_paths + posix_paths + home_paths + if not at_boundary and not parameter_default: + i += 1 + continue + + word_start = max( + (prefix.rfind(char) for char in " \t\r\n<>|;&"), + default=-1, + ) + 1 + word_prefix = prefix[word_start:] + if re.search(r"[A-Za-z][A-Za-z0-9+.-]*://", word_prefix) or re.match( + r"(?:[^/:=\s]+@)?[^/:=\s]+:$", + word_prefix, + ): + # HTTP-style URL path/query fragments and scp-style remote paths + # are not local filesystem references. ``file://`` paths were + # decoded above. Windows drive paths are already captured by the + # platform-specific expression above. + i += 1 + continue + + assignment_value = i > 0 and token[i - 1] == "=" and not any( + char.isspace() for char in prefix + ) + if i == 0 or assignment_value: + end = len(token) + elif token[i - 1] in {"'", '"'}: + quote = token[i - 1] + closing = token.find(quote, i) + end = len(token) if closing < 0 else closing + else: + end_chars = set(" \t\r\n\"'<>|;&") + if prefix.count("(") > prefix.count(")"): + end_chars.add(")") + if prefix.count("{") > prefix.count("}"): + end_chars.update({",", "}"}) + end = i + while end < len(token) and token[end] not in end_chars: + end += 1 + + candidate = token[i:end] + if candidate: + paths.append(candidate) + i = max(end, i + 1) + return paths @staticmethod def _normalize_bind_roots(paths: list[str] | None) -> list[Path]: diff --git a/tests/tools/test_exec_security.py b/tests/tools/test_exec_security.py index aba8c1d65..5c0c4c49d 100644 --- a/tests/tools/test_exec_security.py +++ b/tests/tools/test_exec_security.py @@ -521,6 +521,161 @@ def test_exec_blocks_outside_paths_with_redirection_and_delimiters(tmp_path): assert "path outside working dir" in result +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX symlink and quoting semantics") +@pytest.mark.parametrize("quoted", [True, False]) +def test_exec_does_not_truncate_parentheses_in_symlink_paths(tmp_path, quoted): + workspace = tmp_path / "workspace" + outside = tmp_path / "outside" + workspace.mkdir() + outside.mkdir() + (outside / "secret.txt").write_text("secret") + link = workspace / "linked)dir" + link.symlink_to(outside, target_is_directory=True) + escaped_link = str(link).replace(")", r"\)") + rendered = f'"{link}/secret.txt"' if quoted else f"{escaped_link}/secret.txt" + command = f"cat {rendered}" + tool = ExecTool(working_dir=str(workspace), restrict_to_workspace=True) + + assert f"{link}/secret.txt" in tool._extract_absolute_paths(command) + result = tool._guard_command(command, str(workspace), workspace_root=str(workspace)) + + assert result is not None + assert "path outside working dir" in result + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX command substitution semantics") +def test_exec_checks_leaf_symlink_inside_command_substitution(tmp_path): + workspace = tmp_path / "workspace" + outside = tmp_path / "outside" + workspace.mkdir() + outside.mkdir() + link = workspace / "secret-link" + link.symlink_to(outside, target_is_directory=True) + command = f'cat "$(printf %s {link})"' + tool = ExecTool(working_dir=str(workspace), restrict_to_workspace=True) + + assert str(link) in tool._extract_absolute_paths(command) + result = tool._guard_command(command, str(workspace), workspace_root=str(workspace)) + + assert result is not None + assert "path outside working dir" in result + + +@pytest.mark.parametrize( + ("command", "not_a_posix_path"), + [ + ("curl https://example.com/outside/file", "/outside/file"), + ("curl 'https://example.com/?next=/etc/passwd'", "/etc/passwd"), + ("curl --url=https://example.com/?next=/etc/passwd", "/etc/passwd"), + ("scp host:/etc/passwd .", "/etc/passwd"), + ("echo C:/Windows/System32", "/Windows/System32"), + ], +) +def test_exec_does_not_misclassify_nonlocal_slash_strings(command, not_a_posix_path): + assert not_a_posix_path not in ExecTool._extract_absolute_paths(command) + + +def test_exec_extracts_quoted_path_with_shell_punctuation(): + path = "/tmp/a file)/with, punctuation" + + assert ExecTool._extract_absolute_paths(f'cat "{path}"') == [path] + + +@pytest.mark.parametrize("uri", ["file:///etc/passwd", "file://localhost/%65tc/passwd"]) +def test_exec_extracts_local_file_uri(uri): + assert "/etc/passwd" in ExecTool._extract_absolute_paths(f"curl {uri}") + + +def test_exec_blocks_file_uri_outside_workspace(tmp_path): + workspace = tmp_path / "workspace" + outside = tmp_path / "outside file.txt" + workspace.mkdir() + outside.write_text("secret") + tool = ExecTool(working_dir=str(workspace), restrict_to_workspace=True) + + result = tool._guard_command( + f"curl {outside.as_uri()}", + str(workspace), + workspace_root=str(workspace), + ) + + assert result is not None + assert "path outside working dir" in result + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX command substitution semantics") +def test_exec_checks_file_uri_leaf_symlink_inside_command_substitution(tmp_path): + workspace = tmp_path / "workspace" + outside = tmp_path / "outside" + workspace.mkdir() + outside.mkdir() + link = workspace / "secret-link" + link.symlink_to(outside, target_is_directory=True) + command = f'curl "$(printf file://{link})"' + tool = ExecTool(working_dir=str(workspace), restrict_to_workspace=True) + + assert str(link) in tool._extract_absolute_paths(command) + result = tool._guard_command(command, str(workspace), workspace_root=str(workspace)) + + assert result is not None + assert "path outside working dir" in result + + +def test_exec_keeps_quoted_parenthesis_path_inside_workspace_allowed(tmp_path): + workspace = tmp_path / "workspace" + inside = workspace / "linked)dir" / "file.txt" + inside.parent.mkdir(parents=True) + inside.write_text("safe") + tool = ExecTool(working_dir=str(workspace), restrict_to_workspace=True) + + assert tool._guard_command( + f'cat "{inside}"', + str(workspace), + workspace_root=str(workspace), + ) is None + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX symlink and assignment semantics") +def test_exec_keeps_quoted_assignment_punctuation_inside_workspace_allowed(tmp_path): + workspace = tmp_path / "workspace" + outside = tmp_path / "outside" + workspace.mkdir() + outside.mkdir() + (workspace / "linked").symlink_to(outside, target_is_directory=True) + inside = workspace / "linked;dir" / "file.txt" + inside.parent.mkdir() + inside.write_text("safe") + tool = ExecTool(working_dir=str(workspace), restrict_to_workspace=True) + + assert tool._guard_command( + f'x="{inside}"; cat "$x"', + str(workspace), + workspace_root=str(workspace), + ) is None + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX shell command-string semantics") +def test_exec_recursively_checks_compact_shell_command_string(tmp_path): + workspace = tmp_path / "workspace" + outside = tmp_path / "outside" + workspace.mkdir() + outside.mkdir() + link = workspace / "secret-link" + link.symlink_to(outside, target_is_directory=True) + command = f'sh -c "x={link};cat \\"$x\\""' + tool = ExecTool(working_dir=str(workspace), restrict_to_workspace=True) + + assert str(link) in tool._extract_absolute_paths(command) + result = tool._guard_command(command, str(workspace), workspace_root=str(workspace)) + + assert result is not None + assert "path outside working dir" in result + + +def test_exec_malformed_quote_still_extracts_path(): + assert "/etc/passwd" in ExecTool._extract_absolute_paths('cat "/etc/passwd') + + @pytest.mark.skipif(sys.platform == "win32", reason="POSIX double-slash path semantics") @pytest.mark.parametrize("path", ["//etc/passwd", "///etc/passwd"]) def test_exec_blocks_double_slash_absolute_paths(tmp_path, path):