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.
730 lines
27 KiB
Python
730 lines
27 KiB
Python
"""Tests for the MCP server core: tool list, filters, dispatcher, audit."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from mcp import Client, MCPError
|
|
from mcp.server.auth.provider import AccessToken
|
|
from pytest_httpx import HTTPXMock
|
|
|
|
from aegis_gitea_mcp.audit import validate_audit_log_integrity
|
|
from aegis_gitea_mcp.catalog import load_catalog
|
|
from aegis_gitea_mcp.errors import ToolError
|
|
from aegis_gitea_mcp.gitea import RAW_TOOL_NAME, UNTRUSTED_NOTICE, GiteaExecutor, ToolOutcome
|
|
from aegis_gitea_mcp.identity import Identity
|
|
from aegis_gitea_mcp.server import (
|
|
RAW_INPUT_SCHEMA,
|
|
ServerConfigError,
|
|
ToolRegistry,
|
|
build_raw_tool,
|
|
build_server,
|
|
max_request_body_size,
|
|
validate_arguments,
|
|
)
|
|
from tests.test_server_helpers import (
|
|
API,
|
|
GITEA_TOKEN,
|
|
audit_entries,
|
|
connected,
|
|
mock_user,
|
|
server_settings,
|
|
)
|
|
|
|
CATALOG = load_catalog()
|
|
REPO_ARGS = {"owner": "octo", "repo": "demo"}
|
|
|
|
|
|
def payload_of(result: Any) -> dict[str, Any]:
|
|
"""Decode the JSON text content of a tool result."""
|
|
assert len(result.content) == 1
|
|
decoded: dict[str, Any] = json.loads(result.content[0].text)
|
|
return decoded
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _env(mock_env: None) -> None:
|
|
"""Every test here needs GITEA_URL for the audit logger's settings."""
|
|
|
|
|
|
# Tool list
|
|
|
|
|
|
async def test_list_tools_has_every_operation_plus_raw_tool() -> None:
|
|
async with connected() as client:
|
|
tools = (await client.list_tools()).tools
|
|
|
|
names = [tool.name for tool in tools]
|
|
assert len(tools) == len(CATALOG.operations) + 1 == 538
|
|
assert len(set(names)) == len(names)
|
|
assert names == sorted(names)
|
|
assert RAW_TOOL_NAME in names
|
|
for tool in tools:
|
|
assert tool.input_schema["type"] == "object"
|
|
assert tool.annotations is not None
|
|
assert tool.annotations.open_world_hint is False
|
|
|
|
|
|
async def test_list_tools_annotations_follow_the_catalog() -> None:
|
|
async with connected() as client:
|
|
tools = {tool.name: tool for tool in (await client.list_tools()).tools}
|
|
|
|
get_repo = tools["gitea_repo_get"].annotations
|
|
delete_repo = tools["gitea_repo_delete"].annotations
|
|
create_issue = tools["gitea_issue_create_issue"].annotations
|
|
assert get_repo is not None and delete_repo is not None and create_issue is not None
|
|
assert (get_repo.read_only_hint, get_repo.destructive_hint, get_repo.idempotent_hint) == (
|
|
True,
|
|
False,
|
|
True,
|
|
)
|
|
assert (delete_repo.read_only_hint, delete_repo.destructive_hint) == (False, True)
|
|
assert (create_issue.destructive_hint, create_issue.idempotent_hint) == (False, False)
|
|
assert tools["gitea_repo_get"].title == get_repo.title
|
|
assert tools["gitea_repo_get"].description
|
|
|
|
|
|
async def test_raw_tool_definition() -> None:
|
|
async with connected() as client:
|
|
raw = next(t for t in (await client.list_tools()).tools if t.name == RAW_TOOL_NAME)
|
|
|
|
assert raw.input_schema == RAW_INPUT_SCHEMA
|
|
assert raw.input_schema["additionalProperties"] is False
|
|
assert raw.input_schema["properties"]["method"]["enum"] == [
|
|
"GET",
|
|
"POST",
|
|
"PUT",
|
|
"PATCH",
|
|
"DELETE",
|
|
]
|
|
assert raw.input_schema["properties"]["path"]["maxLength"] == 2048
|
|
assert raw.annotations is not None
|
|
assert (raw.annotations.read_only_hint, raw.annotations.destructive_hint) == (False, True)
|
|
|
|
|
|
async def test_list_tools_ignores_cursor() -> None:
|
|
async with connected() as client:
|
|
result = await client.list_tools(cursor="whatever")
|
|
|
|
assert len(result.tools) == 538
|
|
assert result.next_cursor is None
|
|
|
|
|
|
# Filters
|
|
|
|
|
|
async def test_toolsets_limit_listing_and_calling(httpx_mock: HTTPXMock) -> None:
|
|
settings = server_settings(toolsets=["issue"])
|
|
async with connected(settings) as client:
|
|
tools = (await client.list_tools()).tools
|
|
assert {CATALOG.get(t.name).domain for t in tools if CATALOG.get(t.name)} == {"issue"}
|
|
assert RAW_TOOL_NAME not in {t.name for t in tools}
|
|
with pytest.raises(MCPError):
|
|
await client.call_tool("gitea_repo_get", REPO_ARGS)
|
|
|
|
assert httpx_mock.get_requests() == []
|
|
|
|
|
|
async def test_misc_toolset_includes_the_raw_tool() -> None:
|
|
async with connected(server_settings(toolsets=["misc"])) as client:
|
|
names = {t.name for t in (await client.list_tools()).tools}
|
|
|
|
assert RAW_TOOL_NAME in names
|
|
assert {CATALOG.get(n).domain for n in names - {RAW_TOOL_NAME}} == {"misc"} # type: ignore[union-attr]
|
|
|
|
|
|
async def test_read_only_lists_only_read_operations_and_keeps_the_raw_tool() -> None:
|
|
async with connected(server_settings(read_only=True)) as client:
|
|
tools = {t.name: t for t in (await client.list_tools()).tools}
|
|
|
|
assert RAW_TOOL_NAME in tools
|
|
read_ops = [op for op in CATALOG.operations if op.read_only]
|
|
assert len(tools) == len(read_ops) + 1
|
|
assert "gitea_repo_delete" not in tools
|
|
raw = tools[RAW_TOOL_NAME]
|
|
assert raw.annotations is not None and raw.annotations.read_only_hint is True
|
|
assert raw.description is not None and "only GET" in raw.description
|
|
|
|
|
|
async def test_read_only_rejects_non_get_raw_requests(httpx_mock: HTTPXMock) -> None:
|
|
async with connected(server_settings(read_only=True)) as client:
|
|
blocked = await client.call_tool(
|
|
RAW_TOOL_NAME, {"method": "DELETE", "path": "/repos/octo/demo"}
|
|
)
|
|
|
|
assert blocked.is_error
|
|
assert payload_of(blocked)["error"]["kind"] == "read_only"
|
|
assert httpx_mock.get_requests() == []
|
|
assert audit_entries()[-1]["outcome"] == "read_only"
|
|
|
|
|
|
async def test_read_only_allows_get_raw_requests(httpx_mock: HTTPXMock) -> None:
|
|
mock_user(httpx_mock)
|
|
httpx_mock.add_response(url=f"{API}/version", json={"version": "1.22"})
|
|
async with connected(server_settings(read_only=True)) as client:
|
|
result = await client.call_tool(RAW_TOOL_NAME, {"method": "GET", "path": "/version"})
|
|
|
|
assert not result.is_error
|
|
assert payload_of(result)["data"] == {"version": "1.22"}
|
|
|
|
|
|
async def test_disabled_tools_are_hidden_and_unknown_when_called() -> None:
|
|
settings = server_settings(disabled_tools=["gitea_repo_delete", RAW_TOOL_NAME])
|
|
async with connected(settings) as client:
|
|
names = {t.name for t in (await client.list_tools()).tools}
|
|
assert len(names) == 536
|
|
assert "gitea_repo_delete" not in names
|
|
assert RAW_TOOL_NAME not in names
|
|
with pytest.raises(MCPError) as caught:
|
|
await client.call_tool("gitea_repo_delete", REPO_ARGS)
|
|
|
|
assert caught.value.code == -32602
|
|
|
|
|
|
def test_unknown_disabled_tool_names_are_a_startup_error() -> None:
|
|
settings = server_settings(disabled_tools=["gitea_repo_get", "gitea_nope", "other_nope"])
|
|
|
|
with pytest.raises(ServerConfigError) as caught:
|
|
build_server(settings)
|
|
|
|
message = str(caught.value)
|
|
assert "gitea_nope" in message and "other_nope" in message
|
|
assert "gitea_repo_get" not in message
|
|
|
|
|
|
def test_disabled_tool_hidden_by_read_only_is_still_a_valid_name() -> None:
|
|
settings = server_settings(read_only=True, disabled_tools=["gitea_repo_delete"])
|
|
|
|
registry = ToolRegistry(settings, CATALOG)
|
|
|
|
assert registry.get("gitea_repo_delete") is None
|
|
assert len(registry) == sum(op.read_only for op in CATALOG.operations) + 1
|
|
|
|
|
|
def test_registry_caches_validators() -> None:
|
|
registry = ToolRegistry(server_settings(), CATALOG)
|
|
|
|
assert registry.validator("gitea_repo_get") is registry.validator("gitea_repo_get")
|
|
|
|
|
|
def test_build_raw_tool_read_only_flag() -> None:
|
|
assert build_raw_tool().annotations is not None
|
|
assert build_raw_tool(read_only=True).annotations.read_only_hint is True # type: ignore[union-attr]
|
|
|
|
|
|
# Calls: success and Gitea errors
|
|
|
|
|
|
async def test_call_success_envelope(httpx_mock: HTTPXMock) -> None:
|
|
mock_user(httpx_mock, "admin")
|
|
httpx_mock.add_response(url=f"{API}/repos/octo/demo", json={"full_name": "octo/demo"})
|
|
async with connected() as client:
|
|
result = await client.call_tool("gitea_repo_get", REPO_ARGS)
|
|
|
|
assert result.is_error is False
|
|
payload = payload_of(result)
|
|
assert payload["identity"] == {"login": "admin", "mode": "admin-token", "sudo": None}
|
|
assert payload["untrusted_content"] == UNTRUSTED_NOTICE
|
|
assert payload["data"] == {"full_name": "octo/demo"}
|
|
assert result.structured_content == payload
|
|
request = httpx_mock.get_requests(url=f"{API}/repos/octo/demo")[0]
|
|
assert request.headers["Authorization"] == f"token {GITEA_TOKEN}"
|
|
|
|
|
|
async def test_gitea_404_is_an_error_result_with_kind(httpx_mock: HTTPXMock) -> None:
|
|
mock_user(httpx_mock)
|
|
httpx_mock.add_response(url=f"{API}/repos/octo/demo", status_code=404, json={"message": "nope"})
|
|
async with connected() as client:
|
|
result = await client.call_tool("gitea_repo_get", REPO_ARGS)
|
|
|
|
assert result.is_error is True
|
|
error = payload_of(result)["error"]
|
|
assert error["kind"] == "not_found" and error["status"] == 404
|
|
entry = audit_entries()[-1]
|
|
assert entry["outcome"] == "not_found" and entry["result_status"] == "error"
|
|
assert entry["params"]["http_status"] == 404
|
|
|
|
|
|
async def test_none_arguments_are_treated_as_empty(httpx_mock: HTTPXMock) -> None:
|
|
mock_user(httpx_mock)
|
|
httpx_mock.add_response(url=f"{API}/user", json={"login": "admin"})
|
|
async with connected() as client:
|
|
result = await client.call_tool("gitea_user_get_current")
|
|
|
|
assert not result.is_error
|
|
assert payload_of(result)["data"] == {"login": "admin"}
|
|
|
|
|
|
# Calls: validation
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("tool", "arguments", "path", "needle"),
|
|
[
|
|
("gitea_repo_get", {"owner": "octo"}, "$", "repo"),
|
|
("gitea_repo_get", {"owner": 5, "repo": "demo"}, "$.owner", "type"),
|
|
("gitea_repo_get", {**REPO_ARGS, "surprise": 1}, "$", "unexpected argument"),
|
|
("gitea_repo_get", {"owner": "no spaces!", "repo": "demo"}, "$.owner", "pattern"),
|
|
("gitea_repo_get", {"owner": "o" * 300, "repo": "demo"}, "$.owner", "maxLength"),
|
|
("gitea_repo_get", {"owner": "", "repo": "demo"}, "$.owner", "minLength"),
|
|
(RAW_TOOL_NAME, {"method": "TRACE", "path": "/x"}, "$.method", "one of"),
|
|
(RAW_TOOL_NAME, {"method": "GET", "path": "no-slash"}, "$.path", "pattern"),
|
|
(RAW_TOOL_NAME, {"method": "GET", "path": "/x", "query": {"a": {"b": 1}}}, "$.query.a", ""),
|
|
],
|
|
)
|
|
async def test_validation_errors_do_not_call_gitea(
|
|
httpx_mock: HTTPXMock, tool: str, arguments: dict[str, Any], path: str, needle: str
|
|
) -> None:
|
|
async with connected() as client:
|
|
result = await client.call_tool(tool, arguments)
|
|
|
|
assert result.is_error
|
|
error = payload_of(result)["error"]
|
|
assert error["kind"] == "invalid_argument"
|
|
assert error["path"] == path
|
|
assert needle in error["message"]
|
|
assert httpx_mock.get_requests() == []
|
|
assert audit_entries()[-1]["outcome"] == "invalid_argument"
|
|
|
|
|
|
async def test_validation_error_never_echoes_the_offending_value() -> None:
|
|
secret = "sk-live-SUPERSECRET-0123456789"
|
|
async with connected() as client:
|
|
results = [
|
|
await client.call_tool("gitea_repo_get", {"owner": secret, "repo": "demo"}),
|
|
await client.call_tool("gitea_repo_get", {"owner": 1, "repo": "demo", secret: secret}),
|
|
await client.call_tool("gitea_repo_get", {"owner": ["a", secret], "repo": "demo"}),
|
|
]
|
|
|
|
for result in results:
|
|
assert result.is_error
|
|
assert secret not in result.content[0].text # type: ignore[union-attr]
|
|
assert "unexpected argument" in payload_of(results[1])["error"]["message"]
|
|
|
|
|
|
def test_validate_arguments_returns_none_for_valid_input() -> None:
|
|
registry = ToolRegistry(server_settings(), CATALOG)
|
|
|
|
assert validate_arguments(registry.validator("gitea_repo_get"), REPO_ARGS) is None
|
|
|
|
|
|
def test_validate_arguments_covers_remaining_keyword_messages() -> None:
|
|
from jsonschema import Draft202012Validator
|
|
|
|
validator = Draft202012Validator(
|
|
{
|
|
"type": "object",
|
|
"properties": {
|
|
"n": {"type": "integer", "minimum": 3},
|
|
"t": {"anyOf": [{"type": "string"}, {"type": "integer"}]},
|
|
"k": {"type": ["string", "null"]},
|
|
},
|
|
}
|
|
)
|
|
|
|
low = validate_arguments(validator, {"n": 1})
|
|
anyof = validate_arguments(validator, {"t": [1]})
|
|
multi = validate_arguments(validator, {"k": 5})
|
|
assert low is not None and "minimum 3" in low["message"]
|
|
assert anyof is not None and "'anyOf'" in anyof["message"] and anyof["path"] == "$.t"
|
|
assert multi is not None and '["string", "null"]' in multi["message"]
|
|
|
|
|
|
def test_validate_arguments_array_index_paths() -> None:
|
|
from jsonschema import Draft202012Validator
|
|
|
|
validator = Draft202012Validator(
|
|
{"type": "object", "properties": {"ids": {"type": "array", "items": {"type": "integer"}}}}
|
|
)
|
|
|
|
error = validate_arguments(validator, {"ids": [1, "x"]})
|
|
|
|
assert error is not None and error["path"] == "$.ids[1]"
|
|
|
|
|
|
# Calls: unknown tools
|
|
|
|
|
|
async def test_unknown_tool_is_a_protocol_error_and_is_audited() -> None:
|
|
async with connected() as client:
|
|
with pytest.raises(MCPError) as caught:
|
|
await client.call_tool("gitea_does_not_exist", {"a": 1})
|
|
|
|
assert caught.value.code == -32602
|
|
assert "gitea_does_not_exist" in caught.value.message
|
|
entry = audit_entries()[-1]
|
|
assert entry["tool_name"] == "gitea_does_not_exist"
|
|
assert entry["outcome"] == "unknown_tool"
|
|
|
|
|
|
# Calls: raw tool
|
|
|
|
|
|
async def test_raw_tool_happy_path(httpx_mock: HTTPXMock) -> None:
|
|
mock_user(httpx_mock)
|
|
httpx_mock.add_response(
|
|
url=f"{API}/repos/octo/demo/issues?state=open&labels=a&labels=b",
|
|
method="POST",
|
|
status_code=201,
|
|
json={"number": 7},
|
|
)
|
|
async with connected() as client:
|
|
result = await client.call_tool(
|
|
RAW_TOOL_NAME,
|
|
{
|
|
"method": "post",
|
|
"path": "/repos/octo/demo/issues",
|
|
"query": {"state": "open", "labels": ["a", "b"]},
|
|
"body": {"title": "hello"},
|
|
"sudo": "alice",
|
|
},
|
|
)
|
|
# lower-case verbs are not in the enum
|
|
assert result.is_error
|
|
assert payload_of(result)["error"]["kind"] == "invalid_argument"
|
|
|
|
async with connected() as client:
|
|
result = await client.call_tool(
|
|
RAW_TOOL_NAME,
|
|
{
|
|
"method": "POST",
|
|
"path": "/repos/octo/demo/issues",
|
|
"query": {"state": "open", "labels": ["a", "b"]},
|
|
"body": {"title": "hello"},
|
|
"sudo": "alice",
|
|
},
|
|
)
|
|
|
|
assert not result.is_error
|
|
payload = payload_of(result)
|
|
assert payload["status"] == 201 and payload["data"] == {"number": 7}
|
|
assert payload["identity"]["sudo"] == "alice"
|
|
sent = httpx_mock.get_requests(method="POST")[0]
|
|
assert json.loads(sent.content) == {"title": "hello"}
|
|
assert sent.headers["Sudo"] == "alice"
|
|
entry = audit_entries()[-1]
|
|
assert entry["tool_name"] == RAW_TOOL_NAME
|
|
assert entry["method"] == "POST" and entry["path"] == "/repos/octo/demo/issues"
|
|
assert entry["sudo"] == "alice" and entry["domain"] == "misc"
|
|
|
|
|
|
async def test_raw_tool_without_body_sends_none(httpx_mock: HTTPXMock) -> None:
|
|
mock_user(httpx_mock)
|
|
httpx_mock.add_response(url=f"{API}/version", json={"version": "1"})
|
|
async with connected() as client:
|
|
result = await client.call_tool(RAW_TOOL_NAME, {"method": "GET", "path": "/version"})
|
|
|
|
assert not result.is_error
|
|
assert httpx_mock.get_requests(url=f"{API}/version")[0].content == b""
|
|
|
|
|
|
# Audit
|
|
|
|
|
|
async def test_audit_event_per_call_never_contains_argument_values(
|
|
httpx_mock: HTTPXMock,
|
|
) -> None:
|
|
mock_user(httpx_mock, "admin")
|
|
httpx_mock.add_response(
|
|
url=f"{API}/repos/octo/demo/issues", method="POST", status_code=201, json={"number": 1}
|
|
)
|
|
httpx_mock.add_response(url=f"{API}/repos/octo/demo", status_code=404, json={"message": "x"})
|
|
secret = "BODY-SECRET-VALUE-9f8e7d6c5b4a"
|
|
async with connected() as client:
|
|
await client.call_tool(
|
|
"gitea_issue_create_issue",
|
|
{**REPO_ARGS, "body": {"title": secret, "body": secret}, "sudo": "alice"},
|
|
)
|
|
await client.call_tool("gitea_repo_get", REPO_ARGS)
|
|
await client.call_tool("gitea_repo_get", {"owner": secret + "!", "repo": "x"})
|
|
with pytest.raises(MCPError):
|
|
await client.call_tool("gitea_unknown", {"value": secret})
|
|
|
|
entries = audit_entries()
|
|
assert [e["outcome"] for e in entries] == [
|
|
"success",
|
|
"not_found",
|
|
"invalid_argument",
|
|
"unknown_tool",
|
|
]
|
|
first = entries[0]
|
|
assert first["tool_name"] == "gitea_issue_create_issue"
|
|
assert first["domain"] == "issue" and first["method"] == "POST"
|
|
assert first["path"] == "/repos/{owner}/{repo}/issues"
|
|
assert first["identity_login"] == "admin" and first["identity_mode"] == "admin-token"
|
|
assert first["sudo"] == "alice" and first["result_status"] == "success"
|
|
assert first["params"] == {
|
|
"argument_names": ["body", "owner", "repo", "sudo"],
|
|
"http_status": 201,
|
|
}
|
|
assert first["duration_ms"] >= 0 and first["correlation_id"]
|
|
assert len({e["correlation_id"] for e in entries}) == 4
|
|
from aegis_gitea_mcp.audit import get_audit_logger
|
|
|
|
log_path = get_audit_logger().log_path
|
|
assert secret not in log_path.read_text(encoding="utf-8")
|
|
assert validate_audit_log_integrity(log_path) == (True, [])
|
|
|
|
|
|
async def test_audit_write_failure_after_execution_does_not_break_the_call(
|
|
httpx_mock: HTTPXMock, caplog: pytest.LogCaptureFixture
|
|
) -> None:
|
|
mock_user(httpx_mock)
|
|
httpx_mock.add_response(url=f"{API}/repos/octo/demo", json={})
|
|
|
|
class BrokenAudit:
|
|
def log_tool_invocation(self, *args: Any, **kwargs: Any) -> str:
|
|
raise OSError("disk full: " + GITEA_TOKEN)
|
|
|
|
async with connected(audit=BrokenAudit()) as client: # type: ignore[arg-type]
|
|
result = await client.call_tool("gitea_repo_get", REPO_ARGS)
|
|
|
|
# The call already ran (its side effect cannot be undone), so the result still goes out.
|
|
assert not result.is_error
|
|
assert "OSError" in caplog.text
|
|
assert GITEA_TOKEN not in caplog.text
|
|
|
|
|
|
async def test_unavailable_audit_log_refuses_the_call_before_gitea_is_contacted(
|
|
httpx_mock: HTTPXMock, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
|
|
) -> None:
|
|
def broken() -> None:
|
|
raise OSError("permission denied: " + GITEA_TOKEN)
|
|
|
|
monkeypatch.setattr("aegis_gitea_mcp.server.get_audit_logger", broken)
|
|
async with connected() as client:
|
|
result = await client.call_tool("gitea_repo_get", REPO_ARGS)
|
|
written = await client.call_tool("gitea_issue_create_issue", {**REPO_ARGS, "body": {}})
|
|
|
|
assert result.is_error and written.is_error
|
|
error = payload_of(result)["error"]
|
|
assert error["kind"] == "audit_unavailable"
|
|
assert GITEA_TOKEN not in json.dumps(payload_of(result)) + caplog.text
|
|
assert httpx_mock.get_requests() == []
|
|
|
|
|
|
async def test_audit_correlation_id_is_the_request_id(httpx_mock: HTTPXMock) -> None:
|
|
mock_user(httpx_mock)
|
|
httpx_mock.add_response(url=f"{API}/repos/octo/demo", json={})
|
|
async with connected() as client:
|
|
await client.call_tool("gitea_repo_get", REPO_ARGS)
|
|
|
|
from aegis_gitea_mcp.audit import get_audit_logger
|
|
|
|
line = get_audit_logger().log_path.read_text(encoding="utf-8").splitlines()[-1]
|
|
entry = json.loads(line)
|
|
assert entry["payload"]["correlation_id"] == entry["request_id"] != "-"
|
|
|
|
|
|
async def test_rejected_raw_call_audits_no_unvalidated_path_or_sudo() -> None:
|
|
secret = "SECRET-IN-PATH-9f8e7d6c5b"
|
|
async with connected() as client:
|
|
bad_path = await client.call_tool(
|
|
"gitea_api_request", {"method": "GET", "path": secret, "sudo": secret}
|
|
)
|
|
bad_sudo = await client.call_tool(
|
|
"gitea_api_request", {"method": "GET", "path": "/user", "sudo": "bad name!"}
|
|
)
|
|
|
|
assert bad_path.is_error and bad_sudo.is_error
|
|
first, second = audit_entries()
|
|
assert first["outcome"] == second["outcome"] == "invalid_argument"
|
|
for entry in (first, second):
|
|
assert "path" not in entry and "sudo" not in entry
|
|
from aegis_gitea_mcp.audit import get_audit_logger
|
|
|
|
assert secret not in get_audit_logger().log_path.read_text(encoding="utf-8")
|
|
|
|
|
|
async def test_raw_call_audits_the_validated_method_and_path(httpx_mock: HTTPXMock) -> None:
|
|
mock_user(httpx_mock)
|
|
httpx_mock.add_response(url=f"{API}/version", json={})
|
|
async with connected() as client:
|
|
await client.call_tool(
|
|
"gitea_api_request", {"method": "GET", "path": "/version", "sudo": "alice"}
|
|
)
|
|
|
|
entry = audit_entries()[-1]
|
|
assert (entry["method"], entry["path"], entry["sudo"]) == ("GET", "/version", "alice")
|
|
|
|
|
|
# Internal errors
|
|
|
|
|
|
class ExplodingExecutor(GiteaExecutor):
|
|
"""Executor whose calls raise an exception carrying a secret."""
|
|
|
|
async def execute(self, *args: Any, **kwargs: Any) -> ToolOutcome:
|
|
raise RuntimeError(f"boom with {GITEA_TOKEN} and a stack")
|
|
|
|
|
|
async def test_internal_error_is_generic_and_audited(
|
|
httpx_mock: HTTPXMock, caplog: pytest.LogCaptureFixture
|
|
) -> None:
|
|
mock_user(httpx_mock)
|
|
settings = server_settings()
|
|
async with connected(settings, executor=ExplodingExecutor(settings)) as client:
|
|
result = await client.call_tool("gitea_repo_get", REPO_ARGS)
|
|
|
|
assert result.is_error
|
|
error = payload_of(result)["error"]
|
|
assert error["kind"] == "internal_error"
|
|
assert "boom" not in result.content[0].text # type: ignore[union-attr]
|
|
assert "RuntimeError" not in error["message"]
|
|
assert GITEA_TOKEN not in caplog.text and "Traceback" not in caplog.text
|
|
assert audit_entries()[-1]["outcome"] == "internal_error"
|
|
|
|
|
|
async def test_internal_error_shows_class_name_when_details_are_exposed(
|
|
httpx_mock: HTTPXMock,
|
|
) -> None:
|
|
mock_user(httpx_mock)
|
|
settings = server_settings(expose_error_details=True)
|
|
async with connected(settings, executor=ExplodingExecutor(settings)) as client:
|
|
result = await client.call_tool("gitea_repo_get", REPO_ARGS)
|
|
|
|
message = payload_of(result)["error"]["message"]
|
|
assert "RuntimeError" in message and "boom" not in message
|
|
|
|
|
|
# Identity
|
|
|
|
|
|
class FixedResolver:
|
|
"""Resolver returning a per-user identity (what the OAuth mode will plug in)."""
|
|
|
|
def __init__(self) -> None:
|
|
self.seen: list[AccessToken | None] = []
|
|
|
|
async def resolve(self, access: AccessToken | None) -> Identity:
|
|
self.seen.append(access)
|
|
return Identity(login="alice", mode="oauth", token="user-token-abcdefgh")
|
|
|
|
|
|
async def test_custom_resolver_identity_reaches_gitea_and_audit(httpx_mock: HTTPXMock) -> None:
|
|
httpx_mock.add_response(url=f"{API}/repos/octo/demo", json={})
|
|
resolver = FixedResolver()
|
|
async with connected(resolver=resolver) as client:
|
|
result = await client.call_tool("gitea_repo_get", REPO_ARGS)
|
|
|
|
assert payload_of(result)["identity"] == {"login": "alice", "mode": "oauth", "sudo": None}
|
|
assert httpx_mock.get_requests()[0].headers["Authorization"] == "Bearer user-token-abcdefgh"
|
|
assert resolver.seen == [None] # in-memory transport has no access token
|
|
entry = audit_entries()[-1]
|
|
assert entry["identity_login"] == "alice" and entry["identity_mode"] == "oauth"
|
|
|
|
|
|
async def test_identity_resolution_failure_is_an_error_result(httpx_mock: HTTPXMock) -> None:
|
|
class Refusing:
|
|
async def resolve(self, access: AccessToken | None) -> Identity:
|
|
raise ToolError("unknown identity", status_code=401)
|
|
|
|
async with connected(resolver=Refusing()) as client:
|
|
result = await client.call_tool("gitea_repo_get", REPO_ARGS)
|
|
|
|
assert result.is_error
|
|
assert payload_of(result)["error"] == {"kind": "identity_error", "message": "unknown identity"}
|
|
assert httpx_mock.get_requests() == []
|
|
assert audit_entries()[-1]["outcome"] == "identity_error"
|
|
|
|
|
|
async def test_missing_gitea_token_is_an_identity_error(httpx_mock: HTTPXMock) -> None:
|
|
async with connected(server_settings(gitea_token=None)) as client:
|
|
result = await client.call_tool("gitea_repo_get", REPO_ARGS)
|
|
|
|
assert result.is_error
|
|
assert "GITEA_TOKEN" in payload_of(result)["error"]["message"]
|
|
assert httpx_mock.get_requests() == []
|
|
|
|
|
|
# Uploads
|
|
|
|
|
|
def attachment_args(content: bytes) -> dict[str, Any]:
|
|
return {
|
|
**REPO_ARGS,
|
|
"index": 1,
|
|
"attachment": {
|
|
"filename": "a.bin",
|
|
"content_base64": base64.b64encode(content).decode("ascii"),
|
|
},
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize("size", [10, 9, 8])
|
|
async def test_upload_within_cap_is_sent(httpx_mock: HTTPXMock, size: int) -> None:
|
|
mock_user(httpx_mock)
|
|
httpx_mock.add_response(
|
|
url=f"{API}/repos/octo/demo/issues/1/assets", method="POST", status_code=201, json={"id": 1}
|
|
)
|
|
async with connected(server_settings(max_binary_bytes=10)) as client:
|
|
result = await client.call_tool(
|
|
"gitea_issue_create_issue_attachment", attachment_args(b"x" * size)
|
|
)
|
|
|
|
assert not result.is_error
|
|
|
|
|
|
@pytest.mark.parametrize("size", [11, 12, 13, 1000])
|
|
async def test_upload_over_cap_is_rejected_before_any_request(
|
|
httpx_mock: HTTPXMock, size: int
|
|
) -> None:
|
|
async with connected(server_settings(max_binary_bytes=10)) as client:
|
|
result = await client.call_tool(
|
|
"gitea_issue_create_issue_attachment", attachment_args(b"x" * size)
|
|
)
|
|
|
|
assert result.is_error
|
|
error = payload_of(result)["error"]
|
|
assert error["kind"] == "invalid_argument"
|
|
assert error["path"] == "$.attachment.content_base64"
|
|
assert "MAX_BINARY_BYTES" in error["message"]
|
|
assert httpx_mock.get_requests() == []
|
|
|
|
|
|
def test_max_request_body_size_covers_the_upload_cap() -> None:
|
|
default = max_request_body_size(server_settings(max_binary_bytes=1))
|
|
big = max_request_body_size(server_settings(max_binary_bytes=5_242_880))
|
|
|
|
assert default == 4 * 1024 * 1024
|
|
assert big >= 5_242_880 * 4 // 3 + 1024 * 1024
|
|
assert big > default
|
|
|
|
|
|
# Lifecycle
|
|
|
|
|
|
async def test_server_lifespan_closes_the_executor() -> None:
|
|
settings = server_settings()
|
|
executor = GiteaExecutor(settings)
|
|
server = build_server(settings, executor=executor)
|
|
|
|
async with Client(server) as client:
|
|
await client.list_tools()
|
|
assert executor._client.is_closed is False
|
|
|
|
assert executor._client.is_closed is True
|
|
|
|
|
|
def test_get_tool_input_schema_hook() -> None:
|
|
server = build_server(server_settings(disabled_tools=["gitea_repo_get"]))
|
|
|
|
assert server.get_tool_input_schema is not None
|
|
assert server.get_tool_input_schema("gitea_repo_delete")["type"] == "object" # type: ignore[index]
|
|
assert server.get_tool_input_schema("gitea_repo_get") is None
|
|
assert server.tool_count == 537
|
|
|
|
|
|
def test_upload_cap_ignores_absent_and_malformed_file_arguments() -> None:
|
|
from aegis_gitea_mcp.server import _upload_cap_error
|
|
|
|
operation = CATALOG.get("gitea_issue_create_issue_attachment")
|
|
assert operation is not None
|
|
|
|
assert _upload_cap_error(operation, REPO_ARGS, 10) is None
|
|
assert _upload_cap_error(operation, {"attachment": {"content_base64": 5}}, 10) is None
|
|
assert _upload_cap_error(operation, {"attachment": "not-an-object"}, 10) is None
|