mirror of
https://github.com/HKUDS/nanobot.git
synced 2026-08-06 09:28:34 +00:00
fix(exec): allow scoped tmp cleanup commands
This commit is contained in:
parent
9d830fb6b6
commit
c2b03c5149
@ -40,6 +40,10 @@ from nanobot.security.workspace_policy import is_path_within
|
|||||||
|
|
||||||
_IS_WINDOWS = sys.platform == "win32"
|
_IS_WINDOWS = sys.platform == "win32"
|
||||||
|
|
||||||
|
_RM_COMMAND_RE = re.compile(r"\brm\b")
|
||||||
|
_SHELL_COMMAND_SEPARATOR_RE = re.compile(r"(?:&&|\|\||[;&|\r\n])")
|
||||||
|
_SHELL_TOKEN_RE = re.compile(r'''"[^"]*"|'[^']*'|[^\s]+''')
|
||||||
|
|
||||||
|
|
||||||
def _reap_pid(pid: int) -> None:
|
def _reap_pid(pid: int) -> None:
|
||||||
"""Best-effort ``waitpid`` to reap a child and prevent zombies.
|
"""Best-effort ``waitpid`` to reap a child and prevent zombies.
|
||||||
@ -210,7 +214,6 @@ class ExecTool(Tool):
|
|||||||
self.working_dir = working_dir
|
self.working_dir = working_dir
|
||||||
self.sandbox = sandbox
|
self.sandbox = sandbox
|
||||||
self.deny_patterns = (deny_patterns or []) + [
|
self.deny_patterns = (deny_patterns or []) + [
|
||||||
r"\brm\s+-[rf]{1,2}\b", # rm -r, rm -rf, rm -fr
|
|
||||||
r"\bdel\s+/[fq]\b", # del /f, del /q
|
r"\bdel\s+/[fq]\b", # del /f, del /q
|
||||||
r"\brmdir\s+/s\b", # rmdir /s
|
r"\brmdir\s+/s\b", # rmdir /s
|
||||||
r"(?:^|[;&|]\s*)format(?!=)\b", # format (as standalone command only)
|
r"(?:^|[;&|]\s*)format(?!=)\b", # format (as standalone command only)
|
||||||
@ -704,6 +707,74 @@ class ExecTool(Tool):
|
|||||||
env[key] = val
|
env[key] = val
|
||||||
return env
|
return env
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def _contains_unscoped_recursive_rm(cls, command: str) -> bool:
|
||||||
|
"""Return whether ``command`` contains recursive rm outside a scoped /tmp target.
|
||||||
|
|
||||||
|
The exec guard deliberately remains conservative for recursive deletion, but
|
||||||
|
test and build scripts routinely clean their own named directories below
|
||||||
|
``/tmp``. Treat only static, direct ``/tmp/<name>`` targets as scoped cleanup.
|
||||||
|
Any ambiguous invocation (variables, traversal, broad globs, nested paths,
|
||||||
|
mixed targets) stays blocked.
|
||||||
|
"""
|
||||||
|
for match in _RM_COMMAND_RE.finditer(command):
|
||||||
|
tail = command[match.end():]
|
||||||
|
segment = _SHELL_COMMAND_SEPARATOR_RE.split(tail, maxsplit=1)[0]
|
||||||
|
tokens = _SHELL_TOKEN_RE.findall(segment)
|
||||||
|
recursive = False
|
||||||
|
targets: list[str] = []
|
||||||
|
parsing_options = True
|
||||||
|
unsafe_redirect = False
|
||||||
|
|
||||||
|
for raw_token in tokens:
|
||||||
|
token = raw_token.strip().strip("\"'")
|
||||||
|
if not token:
|
||||||
|
continue
|
||||||
|
if token == "--" and parsing_options:
|
||||||
|
parsing_options = False
|
||||||
|
continue
|
||||||
|
if parsing_options and token.startswith("--"):
|
||||||
|
recursive = recursive or token == "--recursive"
|
||||||
|
continue
|
||||||
|
if parsing_options and re.fullmatch(r"-[a-z]+", token):
|
||||||
|
recursive = recursive or "r" in token[1:]
|
||||||
|
continue
|
||||||
|
|
||||||
|
parsing_options = False
|
||||||
|
if token.startswith("#"):
|
||||||
|
break
|
||||||
|
if re.match(r"^\d*[<>]", token):
|
||||||
|
redirect_target = re.sub(r"^\d*[<>]+", "", token)
|
||||||
|
if redirect_target and redirect_target != "/dev/null":
|
||||||
|
unsafe_redirect = True
|
||||||
|
continue
|
||||||
|
targets.append(token)
|
||||||
|
|
||||||
|
if recursive and (
|
||||||
|
unsafe_redirect
|
||||||
|
or not targets
|
||||||
|
or not all(cls._is_scoped_tmp_cleanup_target(target) for target in targets)
|
||||||
|
):
|
||||||
|
return True
|
||||||
|
|
||||||
|
return False
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _is_scoped_tmp_cleanup_target(raw_target: str) -> bool:
|
||||||
|
"""Accept a static, specifically named descendant of the POSIX /tmp root."""
|
||||||
|
target = raw_target.strip().rstrip("\"'),")
|
||||||
|
if not target.startswith("/tmp/"):
|
||||||
|
return False
|
||||||
|
|
||||||
|
relative = target.removeprefix("/tmp/")
|
||||||
|
if not relative or any(char in relative for char in ("$", "`", "\\", "[", "{")):
|
||||||
|
return False
|
||||||
|
if "/" in relative or relative in {".", ".."}:
|
||||||
|
return False
|
||||||
|
|
||||||
|
literal_prefix = re.split(r"[*?]", relative, maxsplit=1)[0]
|
||||||
|
return any(char.isalnum() or char in "_-" for char in literal_prefix)
|
||||||
|
|
||||||
def _guard_command(
|
def _guard_command(
|
||||||
self,
|
self,
|
||||||
command: str,
|
command: str,
|
||||||
@ -723,6 +794,9 @@ class ExecTool(Tool):
|
|||||||
re.fullmatch(p, lower) for p in self.allow_patterns
|
re.fullmatch(p, lower) for p in self.allow_patterns
|
||||||
)
|
)
|
||||||
if not explicitly_allowed:
|
if not explicitly_allowed:
|
||||||
|
if self._contains_unscoped_recursive_rm(lower):
|
||||||
|
return ToolResult.error("Error: Command blocked by deny pattern filter")
|
||||||
|
|
||||||
for pattern in self.deny_patterns:
|
for pattern in self.deny_patterns:
|
||||||
if re.search(pattern, lower):
|
if re.search(pattern, lower):
|
||||||
return ToolResult.error("Error: Command blocked by deny pattern filter")
|
return ToolResult.error("Error: Command blocked by deny pattern filter")
|
||||||
|
|||||||
@ -2,28 +2,94 @@
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import shlex
|
||||||
|
import sys
|
||||||
|
import tempfile
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
from nanobot.agent.tools.shell import ExecTool
|
from nanobot.agent.tools.shell import ExecTool
|
||||||
|
|
||||||
|
|
||||||
def test_deny_patterns_block_rm_rf():
|
def test_deny_patterns_block_rm_rf():
|
||||||
"""Baseline: rm -rf is blocked by default deny list."""
|
"""Baseline: rm -rf is blocked by default deny list."""
|
||||||
tool = ExecTool()
|
tool = ExecTool()
|
||||||
result = tool._guard_command("rm -rf /tmp/build", "/tmp")
|
result = tool._guard_command("rm -rf /", "/tmp")
|
||||||
assert result is not None
|
assert result is not None
|
||||||
assert "deny pattern filter" in result.lower()
|
assert "deny pattern filter" in result.lower()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
"command",
|
||||||
|
[
|
||||||
|
"rm -rf /tmp/nanobot-test",
|
||||||
|
"rm -fr /tmp/nanobot-test-*",
|
||||||
|
"rm --recursive --force /tmp/nanobot-test-cache",
|
||||||
|
"echo setup && rm -rf /tmp/nanobot-test; echo done",
|
||||||
|
"bash -lc 'pytest tests; rm -rf /tmp/nanobot-test'",
|
||||||
|
"rm -rf '/tmp/nanobot test' 2>/dev/null",
|
||||||
|
],
|
||||||
|
)
|
||||||
|
def test_deny_patterns_allow_scoped_tmp_cleanup(command):
|
||||||
|
"""Named, static /tmp descendants are safe enough for test cleanup."""
|
||||||
|
tool = ExecTool()
|
||||||
|
assert tool._guard_command(command, "/tmp") is None
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
"command",
|
||||||
|
[
|
||||||
|
"rm -rf /tmp",
|
||||||
|
"rm -rf /tmp/*",
|
||||||
|
"rm -rf /tmp/nanobot-test/../../etc",
|
||||||
|
"rm -rf /tmp/nanobot-test/cache",
|
||||||
|
"rm -rf /tmp/$TARGET",
|
||||||
|
"rm -rf /tmp/nanobot-test /etc",
|
||||||
|
"rm -rf /tmp/nanobot-test >/etc/passwd",
|
||||||
|
"echo setup && rm -rf /etc",
|
||||||
|
],
|
||||||
|
)
|
||||||
|
def test_deny_patterns_block_unscoped_recursive_rm(command):
|
||||||
|
"""Broad, dynamic, traversing, or mixed recursive deletions remain blocked."""
|
||||||
|
tool = ExecTool()
|
||||||
|
result = tool._guard_command(command, "/tmp")
|
||||||
|
assert result is not None
|
||||||
|
assert "deny pattern filter" in result.lower()
|
||||||
|
|
||||||
|
|
||||||
|
def test_deny_patterns_allow_non_recursive_rm_f():
|
||||||
|
"""The recursive-delete guard must not mistake rm -f for rm -rf."""
|
||||||
|
tool = ExecTool()
|
||||||
|
assert tool._guard_command("rm -f /tmp/nanobot-test.log", "/tmp") is None
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
@pytest.mark.skipif(sys.platform == "win32", reason="POSIX rm and /tmp syntax")
|
||||||
|
async def test_exec_runs_scoped_tmp_cleanup():
|
||||||
|
"""A real exec call can remove its own directly named temporary directory."""
|
||||||
|
with tempfile.TemporaryDirectory(prefix="nanobot-exec-cleanup-", dir="/tmp") as temp_dir:
|
||||||
|
target = Path(temp_dir)
|
||||||
|
(target / "scratch.txt").write_text("scratch")
|
||||||
|
tool = ExecTool(timeout=5)
|
||||||
|
|
||||||
|
result = await tool.execute(command=f"rm -rf {shlex.quote(temp_dir)}")
|
||||||
|
|
||||||
|
assert "deny pattern filter" not in result.lower()
|
||||||
|
assert not target.exists()
|
||||||
|
|
||||||
|
|
||||||
def test_allow_patterns_bypass_deny():
|
def test_allow_patterns_bypass_deny():
|
||||||
"""allow_patterns take priority: matching command skips deny check."""
|
"""allow_patterns take priority: matching command skips deny check."""
|
||||||
tool = ExecTool(allow_patterns=[r"rm\s+-rf\s+/tmp/.*"])
|
tool = ExecTool(allow_patterns=[r"rm\s+-rf\s+/opt/build"])
|
||||||
result = tool._guard_command("rm -rf /tmp/build", "/tmp")
|
result = tool._guard_command("rm -rf /opt/build", "/tmp")
|
||||||
assert result is None
|
assert result is None
|
||||||
|
|
||||||
|
|
||||||
def test_allow_patterns_must_match_to_bypass():
|
def test_allow_patterns_must_match_to_bypass():
|
||||||
"""Non-matching allow_patterns do NOT bypass deny."""
|
"""Non-matching allow_patterns do NOT bypass deny."""
|
||||||
tool = ExecTool(allow_patterns=[r"rm\s+-rf\s+/opt/"])
|
tool = ExecTool(allow_patterns=[r"rm\s+-rf\s+/tmp/build"])
|
||||||
result = tool._guard_command("rm -rf /tmp/build", "/tmp")
|
result = tool._guard_command("rm -rf /opt/build", "/tmp")
|
||||||
assert result is not None
|
assert result is not None
|
||||||
assert "deny pattern filter" in result.lower()
|
assert "deny pattern filter" in result.lower()
|
||||||
|
|
||||||
@ -34,7 +100,15 @@ def test_extra_deny_patterns_from_config():
|
|||||||
# ping is blocked by extra deny
|
# ping is blocked by extra deny
|
||||||
assert tool._guard_command("ping example.com", "/tmp") is not None
|
assert tool._guard_command("ping example.com", "/tmp") is not None
|
||||||
# rm -rf still blocked by built-in deny
|
# rm -rf still blocked by built-in deny
|
||||||
assert tool._guard_command("rm -rf /tmp/x", "/tmp") is not None
|
assert tool._guard_command("rm -rf /", "/tmp") is not None
|
||||||
|
|
||||||
|
|
||||||
|
def test_extra_deny_patterns_can_block_scoped_tmp_cleanup():
|
||||||
|
"""User-configured policy still takes precedence over the built-in exception."""
|
||||||
|
tool = ExecTool(deny_patterns=[r"\brm\b"])
|
||||||
|
result = tool._guard_command("rm -rf /tmp/nanobot-test", "/tmp")
|
||||||
|
assert result is not None
|
||||||
|
assert "deny pattern filter" in result.lower()
|
||||||
|
|
||||||
|
|
||||||
def test_allow_patterns_bypass_extra_deny():
|
def test_allow_patterns_bypass_extra_deny():
|
||||||
@ -84,6 +158,6 @@ def test_deny_patterns_search_original_command_with_quoted_hash():
|
|||||||
|
|
||||||
def test_allow_patterns_fullmatch_allows_exact_command():
|
def test_allow_patterns_fullmatch_allows_exact_command():
|
||||||
"""A full-command allow pattern can still exempt an exact denied command."""
|
"""A full-command allow pattern can still exempt an exact denied command."""
|
||||||
tool = ExecTool(allow_patterns=[r"rm\s+-rf\s+/tmp/build"])
|
tool = ExecTool(allow_patterns=[r"rm\s+-rf\s+/opt/build"])
|
||||||
result = tool._guard_command("rm -rf /tmp/build", "/tmp")
|
result = tool._guard_command("rm -rf /opt/build", "/tmp")
|
||||||
assert result is None
|
assert result is None
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user