diff --git a/tests/test_admin_listing_authorization.py b/tests/test_admin_listing_authorization.py new file mode 100644 index 000000000..1fd3c4c84 --- /dev/null +++ b/tests/test_admin_listing_authorization.py @@ -0,0 +1,135 @@ +"""Real admin gates must run before MCP and diagnostic data access.""" + +import builtins +import json +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock + +import pytest +from fastapi import APIRouter, FastAPI +from fastapi.testclient import TestClient + +from core.auth import AuthManager +from routes.auth_routes import SESSION_COOKIE +from tests.helpers.database import disposable_database + + +@pytest.fixture +def authenticated_app(tmp_path, monkeypatch): + monkeypatch.setenv("AUTH_ENABLED", "true") + auth = AuthManager(str(tmp_path / "auth.json")) + for user, admin in (("member", False), ("admin", True)): + assert auth.create_user(user, "disposable-test-password", is_admin=admin) + app = FastAPI() + app.state.auth_manager = auth + + @app.middleware("http") + async def fixture_session(request, call_next): + # Supply the authenticated principal, without replacing require_admin. + request.state.current_user = auth.get_username_for_token(request.cookies.get(SESSION_COOKIE)) + return await call_next(request) + + def client_for(user): + client = TestClient(app) + if user != "anonymous": + token = auth.create_session(user, "disposable-test-password") + assert token + client.cookies.set(SESSION_COOKIE, token) + return client + + return app, client_for + + +@pytest.mark.parametrize("user", ["anonymous", "member", "admin"]) +@pytest.mark.parametrize("path", ["/servers", "/tools", "/servers/private-server/tools"]) +def test_mcp_listing_requires_real_admin_before_data_access( + tmp_path, monkeypatch, authenticated_app, user, path, +): + from core.database import McpServer + from routes.mcp import mcp_routes + + app, client_for = authenticated_app + with disposable_database(tmp_path) as factory: + with factory() as db: + db.add(McpServer( + id="private-server", name="private-config-marker", transport="stdio", + command="fixture-only", args="[]", env=json.dumps({"FIXTURE": "private-config-marker"}), + )) + db.commit() + session = Mock(wraps=factory) + manager = SimpleNamespace( + get_server_status=Mock(return_value={"status": "connected", "tool_count": 1}), + get_all_tools=Mock(return_value=[{ + "server_id": "private-server", "name": "private-tool", "description": "private-config-marker", + }]), + ) + # setup_mcp_routes otherwise appends to a shared process-global router. + monkeypatch.setattr(mcp_routes, "router", APIRouter(prefix="/api/mcp")) + monkeypatch.setattr(mcp_routes, "SessionLocal", session) + app.include_router(mcp_routes.setup_mcp_routes(manager)) + with client_for(user) as client: + response = client.get("/api/mcp" + path) + if user == "admin": + assert response.status_code == 200 + assert "private-config-marker" in response.text + payload = response.json() + if path == "/servers": + assert payload[0]["env"] == {"FIXTURE": "private-config-marker"} + else: + assert payload[0]["name"] == "private-tool" + else: + assert response.status_code == 403 + assert response.json() == {"detail": "Admin only"} + assert "private-config-marker" not in response.text + session.assert_not_called() + manager.get_server_status.assert_not_called() + manager.get_all_tools.assert_not_called() + + +@pytest.mark.parametrize("user", ["anonymous", "member", "admin"]) +@pytest.mark.parametrize("method,path", [ + ("GET", "/api/db/stats"), + ("GET", "/api/rag/stats"), + ("GET", "/api/test/youtube?url=https://youtu.be/fixture"), + ("POST", "/api/test-research"), + ("GET", "/api/diagnostics/services"), + ("GET", "/api/diagnostics/logs"), +]) +def test_diagnostics_require_real_admin_before_sensitive_work( + tmp_path, monkeypatch, authenticated_app, user, method, path, +): + import core.database as database + import src.service_health as health + from routes import diagnostics_routes as routes + + app, client_for = authenticated_app + marker = "private-diagnostics-marker" + stats = Mock(return_value={"marker": marker}) + rag = SimpleNamespace(get_stats=Mock(return_value={"marker": marker})) + research = SimpleNamespace(call_research_service=AsyncMock(return_value=marker)) + extract_id = Mock(return_value="fixture") + transcript = AsyncMock(return_value={"success": True, "transcript": marker}) + collect = AsyncMock(return_value={"marker": marker}) + log_dir = tmp_path / "logs" + log_dir.mkdir() + (log_dir / "app.log").write_text(marker + "\n", encoding="utf-8") + read_log = Mock(wraps=builtins.open) + monkeypatch.setattr(database, "get_detailed_stats", stats) + monkeypatch.setattr(routes, "extract_youtube_id", extract_id) + monkeypatch.setattr(routes, "extract_transcript_async", transcript) + monkeypatch.setattr(health, "collect_service_health", collect) + monkeypatch.setattr(routes, "DATA_DIR", str(tmp_path)) + monkeypatch.setattr(routes, "open", read_log, raising=False) + app.include_router(routes.setup_diagnostics_routes(rag, True, research)) + with client_for(user) as client: + response = client.request(method, path) + if user == "admin": + assert response.status_code == 200 + assert marker in response.text + else: + assert response.status_code == 403 + assert response.json() == {"detail": "Admin only"} + assert marker not in response.text + for operation in (stats, rag.get_stats, research.call_research_service, + extract_id, transcript, collect, read_log): + operation.assert_not_called()