mirror of
https://github.com/HKUDS/nanobot.git
synced 2026-08-12 23:29:16 +03:00
fix(exec): parse shell path boundaries safely
Co-authored-by: santhreal <64453045+santhreal@users.noreply.github.com>
This commit is contained in:
+108
-11
@@ -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"(?<![A-Za-z])(?:[A-Za-z]:[^\s\"'|><;]*|\\\\[^\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]:
|
||||
|
||||
@@ -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):
|
||||
|
||||
Reference in New Issue
Block a user