Latte
cf3d2e17d3
The audit logger is opened at startup (exit 2 on failure) and obtained before every call, which is refused with audit_unavailable otherwise. Audit records only take validated path and sudo values and use the request id as correlation id. The rate limiter evicts idle keys, caps tracked keys, audits the first excess per key and window and uses the settings passed to the app. FORWARDED_ALLOW_IPS is now a setting, and startup warns about a missing host allow-list and unavailable OAuth.
429 lines
14 KiB
Python
429 lines
14 KiB
Python
"""Tests for the transport runners: fail-closed startup, uvicorn wiring, stdio, serve CLI."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from mcp import Client
|
|
from mcp.client.stdio import StdioServerParameters, stdio_client
|
|
|
|
from aegis_gitea_mcp import cli, transports
|
|
from aegis_gitea_mcp.cli import main
|
|
from aegis_gitea_mcp.config import Settings
|
|
from aegis_gitea_mcp.transports import (
|
|
HttpAuthExtras,
|
|
http_startup_errors,
|
|
load_oauth_components,
|
|
run_http,
|
|
run_stdio,
|
|
)
|
|
from tests.test_server_helpers import ANY_HOST, BEARER_TOKEN, GITEA_TOKEN, server_settings
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _env(mock_env: None) -> None:
|
|
"""Settings objects here are built explicitly; the audit logger still needs GITEA_URL."""
|
|
|
|
|
|
class UvicornSpy:
|
|
"""Replaces ``uvicorn.run`` and records its arguments."""
|
|
|
|
def __init__(self) -> None:
|
|
self.calls: list[tuple[Any, dict[str, Any]]] = []
|
|
|
|
def __call__(self, app: Any, **kwargs: Any) -> None:
|
|
self.calls.append((app, kwargs))
|
|
|
|
|
|
@pytest.fixture
|
|
def uvicorn_spy(monkeypatch: pytest.MonkeyPatch) -> UvicornSpy:
|
|
spy = UvicornSpy()
|
|
monkeypatch.setattr(transports.uvicorn, "run", spy)
|
|
return spy
|
|
|
|
|
|
def oauth_settings(**overrides: Any) -> Settings:
|
|
values: dict[str, Any] = {
|
|
"oauth_enabled": True,
|
|
"gitea_oauth_client_id": "client",
|
|
"gitea_oauth_client_secret": "s" * 32,
|
|
"token_encryption_key": "k" * 32,
|
|
"public_base_url": "https://mcp.example.com",
|
|
}
|
|
values.update(overrides)
|
|
return server_settings(**values)
|
|
|
|
|
|
# HTTP startup checks
|
|
|
|
|
|
def test_non_loopback_bind_without_auth_is_refused(
|
|
uvicorn_spy: UvicornSpy, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
assert run_http(server_settings(mcp_host=ANY_HOST)) == 2
|
|
|
|
captured = capsys.readouterr()
|
|
assert uvicorn_spy.calls == []
|
|
assert "without HTTP authentication" in captured.err
|
|
assert "MCP_BEARER_TOKEN" in captured.err
|
|
assert captured.out == ""
|
|
|
|
|
|
def test_non_loopback_bind_with_bearer_is_allowed(uvicorn_spy: UvicornSpy) -> None:
|
|
settings = server_settings(mcp_host=ANY_HOST, mcp_port=9000, mcp_bearer_token=BEARER_TOKEN)
|
|
|
|
assert run_http(settings) == 0
|
|
|
|
((app, kwargs),) = uvicorn_spy.calls
|
|
assert callable(app)
|
|
assert kwargs["host"] == ANY_HOST
|
|
assert kwargs["port"] == 9000
|
|
assert kwargs["proxy_headers"] is True
|
|
assert kwargs["forwarded_allow_ips"] == "127.0.0.1"
|
|
assert kwargs["log_level"] == "info"
|
|
assert kwargs["server_header"] is False
|
|
|
|
|
|
def test_loopback_without_auth_runs(uvicorn_spy: UvicornSpy) -> None:
|
|
assert run_http(server_settings(log_level="DEBUG")) == 0
|
|
|
|
assert uvicorn_spy.calls[0][1]["host"] == "127.0.0.1"
|
|
assert uvicorn_spy.calls[0][1]["log_level"] == "debug"
|
|
|
|
|
|
def test_forwarded_allow_ips_setting_is_passed_to_uvicorn(uvicorn_spy: UvicornSpy) -> None:
|
|
assert run_http(server_settings(forwarded_allow_ips="10.0.0.0/8, 172.18.0.2")) == 0
|
|
|
|
assert uvicorn_spy.calls[0][1]["forwarded_allow_ips"] == "10.0.0.0/8,172.18.0.2"
|
|
|
|
|
|
def test_forwarded_allow_ips_wildcard_needs_an_explicit_value_and_warns(
|
|
uvicorn_spy: UvicornSpy, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
assert server_settings().forwarded_allow_ips == "127.0.0.1"
|
|
|
|
assert run_http(server_settings(forwarded_allow_ips="*")) == 0
|
|
|
|
assert uvicorn_spy.calls[0][1]["forwarded_allow_ips"] == "*"
|
|
assert "FORWARDED_ALLOW_IPS=*" in capsys.readouterr().err
|
|
|
|
|
|
def test_non_loopback_bind_without_public_host_warns_which_variables_to_set(
|
|
uvicorn_spy: UvicornSpy, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
assert run_http(server_settings(mcp_host=ANY_HOST, mcp_bearer_token=BEARER_TOKEN)) == 0
|
|
|
|
err = capsys.readouterr().err
|
|
assert "PUBLIC_BASE_URL" in err and "ALLOWED_HOSTS" in err and "421" in err
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"extra",
|
|
[{"public_base_url": "https://mcp.example.com"}, {"allowed_hosts": ["mcp.example.com"]}],
|
|
)
|
|
def test_no_host_warning_when_a_public_host_is_configured(
|
|
uvicorn_spy: UvicornSpy, capsys: pytest.CaptureFixture[str], extra: dict[str, Any]
|
|
) -> None:
|
|
settings = server_settings(mcp_host=ANY_HOST, mcp_bearer_token=BEARER_TOKEN, **extra)
|
|
|
|
assert run_http(settings) == 0
|
|
|
|
assert "421" not in capsys.readouterr().err
|
|
|
|
|
|
def test_no_host_warning_on_loopback(
|
|
uvicorn_spy: UvicornSpy, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
assert run_http(server_settings()) == 0
|
|
|
|
assert capsys.readouterr().err == ""
|
|
|
|
|
|
def test_oauth_enabled_with_a_bearer_warns_that_oauth_is_unavailable(
|
|
uvicorn_spy: UvicornSpy, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
assert run_http(oauth_settings(mcp_bearer_token=BEARER_TOKEN)) == 0
|
|
|
|
assert "per-user OAuth is not available" in capsys.readouterr().err
|
|
|
|
|
|
@pytest.mark.parametrize("value", ["*,10.0.0.1", "not-an-ip", "10.0.0.1/99"])
|
|
def test_invalid_forwarded_allow_ips_is_rejected(value: str) -> None:
|
|
with pytest.raises(ValueError, match="FORWARDED_ALLOW_IPS"):
|
|
server_settings(forwarded_allow_ips=value)
|
|
|
|
|
|
def test_blank_forwarded_allow_ips_means_the_default() -> None:
|
|
assert server_settings(forwarded_allow_ips=" ").forwarded_allow_ips == "127.0.0.1"
|
|
|
|
|
|
# Audit log at startup
|
|
|
|
|
|
@pytest.fixture
|
|
def unwritable_audit(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
|
|
"""An audit path whose parent is a regular file, so it can never be opened."""
|
|
blocker = tmp_path / "blocker"
|
|
blocker.write_text("x", encoding="utf-8")
|
|
path = blocker / "audit.log"
|
|
monkeypatch.setenv("AUDIT_LOG_PATH", str(path))
|
|
return path
|
|
|
|
|
|
def test_http_exits_2_when_the_audit_log_cannot_be_opened(
|
|
uvicorn_spy: UvicornSpy, unwritable_audit: Path, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
assert run_http(server_settings(audit_log_path=unwritable_audit)) == 2
|
|
|
|
captured = capsys.readouterr()
|
|
assert uvicorn_spy.calls == []
|
|
assert "audit log" in captured.err and "AUDIT_LOG_PATH" in captured.err
|
|
assert str(unwritable_audit) in captured.err
|
|
assert captured.out == ""
|
|
|
|
|
|
def test_stdio_exits_2_when_the_audit_log_cannot_be_opened(
|
|
unwritable_audit: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
monkeypatch.setattr(transports.anyio, "run", lambda *a: pytest.fail("must not serve"))
|
|
|
|
assert run_stdio(server_settings(audit_log_path=unwritable_audit)) == 2
|
|
|
|
captured = capsys.readouterr()
|
|
assert "AUDIT_LOG_PATH" in captured.err
|
|
assert captured.out == ""
|
|
|
|
|
|
def test_oauth_misconfiguration_lists_every_missing_setting(
|
|
uvicorn_spy: UvicornSpy, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
settings = server_settings(oauth_enabled=True, mcp_bearer_token=BEARER_TOKEN)
|
|
|
|
assert run_http(settings) == 2
|
|
|
|
err = capsys.readouterr().err
|
|
for name in (
|
|
"GITEA_OAUTH_CLIENT_ID",
|
|
"GITEA_OAUTH_CLIENT_SECRET",
|
|
"TOKEN_ENCRYPTION_KEY",
|
|
"PUBLIC_BASE_URL",
|
|
):
|
|
assert name in err
|
|
assert uvicorn_spy.calls == []
|
|
|
|
|
|
def test_oauth_enabled_but_unavailable_does_not_start_without_a_bearer(
|
|
uvicorn_spy: UvicornSpy, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
assert run_http(oauth_settings()) == 2
|
|
|
|
assert "not available in this build" in capsys.readouterr().err
|
|
assert uvicorn_spy.calls == []
|
|
|
|
|
|
def test_oauth_enabled_with_a_bearer_falls_back_to_the_bearer(uvicorn_spy: UvicornSpy) -> None:
|
|
assert run_http(oauth_settings(mcp_bearer_token=BEARER_TOKEN)) == 0
|
|
assert len(uvicorn_spy.calls) == 1
|
|
|
|
|
|
def test_oauth_components_are_used_when_available(
|
|
uvicorn_spy: UvicornSpy, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
class Verifier:
|
|
async def verify_token(self, token: str) -> None:
|
|
return None
|
|
|
|
extras = HttpAuthExtras(token_verifier=Verifier())
|
|
monkeypatch.setattr(transports, "load_oauth_components", lambda settings: extras)
|
|
settings = oauth_settings(mcp_host=ANY_HOST, gitea_token=None)
|
|
|
|
assert run_http(settings) == 0
|
|
assert len(uvicorn_spy.calls) == 1
|
|
|
|
|
|
def test_default_oauth_components_hook_returns_none() -> None:
|
|
assert load_oauth_components(oauth_settings()) is None
|
|
|
|
|
|
def test_bearer_without_gitea_token_is_refused(
|
|
uvicorn_spy: UvicornSpy, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
assert run_http(server_settings(mcp_bearer_token=BEARER_TOKEN, gitea_token=None)) == 2
|
|
|
|
assert "GITEA_TOKEN" in capsys.readouterr().err
|
|
|
|
|
|
def test_http_startup_errors_is_empty_for_a_valid_setup() -> None:
|
|
assert http_startup_errors(server_settings(mcp_bearer_token=BEARER_TOKEN), None) == []
|
|
|
|
|
|
def test_bad_disabled_tools_stop_http_startup(
|
|
uvicorn_spy: UvicornSpy, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
assert run_http(server_settings(disabled_tools=["gitea_nope"])) == 2
|
|
|
|
assert "gitea_nope" in capsys.readouterr().err
|
|
assert uvicorn_spy.calls == []
|
|
|
|
|
|
# stdio
|
|
|
|
|
|
def test_stdio_without_a_token_exits_2(capsys: pytest.CaptureFixture[str]) -> None:
|
|
assert run_stdio(server_settings(gitea_token=None)) == 2
|
|
|
|
captured = capsys.readouterr()
|
|
assert "GITEA_TOKEN" in captured.err
|
|
assert captured.out == ""
|
|
|
|
|
|
def test_stdio_with_bad_disabled_tools_exits_2(capsys: pytest.CaptureFixture[str]) -> None:
|
|
assert run_stdio(server_settings(disabled_tools=["nope"])) == 2
|
|
|
|
assert "nope" in capsys.readouterr().err
|
|
|
|
|
|
def test_stdio_serves_the_built_server(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
served: list[Any] = []
|
|
|
|
def fake_run(function: Any, server: Any) -> None:
|
|
served.append((function, server))
|
|
|
|
monkeypatch.setattr(transports.anyio, "run", fake_run)
|
|
|
|
assert run_stdio(server_settings()) == 0
|
|
|
|
function, server = served[0]
|
|
assert function is transports._serve_stdio
|
|
assert server.tool_count == 538
|
|
|
|
|
|
async def test_real_stdio_subprocess_serves_the_catalog(tmp_path: Path) -> None:
|
|
env = {
|
|
"GITEA_URL": "http://127.0.0.1:9",
|
|
"GITEA_TOKEN": GITEA_TOKEN,
|
|
"DATA_DIR": str(tmp_path / "state"),
|
|
"AUDIT_LOG_PATH": str(tmp_path / "state" / "audit.log"),
|
|
"REQUEST_TIMEOUT_SECONDS": "2",
|
|
}
|
|
params = StdioServerParameters(
|
|
command=sys.executable,
|
|
args=["-m", "aegis_gitea_mcp.cli", "serve", "--transport", "stdio"],
|
|
env=env,
|
|
cwd=str(tmp_path),
|
|
)
|
|
async with Client(stdio_client(params)) as client:
|
|
tools = (await client.list_tools()).tools
|
|
result = await client.call_tool("gitea_repo_get", {"owner": "o", "repo": "r"})
|
|
|
|
assert len(tools) == 538
|
|
# Gitea is unreachable: the call fails as an error result, stdout stayed valid JSON-RPC.
|
|
assert result.is_error
|
|
payload = json.loads(result.content[0].text) # type: ignore[union-attr]
|
|
assert "error" in payload and GITEA_TOKEN not in json.dumps(payload)
|
|
audit = (tmp_path / "state" / "audit.log").read_text(encoding="utf-8").splitlines()
|
|
assert len(audit) == 1
|
|
assert GITEA_TOKEN not in audit[0]
|
|
|
|
|
|
# serve command
|
|
|
|
|
|
def test_serve_command_applies_flags_and_runs_http(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
# Set first so monkeypatch restores whatever ``serve`` writes into the environment.
|
|
monkeypatch.setenv("MCP_TRANSPORT", "stdio")
|
|
monkeypatch.setenv("MCP_HOST", "127.0.0.1")
|
|
monkeypatch.setenv("MCP_PORT", "8080")
|
|
seen: list[Settings] = []
|
|
monkeypatch.setattr(transports, "run_http", lambda settings: seen.append(settings) or 0)
|
|
monkeypatch.setattr(transports, "run_stdio", lambda settings: pytest.fail("wrong transport"))
|
|
|
|
code = main(["serve", "--transport", "http", "--host", "127.0.0.2", "--port", "9123"])
|
|
|
|
assert code == 0
|
|
assert (seen[0].mcp_transport, seen[0].mcp_host, seen[0].mcp_port) == (
|
|
"http",
|
|
"127.0.0.2",
|
|
9123,
|
|
)
|
|
|
|
|
|
def test_serve_defaults_to_the_configured_transport(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("MCP_TRANSPORT", "http")
|
|
seen: list[str] = []
|
|
monkeypatch.setattr(transports, "run_http", lambda settings: seen.append("http") or 0)
|
|
monkeypatch.setattr(transports, "run_stdio", lambda settings: seen.append("stdio") or 0)
|
|
|
|
assert main([]) == 0
|
|
assert seen == ["http"]
|
|
|
|
|
|
def test_serve_defaults_to_stdio(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
seen: list[str] = []
|
|
monkeypatch.setattr(transports, "run_stdio", lambda settings: seen.append("stdio") or 0)
|
|
|
|
assert main(["serve"]) == 0
|
|
assert seen == ["stdio"]
|
|
|
|
|
|
def test_serve_with_invalid_configuration_never_echoes_values(
|
|
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
monkeypatch.setenv("MCP_BEARER_TOKEN", "too-short-secret-VALUE")
|
|
monkeypatch.setattr(transports, "run_stdio", lambda settings: pytest.fail("must not run"))
|
|
|
|
assert main(["serve"]) == 2
|
|
|
|
captured = capsys.readouterr()
|
|
assert "MCP_BEARER_TOKEN" in captured.err
|
|
assert "too-short-secret-VALUE" not in captured.err + captured.out
|
|
|
|
|
|
def test_serve_without_gitea_url_exits_2(
|
|
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
monkeypatch.delenv("GITEA_URL")
|
|
|
|
assert main(["serve"]) == 2
|
|
|
|
assert "GITEA_URL" in capsys.readouterr().err
|
|
|
|
|
|
def test_serve_rejects_an_out_of_range_port() -> None:
|
|
with pytest.raises(SystemExit):
|
|
cli.build_parser().parse_args(["serve", "--port", "70000"])
|
|
|
|
|
|
async def test_serve_stdio_runs_the_server_over_the_stdio_streams(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""``_serve_stdio`` hands the stdio streams to the server (streams replaced by memory)."""
|
|
from contextlib import asynccontextmanager
|
|
|
|
import anyio
|
|
from mcp.client.session import ClientSession
|
|
from mcp.shared.memory import create_client_server_memory_streams
|
|
|
|
from aegis_gitea_mcp.server import build_server
|
|
|
|
async with create_client_server_memory_streams() as (client_streams, server_streams):
|
|
|
|
@asynccontextmanager
|
|
async def fake_stdio() -> Any:
|
|
yield server_streams
|
|
|
|
monkeypatch.setattr(transports, "stdio_server", fake_stdio)
|
|
server = build_server(server_settings())
|
|
async with anyio.create_task_group() as group:
|
|
group.start_soon(transports._serve_stdio, server)
|
|
async with ClientSession(*client_streams) as session:
|
|
await session.initialize()
|
|
tools = await session.list_tools()
|
|
group.cancel_scope.cancel()
|
|
|
|
assert len(tools.tools) == 538
|