mirror of
https://github.com/pewdiepie-archdaemon/odysseus.git
synced 2026-06-16 01:35:36 -04:00
fix: block app_api access to shell routes (#3225)
This commit is contained in:
@@ -115,6 +115,19 @@ def _install_core_auth_stub(monkeypatch):
|
||||
return auth_mod
|
||||
|
||||
|
||||
def _install_core_middleware_stub(monkeypatch):
|
||||
"""Install the narrow middleware surface needed by loopback tool tests."""
|
||||
core_mod = types.ModuleType("core")
|
||||
core_mod.__path__ = []
|
||||
middleware_mod = types.ModuleType("core.middleware")
|
||||
middleware_mod.INTERNAL_TOOL_HEADER = "X-Internal-Tool"
|
||||
middleware_mod.INTERNAL_TOOL_TOKEN = "test-token"
|
||||
core_mod.middleware = middleware_mod
|
||||
monkeypatch.setitem(sys.modules, "core", core_mod)
|
||||
monkeypatch.setitem(sys.modules, "core.middleware", middleware_mod)
|
||||
return middleware_mod
|
||||
|
||||
|
||||
def test_providers_requires_admin_before_discovery_and_cache(monkeypatch):
|
||||
_install_model_route_import_stubs(monkeypatch)
|
||||
import routes.model_routes as model_routes
|
||||
@@ -428,6 +441,78 @@ async def test_admin_agent_tools_require_admin(monkeypatch):
|
||||
assert "requires an admin" in result["error"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_app_api_blocks_shell_routes_before_loopback(monkeypatch):
|
||||
import httpx
|
||||
from src.tool_implementations import do_app_api
|
||||
|
||||
class UnexpectedAsyncClient:
|
||||
def __init__(self, *args, **kwargs):
|
||||
raise AssertionError("app_api should block shell routes before loopback")
|
||||
|
||||
monkeypatch.setattr(httpx, "AsyncClient", UnexpectedAsyncClient)
|
||||
|
||||
for path in ("/api/shell/exec", "api/shell/stream"):
|
||||
result = await do_app_api(
|
||||
json.dumps(
|
||||
{
|
||||
"action": "call",
|
||||
"method": "POST",
|
||||
"path": path,
|
||||
"body": {"command": "echo should-not-run"},
|
||||
}
|
||||
),
|
||||
owner="admin",
|
||||
)
|
||||
|
||||
assert result["exit_code"] == 1
|
||||
assert "Path blocked for safety" in result["error"]
|
||||
assert "Sensitive endpoints" in result["error"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_app_api_endpoint_discovery_hides_shell_routes(monkeypatch):
|
||||
_install_core_middleware_stub(monkeypatch)
|
||||
import httpx
|
||||
from src.tool_implementations import do_app_api
|
||||
|
||||
class FakeResponse:
|
||||
def json(self):
|
||||
return {
|
||||
"paths": {
|
||||
"/api/shell/exec": {"post": {"summary": "Execute Shell Command"}},
|
||||
"/api/shell/stream": {"post": {"summary": "Stream Shell Command"}},
|
||||
"/api/auth/settings": {"get": {"summary": "Auth Settings"}},
|
||||
"/api/cookbook/gpus": {"get": {"summary": "List GPUs"}},
|
||||
}
|
||||
}
|
||||
|
||||
class FakeAsyncClient:
|
||||
def __init__(self, *args, **kwargs):
|
||||
pass
|
||||
|
||||
async def __aenter__(self):
|
||||
return self
|
||||
|
||||
async def __aexit__(self, exc_type, exc, tb):
|
||||
return False
|
||||
|
||||
async def get(self, *args, **kwargs):
|
||||
return FakeResponse()
|
||||
|
||||
monkeypatch.setattr(httpx, "AsyncClient", FakeAsyncClient)
|
||||
|
||||
result = await do_app_api(json.dumps({"action": "endpoints"}), owner="admin")
|
||||
|
||||
assert result["exit_code"] == 0
|
||||
paths = {(endpoint["method"], endpoint["path"]) for endpoint in result["endpoints"]}
|
||||
assert ("GET", "/api/cookbook/gpus") in paths
|
||||
assert ("POST", "/api/shell/exec") not in paths
|
||||
assert ("POST", "/api/shell/stream") not in paths
|
||||
assert ("GET", "/api/auth/settings") not in paths
|
||||
assert all(not endpoint["path"].startswith("/api/shell") for endpoint in result["endpoints"])
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_public_agent_policy_blocks_sensitive_tools(monkeypatch):
|
||||
auth_mod = _install_core_auth_stub(monkeypatch)
|
||||
|
||||
Reference in New Issue
Block a user