mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-09 13:21:42 +00:00
81595cd588
Follow-up on the salvaged #47491 commits: - Register _plugin_api_runtime_gate BEFORE the auth middlewares so it executes AFTER them, and add an explicit auth check: unauthenticated requests to /api/plugins/<name>/ fall through to auth's 401 instead of this gate's 404. Prevents the gate from becoming a plugin-name oracle (an unauthenticated caller could otherwise fingerprint installed/enabled plugins by status code). Keeps test_non_kanban_plugin_route_requires_auth green. - Enable the 'example' user plugin in the _install_example_plugin test fixture so the auth / static-asset-allowlist tests still reach the real serving paths now that user plugins are gated on plugins.enabled. - Mark the runtime-gate unit-test scopes as authenticated so they exercise the enabled/disabled policy under the new auth-first ordering.
401 lines
15 KiB
Python
401 lines
15 KiB
Python
"""Regression tests for runtime plugin disable gating.
|
|
|
|
Covers two residual bypasses addressed in the PR:
|
|
|
|
1. Plugin API routes mounted at startup remain callable even after the
|
|
plugin is added to ``plugins.disabled`` at runtime. The new
|
|
``_plugin_api_runtime_gate`` middleware blocks these requests.
|
|
|
|
2. Bundled plugin assets were served from the unauthenticated
|
|
``/dashboard-plugins/{name}/{path}`` route even when the bundled
|
|
plugin was in ``plugins.disabled``. The updated ``serve_plugin_asset``
|
|
now applies the disabled check to bundled plugins too.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from unittest.mock import patch, AsyncMock
|
|
|
|
import pytest
|
|
|
|
from hermes_cli import web_server
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_plugin_cache():
|
|
"""Bust the plugin cache before and after each test."""
|
|
web_server._dashboard_plugins_cache = None
|
|
yield
|
|
web_server._dashboard_plugins_cache = None
|
|
|
|
|
|
@pytest.fixture
|
|
def test_client(monkeypatch, tmp_path):
|
|
"""Set up a Starlette TestClient with auth bypassed."""
|
|
try:
|
|
from starlette.testclient import TestClient
|
|
except ImportError:
|
|
pytest.skip("fastapi/starlette not installed")
|
|
|
|
from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN
|
|
|
|
# Isolate HERMES_HOME so config reads go to our tmp.
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "home"))
|
|
(tmp_path / "home").mkdir(parents=True)
|
|
|
|
client = TestClient(app)
|
|
client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN
|
|
return client
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_user_plugin(tmp_path, name="hot"):
|
|
"""Create a minimal user plugin with a JS asset."""
|
|
dashboard_dir = tmp_path / "plugins" / name / "dashboard"
|
|
dashboard_dir.mkdir(parents=True)
|
|
dist_dir = dashboard_dir / "dist"
|
|
dist_dir.mkdir()
|
|
(dist_dir / "index.js").write_text("console.log('hello');")
|
|
(dashboard_dir / "manifest.json").write_text(json.dumps({
|
|
"name": name,
|
|
"label": name.title(),
|
|
"entry": "dist/index.js",
|
|
}))
|
|
return dashboard_dir
|
|
|
|
|
|
def _make_bundled_plugin(tmp_path, name="bundledx"):
|
|
"""Create a minimal bundled plugin with a JS asset."""
|
|
dashboard_dir = tmp_path / "bundled" / name / "dashboard"
|
|
dashboard_dir.mkdir(parents=True)
|
|
dist_dir = dashboard_dir / "dist"
|
|
dist_dir.mkdir()
|
|
(dist_dir / "index.js").write_text("console.log('bundled');")
|
|
(dashboard_dir / "manifest.json").write_text(json.dumps({
|
|
"name": name,
|
|
"label": name.title(),
|
|
"entry": "dist/index.js",
|
|
}))
|
|
return dashboard_dir
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test 1: Runtime-disabled user plugin API routes return 404
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestPluginApiRuntimeGate:
|
|
"""After a user plugin is disabled at runtime, its mounted API routes
|
|
must return 404 — not 200 — even though the router was already
|
|
included at startup. The _plugin_api_runtime_gate middleware enforces
|
|
this at request time."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_blocks_disabled_user_plugin(self):
|
|
"""Middleware returns 404 for a user plugin added to disabled set."""
|
|
from starlette.requests import Request
|
|
from starlette.responses import JSONResponse
|
|
|
|
fake_plugin = {
|
|
"name": "hot",
|
|
"source": "user",
|
|
}
|
|
|
|
# Simulate a request to /api/plugins/hot/probe
|
|
scope = {
|
|
"type": "http",
|
|
"method": "GET",
|
|
"path": "/api/plugins/hot/probe",
|
|
"query_string": b"",
|
|
"headers": [],
|
|
"state": {"token_authenticated": True},
|
|
}
|
|
request = Request(scope)
|
|
|
|
call_next = AsyncMock(return_value=JSONResponse({"ok": True}))
|
|
|
|
with patch.object(web_server, "_get_dashboard_plugins", return_value=[fake_plugin]), \
|
|
patch("hermes_cli.plugins_cmd._get_enabled_set", return_value={"hot"}), \
|
|
patch("hermes_cli.plugins_cmd._get_disabled_set", return_value={"hot"}):
|
|
response = await web_server._plugin_api_runtime_gate(request, call_next)
|
|
|
|
assert response.status_code == 404
|
|
call_next.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_blocks_unenabled_user_plugin(self):
|
|
"""Middleware returns 404 when user plugin not in enabled set."""
|
|
from starlette.requests import Request
|
|
from starlette.responses import JSONResponse
|
|
|
|
fake_plugin = {
|
|
"name": "hot",
|
|
"source": "user",
|
|
}
|
|
|
|
scope = {
|
|
"type": "http",
|
|
"method": "GET",
|
|
"path": "/api/plugins/hot/probe",
|
|
"query_string": b"",
|
|
"headers": [],
|
|
"state": {"token_authenticated": True},
|
|
}
|
|
request = Request(scope)
|
|
|
|
call_next = AsyncMock(return_value=JSONResponse({"ok": True}))
|
|
|
|
with patch.object(web_server, "_get_dashboard_plugins", return_value=[fake_plugin]), \
|
|
patch("hermes_cli.plugins_cmd._get_enabled_set", return_value=set()), \
|
|
patch("hermes_cli.plugins_cmd._get_disabled_set", return_value=set()):
|
|
response = await web_server._plugin_api_runtime_gate(request, call_next)
|
|
|
|
assert response.status_code == 404
|
|
call_next.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_passes_enabled_user_plugin(self):
|
|
"""Middleware passes through for an enabled user plugin."""
|
|
from starlette.requests import Request
|
|
from starlette.responses import JSONResponse
|
|
|
|
fake_plugin = {
|
|
"name": "hot",
|
|
"source": "user",
|
|
}
|
|
|
|
scope = {
|
|
"type": "http",
|
|
"method": "GET",
|
|
"path": "/api/plugins/hot/probe",
|
|
"query_string": b"",
|
|
"headers": [],
|
|
"state": {"token_authenticated": True},
|
|
}
|
|
request = Request(scope)
|
|
|
|
expected_resp = JSONResponse({"ok": True})
|
|
call_next = AsyncMock(return_value=expected_resp)
|
|
|
|
with patch.object(web_server, "_get_dashboard_plugins", return_value=[fake_plugin]), \
|
|
patch("hermes_cli.plugins_cmd._get_enabled_set", return_value={"hot"}), \
|
|
patch("hermes_cli.plugins_cmd._get_disabled_set", return_value=set()):
|
|
response = await web_server._plugin_api_runtime_gate(request, call_next)
|
|
|
|
assert response is expected_resp
|
|
call_next.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_blocks_disabled_bundled_plugin(self):
|
|
"""Middleware returns 404 for a bundled plugin in disabled set."""
|
|
from starlette.requests import Request
|
|
from starlette.responses import JSONResponse
|
|
|
|
fake_plugin = {
|
|
"name": "bundledx",
|
|
"source": "bundled",
|
|
}
|
|
|
|
scope = {
|
|
"type": "http",
|
|
"method": "GET",
|
|
"path": "/api/plugins/bundledx/probe",
|
|
"query_string": b"",
|
|
"headers": [],
|
|
"state": {"token_authenticated": True},
|
|
}
|
|
request = Request(scope)
|
|
|
|
call_next = AsyncMock(return_value=JSONResponse({"ok": True}))
|
|
|
|
with patch.object(web_server, "_get_dashboard_plugins", return_value=[fake_plugin]), \
|
|
patch("hermes_cli.plugins_cmd._get_enabled_set", return_value=set()), \
|
|
patch("hermes_cli.plugins_cmd._get_disabled_set", return_value={"bundledx"}):
|
|
response = await web_server._plugin_api_runtime_gate(request, call_next)
|
|
|
|
assert response.status_code == 404
|
|
call_next.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_passes_enabled_bundled_plugin(self):
|
|
"""Middleware passes through for a bundled plugin not in disabled set."""
|
|
from starlette.requests import Request
|
|
from starlette.responses import JSONResponse
|
|
|
|
fake_plugin = {
|
|
"name": "bundledx",
|
|
"source": "bundled",
|
|
}
|
|
|
|
scope = {
|
|
"type": "http",
|
|
"method": "GET",
|
|
"path": "/api/plugins/bundledx/probe",
|
|
"query_string": b"",
|
|
"headers": [],
|
|
"state": {"token_authenticated": True},
|
|
}
|
|
request = Request(scope)
|
|
|
|
expected_resp = JSONResponse({"ok": True})
|
|
call_next = AsyncMock(return_value=expected_resp)
|
|
|
|
with patch.object(web_server, "_get_dashboard_plugins", return_value=[fake_plugin]), \
|
|
patch("hermes_cli.plugins_cmd._get_enabled_set", return_value=set()), \
|
|
patch("hermes_cli.plugins_cmd._get_disabled_set", return_value=set()):
|
|
response = await web_server._plugin_api_runtime_gate(request, call_next)
|
|
|
|
assert response is expected_resp
|
|
call_next.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_passes_non_plugin_api_routes(self):
|
|
"""Middleware ignores non-plugin API routes."""
|
|
from starlette.requests import Request
|
|
from starlette.responses import JSONResponse
|
|
|
|
scope = {
|
|
"type": "http",
|
|
"method": "GET",
|
|
"path": "/api/status",
|
|
"query_string": b"",
|
|
"headers": [],
|
|
"state": {"token_authenticated": True},
|
|
}
|
|
request = Request(scope)
|
|
|
|
expected_resp = JSONResponse({"ok": True})
|
|
call_next = AsyncMock(return_value=expected_resp)
|
|
|
|
response = await web_server._plugin_api_runtime_gate(request, call_next)
|
|
|
|
assert response is expected_resp
|
|
call_next.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_unknown_plugin_defaults_to_user_blocks(self):
|
|
"""Unknown plugin name (not in discovery cache) is treated as user
|
|
plugin and blocked when not enabled."""
|
|
from starlette.requests import Request
|
|
from starlette.responses import JSONResponse
|
|
|
|
scope = {
|
|
"type": "http",
|
|
"method": "GET",
|
|
"path": "/api/plugins/unknown/action",
|
|
"query_string": b"",
|
|
"headers": [],
|
|
"state": {"token_authenticated": True},
|
|
}
|
|
request = Request(scope)
|
|
|
|
call_next = AsyncMock(return_value=JSONResponse({"ok": True}))
|
|
|
|
with patch.object(web_server, "_get_dashboard_plugins", return_value=[]), \
|
|
patch("hermes_cli.plugins_cmd._get_enabled_set", return_value=set()), \
|
|
patch("hermes_cli.plugins_cmd._get_disabled_set", return_value=set()):
|
|
response = await web_server._plugin_api_runtime_gate(request, call_next)
|
|
|
|
assert response.status_code == 404
|
|
call_next.assert_not_called()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test 2: Disabled bundled plugin assets return 404
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestBundledPluginAssetGate:
|
|
"""Bundled plugins in ``plugins.disabled`` must have their static
|
|
assets blocked — not just hidden from the listing endpoint."""
|
|
|
|
def test_bundled_asset_returns_404_when_disabled(self, test_client, tmp_path, monkeypatch):
|
|
"""A disabled bundled plugin's JS asset must return 404."""
|
|
plugin_dir = _make_bundled_plugin(tmp_path, "bundledx")
|
|
|
|
fake_plugin = {
|
|
"name": "bundledx",
|
|
"label": "Bundledx",
|
|
"source": "bundled",
|
|
"entry": "dist/index.js",
|
|
"_dir": str(plugin_dir),
|
|
}
|
|
|
|
with patch.object(web_server, "_get_dashboard_plugins", return_value=[fake_plugin]):
|
|
# Sanity: asset is served when not disabled.
|
|
with patch(
|
|
"hermes_cli.plugins_cmd._get_enabled_set", return_value=set()
|
|
), patch(
|
|
"hermes_cli.plugins_cmd._get_disabled_set", return_value=set()
|
|
):
|
|
resp = test_client.get("/dashboard-plugins/bundledx/dist/index.js")
|
|
assert resp.status_code == 200, (
|
|
"Sanity: bundled plugin asset should be served when not disabled"
|
|
)
|
|
|
|
# Disable it.
|
|
with patch(
|
|
"hermes_cli.plugins_cmd._get_enabled_set", return_value=set()
|
|
), patch(
|
|
"hermes_cli.plugins_cmd._get_disabled_set", return_value={"bundledx"}
|
|
):
|
|
resp = test_client.get("/dashboard-plugins/bundledx/dist/index.js")
|
|
assert resp.status_code == 404, (
|
|
"Disabled bundled plugin asset must return 404"
|
|
)
|
|
|
|
def test_bundled_asset_served_when_not_disabled(self, test_client, tmp_path, monkeypatch):
|
|
"""Bundled plugin assets are served normally when not in disabled set."""
|
|
plugin_dir = _make_bundled_plugin(tmp_path, "goodbundled")
|
|
|
|
fake_plugin = {
|
|
"name": "goodbundled",
|
|
"label": "Good Bundled",
|
|
"source": "bundled",
|
|
"entry": "dist/index.js",
|
|
"_dir": str(plugin_dir),
|
|
}
|
|
|
|
with patch.object(web_server, "_get_dashboard_plugins", return_value=[fake_plugin]):
|
|
with patch(
|
|
"hermes_cli.plugins_cmd._get_enabled_set", return_value=set()
|
|
), patch(
|
|
"hermes_cli.plugins_cmd._get_disabled_set", return_value=set()
|
|
):
|
|
resp = test_client.get("/dashboard-plugins/goodbundled/dist/index.js")
|
|
assert resp.status_code == 200
|
|
|
|
def test_user_plugin_asset_still_gated(self, test_client, tmp_path, monkeypatch):
|
|
"""User plugins still require enabled set membership for assets."""
|
|
plugin_dir = _make_user_plugin(tmp_path, "userplugin")
|
|
|
|
fake_plugin = {
|
|
"name": "userplugin",
|
|
"label": "User Plugin",
|
|
"source": "user",
|
|
"entry": "dist/index.js",
|
|
"_dir": str(plugin_dir),
|
|
}
|
|
|
|
with patch.object(web_server, "_get_dashboard_plugins", return_value=[fake_plugin]):
|
|
# Not in enabled set → 404.
|
|
with patch(
|
|
"hermes_cli.plugins_cmd._get_enabled_set", return_value=set()
|
|
), patch(
|
|
"hermes_cli.plugins_cmd._get_disabled_set", return_value=set()
|
|
):
|
|
resp = test_client.get("/dashboard-plugins/userplugin/dist/index.js")
|
|
assert resp.status_code == 404
|
|
|
|
# In enabled set → 200.
|
|
with patch(
|
|
"hermes_cli.plugins_cmd._get_enabled_set", return_value={"userplugin"}
|
|
), patch(
|
|
"hermes_cli.plugins_cmd._get_disabled_set", return_value=set()
|
|
):
|
|
resp = test_client.get("/dashboard-plugins/userplugin/dist/index.js")
|
|
assert resp.status_code == 200
|