diff --git a/src/aegis_gitea_mcp/gitea/encoding.py b/src/aegis_gitea_mcp/gitea/encoding.py index dc11a5d..610dcc9 100644 --- a/src/aegis_gitea_mcp/gitea/encoding.py +++ b/src/aegis_gitea_mcp/gitea/encoding.py @@ -273,7 +273,8 @@ def validate_sudo(sudo: Any) -> str | None: """Validate the optional ``sudo`` login (None when not given).""" if sudo is None: return None - if not isinstance(sudo, str) or not _SUDO_PATTERN.match(sudo): + # fullmatch: ``$`` would also accept a trailing newline. + if not isinstance(sudo, str) or not _SUDO_PATTERN.fullmatch(sudo): raise ArgumentError("sudo must be a Gitea username (letters, digits, '-', '_', '.')") return sudo diff --git a/tests/security/test_malformed_input.py b/tests/security/test_malformed_input.py new file mode 100644 index 0000000..ffb9b7c --- /dev/null +++ b/tests/security/test_malformed_input.py @@ -0,0 +1,498 @@ +"""Malformed input is rejected at the boundary, before Gitea is contacted.""" + +from __future__ import annotations + +import base64 +import json +from typing import Any + +import pytest +from mcp import Client, MCPError +from pytest_httpx import HTTPXMock +from starlette.testclient import TestClient + +from aegis_gitea_mcp import server as server_module +from aegis_gitea_mcp.server import max_request_body_size +from aegis_gitea_mcp.transports import build_http_app +from tests.test_server_helpers import API, BEARER_TOKEN, connected, mock_user, server_settings + +# Many cases are refused before anything is sent, others only after the admin identity lookup: +# what matters here is which requests reach Gitea, asserted explicitly in every test. +pytestmark = pytest.mark.httpx_mock( + assert_all_responses_were_requested=False, assert_all_requests_were_expected=False +) + +REPO = {"owner": "octo", "repo": "demo"} +UPLOAD = "gitea_release_create_release_attachment" +UPLOAD_ARGS = {**REPO, "id": 1} + + +@pytest.fixture(autouse=True) +def _env(mock_env: None) -> None: + """The audit logger reads the process-wide settings.""" + + +def error_of(result: Any) -> dict[str, Any]: + assert result.is_error, result.content[0].text + error: dict[str, Any] = json.loads(result.content[0].text)["error"] + return error + + +def gitea_calls(httpx_mock: HTTPXMock) -> list[str]: + """Requests that reached Gitea, without the admin identity lookup.""" + return [ + str(r.url) for r in httpx_mock.get_requests() if not str(r.url).endswith("/api/v1/user") + ] + + +# Schema: unknown arguments, wrong types, missing arguments + + +@pytest.mark.parametrize( + ("tool", "arguments", "fragment"), + [ + ("gitea_repo_get", {**REPO, "unexpected": 1}, "unexpected argument"), + ("gitea_repo_get", {"owner": "octo"}, "missing required"), + ("gitea_repo_get", {}, "missing required"), + ("gitea_repo_get", {"owner": 7, "repo": "demo"}, "type"), + ("gitea_repo_get", {"owner": ["octo"], "repo": "demo"}, "type"), + ("gitea_repo_get", {"owner": None, "repo": "demo"}, "type"), + ("gitea_repo_get", {"owner": "", "repo": "demo"}, "minLength"), + ("gitea_repo_get", {"owner": "a" * 256, "repo": "demo"}, "maxLength"), + ("gitea_repo_get", {"owner": "bad owner", "repo": "demo"}, "pattern"), + ("gitea_repo_get", {"owner": "octo/evil", "repo": "demo"}, "pattern"), + ("gitea_repo_get", {"owner": "octo", "repo": "de%6Dmo"}, "pattern"), + ("gitea_repo_get", {**REPO, "sudo": 1}, "type"), + ("gitea_issue_get_issue", {**REPO, "index": "one"}, "type"), + ("gitea_issue_get_issue", {**REPO, "index": 1.5}, "type"), + ("gitea_issue_get_issue", {**REPO, "index": True}, "type"), + ( + "gitea_issue_create_issue", + {**REPO, "body": {"title": "x", "labels": ["a"]}}, + "type", + ), + ("gitea_issue_create_issue", {**REPO, "body": {"title": "x", "extra": 1}}, "unexpected"), + ("gitea_issue_create_issue", {**REPO, "body": {"body": "no title"}}, "missing required"), + ("gitea_issue_create_issue", {**REPO, "body": "not an object"}, "type"), + ("gitea_api_request", {"method": "GET"}, "missing required"), + ("gitea_api_request", {"method": "GET", "path": "/x", "extra": 1}, "unexpected"), + ("gitea_api_request", {"method": "GET", "path": 5}, "type"), + ("gitea_api_request", {"method": "GET", "path": ""}, "minLength"), + ("gitea_api_request", {"method": "GET", "path": "x"}, "pattern"), + ("gitea_api_request", {"method": "GET", "path": "/" + "a" * 2048}, "maxLength"), + ("gitea_api_request", {"method": "GET", "path": "/x", "query": [1]}, "type"), + ("gitea_api_request", {"method": "GET", "path": "/x", "query": {"a": {"b": 1}}}, ""), + ], +) +async def test_invalid_arguments_are_rejected_before_gitea( + httpx_mock: HTTPXMock, tool: str, arguments: dict[str, Any], fragment: str +) -> None: + async with connected() as client: + result = await client.call_tool(tool, arguments) + + error = error_of(result) + assert error["kind"] == "invalid_argument" + assert fragment in error["message"] + assert httpx_mock.get_requests() == [] + + +async def test_a_trailing_newline_passes_the_schema_but_not_the_executor( + httpx_mock: HTTPXMock, +) -> None: + # Python's `$` also matches before a trailing newline, so the schema pattern lets + # this through and the executor's own character check is what refuses it. + mock_user(httpx_mock) + async with connected() as client: + result = await client.call_tool("gitea_repo_get", {"owner": "octo", "repo": "demo\n"}) + + assert error_of(result)["kind"] == "invalid_argument" + assert gitea_calls(httpx_mock) == [] + + +async def test_rejection_messages_never_echo_the_values() -> None: + planted = "planted-secret-value-123456" + async with connected() as client: + results = [ + await client.call_tool("gitea_repo_get", {"owner": planted + " x", "repo": "demo"}), + await client.call_tool("gitea_repo_get", {"owner": 1, "repo": planted}), + await client.call_tool("gitea_repo_get", {**REPO, planted: planted}), + ] + + for result in results: + assert planted not in result.content[0].text + + +# Path values (catalog tools and the raw tool) + +BAD_PATH_VALUES = [ + "..", + "a/../b", + "../a", + "a/..", + ".", + "./a", + "a/./b", + "a//b", + "/a", + "a/", + "a\\b", + "a?b", + "a#b", + "a%b", + "a%2e%2e/b", + "%2e%2e", + "a%2Fb", + "a\x00b", + "a\rb", + "a\nb", + "a" + chr(0x2028) + "b", + "a" + chr(0x2029) + "b", + "a\tb", + "a\x7fb", +] + + +@pytest.mark.parametrize("value", BAD_PATH_VALUES, ids=lambda v: repr(v)) +async def test_unsafe_path_values_are_rejected_for_catalog_tools( + httpx_mock: HTTPXMock, value: str +) -> None: + mock_user(httpx_mock) + async with connected() as client: + result = await client.call_tool("gitea_repo_get_raw_file", {**REPO, "filepath": value}) + + assert error_of(result)["kind"] == "invalid_argument" + assert gitea_calls(httpx_mock) == [] + + +@pytest.mark.parametrize("value", [v for v in BAD_PATH_VALUES if v not in {"/a"}], ids=repr) +async def test_unsafe_path_values_are_rejected_for_the_raw_tool( + httpx_mock: HTTPXMock, value: str +) -> None: + mock_user(httpx_mock) + async with connected() as client: + result = await client.call_tool( + "gitea_api_request", {"method": "GET", "path": f"/repos/{value}/x"} + ) + + assert error_of(result)["kind"] == "invalid_argument" + assert gitea_calls(httpx_mock) == [] + + +@pytest.mark.parametrize("path", ["//evil.example/x", "///x", "//", "/a/../b", "/.", "/.."]) +async def test_raw_paths_that_reshape_the_url_are_rejected( + httpx_mock: HTTPXMock, path: str +) -> None: + mock_user(httpx_mock) + async with connected() as client: + result = await client.call_tool("gitea_api_request", {"method": "GET", "path": path}) + + assert error_of(result)["kind"] == "invalid_argument" + assert gitea_calls(httpx_mock) == [] + + +async def test_safe_path_values_are_encoded_not_interpreted(httpx_mock: HTTPXMock) -> None: + mock_user(httpx_mock) + httpx_mock.add_response(url=f"{API}/repos/octo/demo/raw/docs/a%20b.md", text="ok") + async with connected() as client: + result = await client.call_tool( + "gitea_repo_get_raw_file", {**REPO, "filepath": "docs/a b.md"} + ) + + assert not result.is_error + + +# Query names and sudo + + +@pytest.mark.parametrize( + "name", ["token", "access_token", "sudo", "_sudo", "TOKEN", "Access_Token", "SuDo", "_SUDO"] +) +async def test_credential_query_names_are_refused_on_the_raw_tool( + httpx_mock: HTTPXMock, name: str +) -> None: + mock_user(httpx_mock) + async with connected() as client: + result = await client.call_tool( + "gitea_api_request", {"method": "GET", "path": "/user", "query": {name: "x"}} + ) + + assert error_of(result)["kind"] == "invalid_argument" + assert gitea_calls(httpx_mock) == [] + + +@pytest.mark.parametrize( + "sudo", + [ + "bob\r\nX-Injected: 1", + "bob\n", + "bob ", + " bob", + "bo b", + "a" * 256, + "", + "b" + chr(0xF6) + "b", + "bob;", + "bob\x00", + ], + ids=repr, +) +async def test_sudo_values_that_could_inject_headers_are_rejected( + httpx_mock: HTTPXMock, sudo: str +) -> None: + async with connected() as client: + for tool, arguments in ( + ("gitea_repo_get", {**REPO, "sudo": sudo}), + ("gitea_api_request", {"method": "GET", "path": "/user", "sudo": sudo}), + ): + result = await client.call_tool(tool, arguments) + assert error_of(result)["kind"] == "invalid_argument" + + assert gitea_calls(httpx_mock) == [] + + +# Uploads + + +async def test_an_oversized_upload_is_rejected_before_it_is_decoded( + httpx_mock: HTTPXMock, +) -> None: + # Not valid base64 at all: only a size check that runs before decoding reports "limit". + huge = "!" * 4000 + async with connected(server_settings(max_binary_bytes=1000)) as client: + result = await client.call_tool( + UPLOAD, {**UPLOAD_ARGS, "attachment": {"filename": "a.bin", "content_base64": huge}} + ) + + error = error_of(result) + assert error["kind"] == "invalid_argument" + assert "limit" in error["message"] and "MAX_BINARY_BYTES" in error["message"] + assert httpx_mock.get_requests() == [] + + +async def test_an_upload_at_the_limit_is_accepted(httpx_mock: HTTPXMock) -> None: + mock_user(httpx_mock) + httpx_mock.add_response( + method="POST", url=f"{API}/repos/octo/demo/releases/1/assets", status_code=201, json={} + ) + content = base64.b64encode(b"x" * 1000).decode() + async with connected(server_settings(max_binary_bytes=1000)) as client: + result = await client.call_tool( + UPLOAD, {**UPLOAD_ARGS, "attachment": {"filename": "a.bin", "content_base64": content}} + ) + + assert not result.is_error + + +@pytest.mark.parametrize( + "attachment", + [ + {"filename": "a.bin", "content_base64": "not base64!!"}, + {"filename": "a.bin", "content_base64": "YQ"}, # missing padding + {"filename": 'a".bin', "content_base64": "YQ=="}, + {"filename": "a\r\nb.bin", "content_base64": "YQ=="}, + {"filename": "a\nb.bin", "content_base64": "YQ=="}, + {"filename": "a" + chr(0x2028) + "b.bin", "content_base64": "YQ=="}, + {"filename": "", "content_base64": "YQ=="}, + {"filename": "x" * 300, "content_base64": "YQ=="}, + {"filename": "a.bin", "content_base64": "YQ==", "content_type": "text/plain\r\nX: 1"}, + {"filename": "a.bin", "content_base64": "YQ==", "content_type": 5}, + {"filename": "a.bin", "content_base64": "YQ==", "path": "/etc/passwd"}, + {"filename": "a.bin", "content_base64": "YQ==", "unknown": 1}, + {"filename": "a.bin"}, + {"content_base64": "YQ=="}, + {"filename": 1, "content_base64": "YQ=="}, + {"filename": "a.bin", "content_base64": 1}, + "just a string", + ["a.bin", "YQ=="], + ], + ids=lambda a: json.dumps(a)[:60], +) +async def test_malformed_uploads_are_rejected_before_gitea( + httpx_mock: HTTPXMock, attachment: Any +) -> None: + mock_user(httpx_mock) + async with connected() as client: + result = await client.call_tool(UPLOAD, {**UPLOAD_ARGS, "attachment": attachment}) + + assert error_of(result)["kind"] == "invalid_argument" + assert gitea_calls(httpx_mock) == [] + + +# Raw method and READ_ONLY + + +@pytest.mark.parametrize( + "method", ["TRACE", "CONNECT", "OPTIONS", "HEAD", "get", "G ET", "", "GET\r\n"] +) +async def test_raw_methods_outside_the_enum_are_rejected( + httpx_mock: HTTPXMock, method: str +) -> None: + async with connected() as client: + result = await client.call_tool("gitea_api_request", {"method": method, "path": "/user"}) + + assert error_of(result)["kind"] == "invalid_argument" + assert httpx_mock.get_requests() == [] + + +@pytest.mark.parametrize("method", ["POST", "PUT", "PATCH", "DELETE"]) +async def test_read_only_mode_refuses_every_non_get_raw_call( + httpx_mock: HTTPXMock, method: str +) -> None: + async with connected(server_settings(read_only=True)) as client: + result = await client.call_tool( + "gitea_api_request", {"method": method, "path": "/repos/o/r/issues", "body": {}} + ) + + assert error_of(result)["kind"] == "read_only" + assert httpx_mock.get_requests() == [] + + +async def test_read_only_mode_does_not_list_or_run_write_tools(httpx_mock: HTTPXMock) -> None: + async with connected(server_settings(read_only=True)) as client: + names = {tool.name for tool in (await client.list_tools()).tools} + with pytest.raises(MCPError): + await client.call_tool("gitea_issue_create_issue", {**REPO, "body": {"title": "x"}}) + with pytest.raises(MCPError): + await client.call_tool("gitea_repo_delete", REPO) + + assert "gitea_issue_create_issue" not in names and "gitea_repo_delete" not in names + assert "gitea_repo_get" in names + assert httpx_mock.get_requests() == [] + + +# Deep JSON + + +def nest(depth: int, leaf: Any = "x") -> Any: + value = leaf + for _ in range(depth): + value = {"a": value} + return value + + +@pytest.mark.parametrize("depth", [10, 60, 95]) +async def test_deeply_nested_arguments_give_a_result_not_an_exception( + httpx_mock: HTTPXMock, depth: int +) -> None: + mock_user(httpx_mock) + httpx_mock.add_response( + method="POST", url=f"{API}/repos/o/r/issues", status_code=422, json={"message": "no"} + ) + async with connected() as client: + result = await client.call_tool( + "gitea_api_request", + {"method": "POST", "path": "/repos/o/r/issues", "body": nest(depth)}, + ) + + assert result.is_error + assert json.loads(result.content[0].text)["error"]["kind"] == "validation_failed" + + +async def test_a_recursion_error_during_validation_is_a_generic_error_result( + httpx_mock: HTTPXMock, monkeypatch: pytest.MonkeyPatch +) -> None: + def explode(*args: Any, **kwargs: Any) -> None: + raise RecursionError("maximum recursion depth exceeded") + + monkeypatch.setattr(server_module, "validate_arguments", explode) + async with connected() as client: + result = await client.call_tool("gitea_repo_get", REPO) + + assert error_of(result)["kind"] == "internal_error" + assert "RecursionError" not in result.content[0].text + assert httpx_mock.get_requests() == [] + + +@pytest.mark.parametrize("depth", [101, 2000, 50_000]) +async def test_deeply_nested_gitea_responses_are_returned_as_text( + httpx_mock: HTTPXMock, depth: int +) -> None: + mock_user(httpx_mock) + httpx_mock.add_response( + url=f"{API}/repos/octo/demo", + content=("[" * depth + "]" * depth).encode(), + headers={"content-type": "application/json"}, + ) + async with connected() as client: + result = await client.call_tool("gitea_repo_get", REPO) + + assert not result.is_error + assert isinstance(result.structured_content["data"], str) # type: ignore[index] + + +# HTTP boundary + + +def http_client(app: Any) -> TestClient: + return TestClient(app, base_url="http://127.0.0.1:8080") + + +def mcp_headers(**extra: str) -> dict[str, str]: + return { + "Authorization": f"Bearer {BEARER_TOKEN}", + "Content-Type": "application/json", + "Accept": "application/json, text/event-stream", + **extra, + } + + +def test_a_body_over_the_cap_gets_413() -> None: + settings = server_settings(mcp_bearer_token=BEARER_TOKEN) + with http_client(build_http_app(settings)) as client: + response = client.post( + "/mcp", content=b"x" * (max_request_body_size(settings) + 1), headers=mcp_headers() + ) + + assert response.status_code == 413 + + +def test_a_non_json_content_type_gets_400() -> None: + with http_client(build_http_app(server_settings(mcp_bearer_token=BEARER_TOKEN))) as client: + for content_type in ( + "text/plain", + "application/x-www-form-urlencoded", + "multipart/form-data", + ): + response = client.post( + "/mcp", content=b"{}", headers=mcp_headers(**{"Content-Type": content_type}) + ) + assert response.status_code == 400, content_type + + +def test_a_foreign_host_header_gets_421() -> None: + with http_client(build_http_app(server_settings(mcp_bearer_token=BEARER_TOKEN))) as client: + for host in ("evil.example.com", "evil.example.com:8080", "127.0.0.1.evil.example.com"): + response = client.post("/mcp", content=b"{}", headers=mcp_headers(Host=host)) + assert response.status_code == 421, host + + +def test_a_foreign_origin_gets_403() -> None: + with http_client(build_http_app(server_settings(mcp_bearer_token=BEARER_TOKEN))) as client: + for origin in ("https://evil.example.com", "http://127.0.0.1.evil.example.com", "null"): + response = client.post("/mcp", content=b"{}", headers=mcp_headers(Origin=origin)) + assert response.status_code == 403, origin + + +def test_malformed_and_deeply_nested_json_rpc_gets_a_parse_error_not_a_crash() -> None: + deep = "[" * 5000 + "]" * 5000 + bodies = [ + b"not json", + b"{", + b"[]", + b'{"jsonrpc":"2.0"}', + ( + '{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"gitea_api_request",' + f'"arguments":{{"method":"POST","path":"/x","body":{deep}}}}}}}' + ).encode(), + ] + for body in bodies: + with http_client(build_http_app(server_settings(mcp_bearer_token=BEARER_TOKEN))) as client: + response = client.post("/mcp", content=body, headers=mcp_headers()) + assert response.status_code in {400, 422}, body[:20] + assert "Traceback" not in response.text + + +async def test_the_listing_is_unaffected_by_all_of_the_above() -> None: + async with connected() as client: + assert isinstance(client, Client) + assert len((await client.list_tools()).tools) == 538