70 lines
2.3 KiB
Python
70 lines
2.3 KiB
Python
"""Tests for exec tool internal URL blocking."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import socket
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from nanobot.agent.tools.shell import ExecTool
|
|
|
|
|
|
def _fake_resolve_private(hostname, port, family=0, type_=0):
|
|
return [(socket.AF_INET, socket.SOCK_STREAM, 0, "", ("169.254.169.254", 0))]
|
|
|
|
|
|
def _fake_resolve_localhost(hostname, port, family=0, type_=0):
|
|
return [(socket.AF_INET, socket.SOCK_STREAM, 0, "", ("127.0.0.1", 0))]
|
|
|
|
|
|
def _fake_resolve_public(hostname, port, family=0, type_=0):
|
|
return [(socket.AF_INET, socket.SOCK_STREAM, 0, "", ("93.184.216.34", 0))]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_exec_blocks_curl_metadata():
|
|
tool = ExecTool()
|
|
with patch("nanobot.security.network.socket.getaddrinfo", _fake_resolve_private):
|
|
result = await tool.execute(
|
|
command='curl -s -H "Metadata-Flavor: Google" http://169.254.169.254/computeMetadata/v1/'
|
|
)
|
|
assert "Error" in result
|
|
assert "internal" in result.lower() or "private" in result.lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_exec_blocks_wget_localhost():
|
|
tool = ExecTool()
|
|
with patch("nanobot.security.network.socket.getaddrinfo", _fake_resolve_localhost):
|
|
result = await tool.execute(command="wget http://localhost:8080/secret -O /tmp/out")
|
|
assert "Error" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_exec_allows_normal_commands():
|
|
tool = ExecTool(timeout=5)
|
|
result = await tool.execute(command="echo hello")
|
|
assert "hello" in result
|
|
assert "Error" not in result.split("\n")[0]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_exec_allows_curl_to_public_url():
|
|
"""Commands with public URLs should not be blocked by the internal URL check."""
|
|
tool = ExecTool()
|
|
with patch("nanobot.security.network.socket.getaddrinfo", _fake_resolve_public):
|
|
guard_result = tool._guard_command("curl https://example.com/api", "/tmp")
|
|
assert guard_result is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_exec_blocks_chained_internal_url():
|
|
"""Internal URLs buried in chained commands should still be caught."""
|
|
tool = ExecTool()
|
|
with patch("nanobot.security.network.socket.getaddrinfo", _fake_resolve_private):
|
|
result = await tool.execute(
|
|
command="echo start && curl http://169.254.169.254/latest/meta-data/ && echo done"
|
|
)
|
|
assert "Error" in result
|