from __future__ import annotations import asyncio from contextlib import asynccontextmanager from pathlib import Path from types import SimpleNamespace import pytest from fastapi import HTTPException from httpx import ASGITransport, AsyncClient from mcp import Client from backend.app.core.config import Settings from backend.app.core.security import require_service_request from backend.app.runner_statistics_mcp import RUNNER_STATISTICS_TOOL, create_runner_statistics_mcp STATISTICS_TOKEN = "runner-statistics-token-that-is-distinct-and-long" GOVERNED_MCP_TOKEN = "governed-operations-token-that-is-distinct-and-long" def _request(authorization: str | None = None): from starlette.requests import Request headers = [(b"authorization", authorization.encode("ascii"))] if authorization else [] return Request( { "type": "http", "method": "POST", "path": "/mcp-runner-statistics/", "headers": headers, "client": ("127.0.0.1", 12345), } ) def _snapshot() -> dict[str, object]: return { "schemaVersion": 1, "status": "healthy", "scope": "ci-fleet", "generatedAt": "2026-10-01T12:00:00+00:00", "budget": { "cpuLimit": 8, "memoryLimitBytes": 17_179_869_184, "activeJobLimit": 2, "heavyJobLimit": 1, "reservedCpu": 1, "reservedMemoryBytes": 2_147_483_648, "activeJobs": 0, "heavyJobs": 0, }, "host": { "cpuBusyPercent": None, "loadRatio": None, "availableMemoryBytes": None, "cacheFreeBytes": None, }, "profiles": [], "window": { "startAt": None, "endAt": None, "completedJobs": 0, "samples": 0, "coverage": "unavailable", }, "history": [], } def test_separate_mcp_catalog_and_tool_are_read_only(): server, _http_app = create_runner_statistics_mcp( snapshot_provider=_snapshot, allowed_hosts=["testserver"], allowed_origins=[], ) async def exercise() -> None: async with Client(server) as client: listing = await client.list_tools() assert [tool.name for tool in listing.tools] == [RUNNER_STATISTICS_TOOL] tool = listing.tools[0] assert tool.annotations.read_only_hint is True assert tool.annotations.destructive_hint is False result = await client.call_tool(RUNNER_STATISTICS_TOOL, {}) assert result.structured_content["scope"] == "ci-fleet" unknown = await client.call_tool("container.restart", {"target": "anything"}) assert unknown.is_error is True asyncio.run(exercise()) def test_dedicated_service_auth_requires_strong_distinct_token(monkeypatch): from backend.app.core import security from backend.app.core.config import settings monkeypatch.setattr(security, "settings", settings) monkeypatch.setattr(settings, "read_runner_statistics_token", lambda: STATISTICS_TOKEN) monkeypatch.setattr(settings, "mcp_token", GOVERNED_MCP_TOKEN) assert require_service_request(_request(f"Bearer {STATISTICS_TOKEN}"), "mcp_runner_statistics") == "mcp_runner_statistics" with pytest.raises(HTTPException) as wrong: require_service_request(_request(f"Bearer {GOVERNED_MCP_TOKEN}"), "mcp_runner_statistics") assert wrong.value.status_code == 403 monkeypatch.setattr(settings, "read_runner_statistics_token", lambda: GOVERNED_MCP_TOKEN) with pytest.raises(HTTPException) as collision: require_service_request(_request(f"Bearer {GOVERNED_MCP_TOKEN}"), "mcp_runner_statistics") assert collision.value.status_code == 503 monkeypatch.setattr(settings, "read_runner_statistics_token", lambda: "short") with pytest.raises(HTTPException) as weak: require_service_request(_request("Bearer short"), "mcp_runner_statistics") assert weak.value.status_code == 503 def test_statistics_token_file_is_private_and_uses_data_dir_default(tmp_path: Path, monkeypatch): monkeypatch.delenv("APPOPS_RUNNER_STATISTICS_TOKEN_FILE", raising=False) monkeypatch.delenv("APPOPS_RUNNER_STATISTICS_TOKEN", raising=False) token_path = tmp_path / "runner-statistics-token" token_path.write_text(STATISTICS_TOKEN, encoding="utf-8") if __import__("os").name != "nt": token_path.chmod(0o600) configured = Settings(data_dir=str(tmp_path), mcp_token=GOVERNED_MCP_TOKEN) assert configured.runner_statistics_token_file == str(token_path) assert configured.read_runner_statistics_token() == STATISTICS_TOKEN assert STATISTICS_TOKEN not in str(configured.redact_env_preview()) def test_statistics_token_reader_rejects_oversized_and_symlink_files(tmp_path: Path): from backend.app.core.config import _private_secret_file oversized = tmp_path / "oversized-token" oversized.write_text("x" * 513, encoding="utf-8") assert _private_secret_file(str(oversized)) is None target = tmp_path / "target-token" target.write_text(STATISTICS_TOKEN, encoding="utf-8") if __import__("os").name != "nt": target.chmod(0o600) linked = tmp_path / "linked-token" try: linked.symlink_to(target) except (OSError, NotImplementedError): pytest.skip("symlink creation is unavailable on this host") assert _private_secret_file(str(linked)) is None def test_token_file_can_be_provisioned_after_app_start_without_restart(tmp_path: Path, monkeypatch): monkeypatch.delenv("APPOPS_RUNNER_STATISTICS_TOKEN_FILE", raising=False) token_path = tmp_path / "runner-statistics-token" configured = Settings(data_dir=str(tmp_path), mcp_token=GOVERNED_MCP_TOKEN) assert configured.read_runner_statistics_token() is None token_path.write_text(STATISTICS_TOKEN, encoding="utf-8") if __import__("os").name != "nt": token_path.chmod(0o600) assert configured.read_runner_statistics_token() == STATISTICS_TOKEN def test_plaintext_environment_value_is_not_a_runner_statistics_credential(tmp_path: Path, monkeypatch): monkeypatch.setenv("APPOPS_RUNNER_STATISTICS_TOKEN", STATISTICS_TOKEN) monkeypatch.setenv("APPOPS_RUNNER_STATISTICS_TOKEN_FILE", str(tmp_path / "missing-token")) configured = Settings(data_dir=str(tmp_path), mcp_token=GOVERNED_MCP_TOKEN) assert configured.read_runner_statistics_token() is None def test_readonly_transport_adds_only_validated_auth_public_host(monkeypatch): from backend.app.core.config import settings from backend.app.runner_statistics_mcp import _readonly_allowed_hosts, _readonly_allowed_origins monkeypatch.setattr(settings, "mcp_allowed_hosts", ["localhost:*"]) monkeypatch.setattr(settings, "auth_public_url", "https://appops.example.test") monkeypatch.setattr(settings, "frontend_origin", "https://frontend.example.test") assert _readonly_allowed_hosts() == ["localhost:*", "appops.example.test"] assert _readonly_allowed_origins() == ["https://frontend.example.test", "https://appops.example.test"] assert settings.mcp_allowed_hosts == ["localhost:*"] monkeypatch.setattr(settings, "auth_public_url", "http://attacker.example.test") assert _readonly_allowed_hosts() == ["localhost:*"] assert _readonly_allowed_origins() == ["https://frontend.example.test"] monkeypatch.setattr(settings, "auth_public_url", "https://attacker_host.example.test") assert _readonly_allowed_hosts() == ["localhost:*"] def test_transport_factory_uses_readonly_hosts_without_widening_full_mcp(monkeypatch): from backend.app.core.config import settings from mcp.server import MCPServer from starlette.applications import Starlette captured: dict[str, object] = {} def capture_transport(self, **kwargs): captured.update(kwargs) return Starlette() monkeypatch.setattr(MCPServer, "streamable_http_app", capture_transport) monkeypatch.setattr(settings, "mcp_allowed_hosts", ["localhost:*"]) monkeypatch.setattr(settings, "auth_public_url", "https://appops.example.test") create_runner_statistics_mcp() transport_security = captured["transport_security"] assert transport_security.allowed_hosts == ["localhost:*", "appops.example.test"] assert settings.mcp_allowed_hosts == ["localhost:*"] def test_readonly_transport_preserves_explicit_port_and_ipv6(monkeypatch): from backend.app.core.config import settings from backend.app.runner_statistics_mcp import _readonly_allowed_hosts monkeypatch.setattr(settings, "mcp_allowed_hosts", []) monkeypatch.setattr(settings, "auth_public_url", "https://[2001:db8::1]:9443") assert _readonly_allowed_hosts() == ["[2001:db8::1]:9443"] def test_main_http_boundary_rejects_cross_tokens_and_hides_missing_endpoint(monkeypatch): from backend.app import main from backend.app.core.config import settings def no_session_for_machine_token(_request): raise HTTPException(401, "Sign in with Authentik or Microsoft Entra") monkeypatch.setattr(main, "authorize_request", no_session_for_machine_token) monkeypatch.setattr(settings, "mcp_enabled", True) monkeypatch.setattr(settings, "mcp_token", GOVERNED_MCP_TOKEN) monkeypatch.setattr(settings, "read_runner_statistics_token", lambda: STATISTICS_TOKEN) monkeypatch.setattr(settings, "mcp_allowed_hosts", ["testserver"]) async def exercise() -> None: transport = ASGITransport(app=main.app) async with AsyncClient(transport=transport, base_url="http://testserver") as client: stats_with_main_token = await client.post( "/mcp-runner-statistics/", headers={"Authorization": f"Bearer {GOVERNED_MCP_TOKEN}"}, json={}, ) main_with_stats_token = await client.post( "/mcp/", headers={"Authorization": f"Bearer {STATISTICS_TOKEN}"}, json={}, ) native_api_with_stats_token = await client.get( "/api", headers={"Authorization": f"Bearer {STATISTICS_TOKEN}"}, ) assert stats_with_main_token.status_code == 403 assert main_with_stats_token.status_code == 403 assert native_api_with_stats_token.status_code == 401 monkeypatch.setattr(settings, "read_runner_statistics_token", lambda: None) disabled = await client.post("/mcp-runner-statistics/", json={}) assert disabled.status_code == 404 asyncio.run(exercise()) @pytest.mark.parametrize( ("main_enabled", "expected"), [(False, ["statistics"]), (True, ["main", "statistics"])], ) def test_lifespan_always_runs_readonly_mcp_and_gates_governed_mcp(monkeypatch, main_enabled, expected): from backend.app import main from backend.app.core.config import settings entered: list[str] = [] @asynccontextmanager async def run_manager(name: str): entered.append(name) try: yield finally: entered.remove(name) monkeypatch.setattr(main, "appops_mcp", SimpleNamespace(session_manager=SimpleNamespace(run=lambda: run_manager("main")))) monkeypatch.setattr(main, "run_runner_statistics_mcp", lambda: run_manager("statistics")) monkeypatch.setattr(main, "init_db", lambda: None) monkeypatch.setattr(main, "Session", lambda _engine: SimpleNamespace(__enter__=lambda self: self, __exit__=lambda *_args: False)) monkeypatch.setattr(main, "restore_request_metrics", lambda _session: None) monkeypatch.setattr(main, "restore_tool_metrics", lambda _session: None) monkeypatch.setattr(main.knowledge_service, "ensure_index_current", lambda _session: None) monkeypatch.setattr(main, "persist_observability", lambda: None) monkeypatch.setattr(main, "set_session_running", lambda _value: None) monkeypatch.setattr(settings, "mcp_enabled", main_enabled) for field in ( "notification_interval_seconds", "connector_health_interval_seconds", "observability_persist_interval_seconds", "restore_drill_interval_seconds", "mcp_canary_interval_seconds", "discovery_interval_seconds", ): monkeypatch.setattr(settings, field, 0) async def exercise() -> None: async with main.lifespan(main.app): assert entered == expected asyncio.run(exercise())