mirror of
https://github.com/HKUDS/nanobot.git
synced 2026-08-31 08:13:11 +03:00
fix(exec): disable command guard in full access
This commit is contained in:
@@ -470,10 +470,14 @@ class ExecTool(Tool):
|
|||||||
+ _WORKSPACE_BOUNDARY_NOTE
|
+ _WORKSPACE_BOUNDARY_NOTE
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# Full access is an explicit trust decision. Keep the application-level
|
||||||
|
# command guard aligned with the selected access mode instead of
|
||||||
|
# continuing to block commands after workspace restriction is disabled.
|
||||||
|
if access.restrict_to_workspace:
|
||||||
guard_error = self._guard_command(
|
guard_error = self._guard_command(
|
||||||
command,
|
command,
|
||||||
cwd,
|
cwd,
|
||||||
restrict_to_workspace=access.restrict_to_workspace,
|
restrict_to_workspace=True,
|
||||||
workspace_root=workspace_root,
|
workspace_root=workspace_root,
|
||||||
)
|
)
|
||||||
if guard_error:
|
if guard_error:
|
||||||
|
|||||||
@@ -30,7 +30,7 @@ def _fake_resolve_public(hostname, port, family=0, type_=0):
|
|||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_exec_blocks_curl_metadata():
|
async def test_exec_blocks_curl_metadata():
|
||||||
tool = ExecTool()
|
tool = ExecTool(restrict_to_workspace=True)
|
||||||
with patch("nanobot.security.network.socket.getaddrinfo", _fake_resolve_private):
|
with patch("nanobot.security.network.socket.getaddrinfo", _fake_resolve_private):
|
||||||
result = await tool.execute(
|
result = await tool.execute(
|
||||||
command='curl -s -H "Metadata-Flavor: Google" http://169.254.169.254/computeMetadata/v1/'
|
command='curl -s -H "Metadata-Flavor: Google" http://169.254.169.254/computeMetadata/v1/'
|
||||||
@@ -41,7 +41,7 @@ async def test_exec_blocks_curl_metadata():
|
|||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_exec_blocks_wget_localhost():
|
async def test_exec_blocks_wget_localhost():
|
||||||
tool = ExecTool()
|
tool = ExecTool(restrict_to_workspace=True)
|
||||||
with patch("nanobot.security.network.socket.getaddrinfo", _fake_resolve_localhost):
|
with patch("nanobot.security.network.socket.getaddrinfo", _fake_resolve_localhost):
|
||||||
result = await tool.execute(command="wget http://localhost:8080/secret -O /tmp/out")
|
result = await tool.execute(command="wget http://localhost:8080/secret -O /tmp/out")
|
||||||
assert "Error" in result
|
assert "Error" in result
|
||||||
@@ -111,6 +111,40 @@ def test_exec_full_workspace_scope_still_blocks_metadata(tmp_path):
|
|||||||
assert "internal/private" in error
|
assert "internal/private" in error
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
"command",
|
||||||
|
[
|
||||||
|
"echo blocked",
|
||||||
|
"echo http://169.254.169.254/latest/meta-data/",
|
||||||
|
],
|
||||||
|
)
|
||||||
|
async def test_exec_full_access_skips_command_guard(tmp_path, command):
|
||||||
|
tool = ExecTool(
|
||||||
|
working_dir=str(tmp_path),
|
||||||
|
restrict_to_workspace=False,
|
||||||
|
deny_patterns=[r"echo\s+blocked"],
|
||||||
|
)
|
||||||
|
result = await tool.execute(command=command)
|
||||||
|
|
||||||
|
assert "Exit code: 0" in result
|
||||||
|
assert "Command blocked" not in result
|
||||||
|
|
||||||
|
|
||||||
|
async def test_exec_full_workspace_scope_skips_command_guard(tmp_path):
|
||||||
|
tool = ExecTool(working_dir=str(tmp_path), restrict_to_workspace=True)
|
||||||
|
scope = build_workspace_scope(tmp_path, "full", source_channel="websocket")
|
||||||
|
token = bind_workspace_scope(scope)
|
||||||
|
try:
|
||||||
|
result = await tool.execute(
|
||||||
|
command="echo http://169.254.169.254/latest/meta-data/",
|
||||||
|
)
|
||||||
|
finally:
|
||||||
|
reset_workspace_scope(token)
|
||||||
|
|
||||||
|
assert "Exit code: 0" in result
|
||||||
|
assert "Command blocked" not in result
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_exec_allows_normal_commands():
|
async def test_exec_allows_normal_commands():
|
||||||
tool = ExecTool(timeout=5)
|
tool = ExecTool(timeout=5)
|
||||||
@@ -131,7 +165,7 @@ async def test_exec_allows_curl_to_public_url():
|
|||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_exec_blocks_chained_internal_url():
|
async def test_exec_blocks_chained_internal_url():
|
||||||
"""Internal URLs buried in chained commands should still be caught."""
|
"""Internal URLs buried in chained commands should still be caught."""
|
||||||
tool = ExecTool()
|
tool = ExecTool(restrict_to_workspace=True)
|
||||||
with patch("nanobot.security.network.socket.getaddrinfo", _fake_resolve_private):
|
with patch("nanobot.security.network.socket.getaddrinfo", _fake_resolve_private):
|
||||||
result = await tool.execute(
|
result = await tool.execute(
|
||||||
command="echo start && curl http://169.254.169.254/latest/meta-data/ && echo done"
|
command="echo start && curl http://169.254.169.254/latest/meta-data/ && echo done"
|
||||||
|
|||||||
@@ -525,6 +525,7 @@ def test_exec_session_mode_reuses_exec_safety_guard(tmp_path):
|
|||||||
tool = ExecTool(
|
tool = ExecTool(
|
||||||
working_dir=str(tmp_path),
|
working_dir=str(tmp_path),
|
||||||
deny_patterns=[r"echo\s+blocked"],
|
deny_patterns=[r"echo\s+blocked"],
|
||||||
|
restrict_to_workspace=True,
|
||||||
session_manager=manager,
|
session_manager=manager,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user