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
 

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