mirror of
https://github.com/pewdiepie-archdaemon/odysseus.git
synced 2026-10-29 09:12:24 +01:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
1ddda10716 |
@@ -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()
|
||||||
@@ -266,3 +266,9 @@ def test_more6_preferred_wind_is_present():
|
|||||||
assert 'data-filter-action="stylize-wind"' in TOPBAR
|
assert 'data-filter-action="stylize-wind"' in TOPBAR
|
||||||
assert "else if (action === 'stylize-wind') applyWind?.();" in WIRE
|
assert "else if (action === 'stylize-wind') applyWind?.();" in WIRE
|
||||||
assert "export function wind(snap, v, dst)" in EFFECTS
|
assert "export function wind(snap, v, dst)" in EFFECTS
|
||||||
|
|
||||||
|
|
||||||
|
def test_more6_sw_version_is_current():
|
||||||
|
m = re.search(r"const CACHE_NAME = '([^']+)'", SW)
|
||||||
|
assert m, "CACHE_NAME not found"
|
||||||
|
assert re.fullmatch(r"odysseus-v\d+-pr40-editor-precache", m.group(1)), m.group(1)
|
||||||
|
|||||||
Reference in New Issue
Block a user