feat: add MCP server core, stdio and streamable HTTP transports and serve command

Low-level SDK server built from the catalog plus gitea_api_request, with TOOLSETS/READ_ONLY/DISABLED_TOOLS filters, schema validation without value echo, one audit event per call, static bearer verifier, identity resolver hook, /health, DNS-rebinding protection, rate limiting and fail-closed startup checks.
This commit is contained in:
Latte committed 2026-10-14 16:13:11 +02:00
1 parent 581ca67bf8
commit c9e2c481e7
11 files changed
+2846 -14

No files matched your search

+59
View File
@@ -0,0 +1,59 @@
"""Shared builders for the server, auth and transport tests."""
from __future__ import annotations
import json
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import Any
from mcp import Client
from pytest_httpx import HTTPXMock
from aegis_gitea_mcp.audit import get_audit_logger
from aegis_gitea_mcp.config import Settings
from aegis_gitea_mcp.server import AegisServer, build_server
from tests.test_executor_helpers import API, BEARER_TOKEN, GITEA_TOKEN
ANY_HOST = "0.0.0.0" # noqa: S104 (a wildcard bind address used as test input)
__all__ = [
"ANY_HOST",
"API",
"BEARER_TOKEN",
"GITEA_TOKEN",
"audit_entries",
"connected",
"mock_user",
"server_settings",
]
def server_settings(**overrides: Any) -> Settings:
"""Settings for server tests: admin token, no bearer, no OAuth secrets."""
values: dict[str, Any] = {
"gitea_url": "https://gitea.example.com",
"gitea_token": GITEA_TOKEN,
}
values.update(overrides)
return Settings(**values)
def mock_user(httpx_mock: HTTPXMock, login: str = "admin") -> None:
"""Answer the admin login lookup (``GET /user``) for as many calls as needed."""
httpx_mock.add_response(url=f"{API}/user", json={"login": login}, is_reusable=True)
@asynccontextmanager
async def connected(settings: Settings | None = None, **kwargs: Any) -> AsyncIterator[Client]:
"""An in-memory SDK client connected to a freshly built server."""
server: AegisServer = build_server(settings or server_settings(), **kwargs)
async with Client(server) as client:
yield client
def audit_entries() -> list[dict[str, Any]]:
"""Payloads of every tool invocation written to the (test) audit log."""
path = get_audit_logger().log_path
entries = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()]
return [entry["payload"] for entry in entries if entry["event_type"] == "tool_invocation"]