diff --git a/src/aegis_gitea_mcp/gitea/encoding.py b/src/aegis_gitea_mcp/gitea/encoding.py index df1d7b9..a7f32fb 100644 --- a/src/aegis_gitea_mcp/gitea/encoding.py +++ b/src/aegis_gitea_mcp/gitea/encoding.py @@ -9,11 +9,14 @@ Path safety rules Every value substituted into a path template (and every raw ``request()`` path) is rejected when it is empty, contains a ``..`` or ``.`` path segment, an empty segment, a backslash, ``?``, ``#``, ``%`` or a control character. After -that check the value is percent-encoded. Most parameters are quoted with -``safe=""`` so a ``/`` becomes ``%2F`` and can never open a new path segment. -The parameters in :data:`SLASH_PATH_PARAMS` legitimately hold slashes (file -paths, branch, tag and ref names) and are routed by Gitea through wildcard -routes, so they are quoted with ``safe="/"``. +that check the value is percent-encoded. Parameters are quoted with +``safe=""`` so a ``/`` becomes ``%2F`` and can never open a new path segment, +except the ``(path template, parameter)`` pairs in :data:`SLASH_ROUTES`. +Those are routed by Gitea 28.1.0 through wildcard routes (file paths, branch, +tag and ref names) and only resolve with real slashes, so they are quoted with +``safe="/"``. The decision is keyed on the route and not on the parameter name +because the same name is a single segment elsewhere (``releases/tags/{tag}`` and +``editorconfig/{filepath}`` answer 404 for a raw slash and only accept ``%2F``). """ from __future__ import annotations @@ -29,12 +32,32 @@ from urllib.parse import quote from aegis_gitea_mcp.catalog.model import Operation, Parameter -#: Path parameters that may contain ``/``. Gitea serves these routes with -#: wildcard patterns (``contents/*``, ``branches/*``, ``compare/*``, -#: ``git/refs/*``, ``tags/*``, ``pulls/{base}/*``), so refs such as -#: ``feature/login`` and file paths such as ``src/app/main.py`` are sent with -#: real slashes. Every other parameter is quoted with ``safe=""``. -SLASH_PATH_PARAMS = frozenset({"filepath", "branch", "ref", "basehead", "head", "tag"}) +#: ``(path template, parameter name)`` pairs that Gitea 28.1.0 serves through a +#: wildcard route, verified against a live instance: a raw ``/`` resolves and +#: ``%2F`` would not. Everything else (including same-named parameters on +#: single-segment routes such as ``releases/tags/{tag}`` and +#: ``editorconfig/{filepath}``) is quoted with ``safe=""``. +SLASH_ROUTES = frozenset( + { + ("/repos/{owner}/{repo}/branches/{branch}", "branch"), + ("/repos/{owner}/{repo}/tags/{tag}", "tag"), + ("/repos/{owner}/{repo}/contents/{filepath}", "filepath"), + ("/repos/{owner}/{repo}/contents-ext/{filepath}", "filepath"), + ("/repos/{owner}/{repo}/raw/{filepath}", "filepath"), + ("/repos/{owner}/{repo}/media/{filepath}", "filepath"), + ("/repos/{owner}/{repo}/compare/{basehead}", "basehead"), + ("/repos/{owner}/{repo}/git/refs/{ref}", "ref"), + ("/repos/{owner}/{repo}/commits/{ref}/status", "ref"), + ("/repos/{owner}/{repo}/commits/{ref}/statuses", "ref"), + ("/repos/{owner}/{repo}/pulls/{base}/{head}", "head"), + } +) + + +def allows_slash(template: str, name: str) -> bool: + """Whether ``name`` in the path ``template`` is a wildcard that takes raw slashes.""" + return (template, name) in SLASH_ROUTES + #: Query parameter names that carry credentials or switch the acting user in #: Gitea. They are never forwarded: the identity and the ``sudo`` argument are @@ -63,6 +86,12 @@ _SENSITIVE_BODY_KEYS = frozenset( "auth_password", "auth_token", "private_key", + "authorization_header", + "remote_password", + "mirror_password", + "mirror_token", + "aws_secret_access_key", + "aws_access_key_id", } ) _MIN_SENSITIVE_VALUE_CHARS = 4 @@ -72,6 +101,10 @@ class ArgumentError(ValueError): """An argument cannot be turned into a safe request (message never echoes the value).""" +class PayloadTooLargeError(ArgumentError): + """A request payload is larger than the configured limit (maps to ``payload_too_large``).""" + + @dataclass(frozen=True) class RequestBody: """Body of an outgoing request; at most one of the fields is set. @@ -151,7 +184,7 @@ def render_path(operation: Operation, arguments: Mapping[str, Any]) -> tuple[str raise ArgumentError(f"path placeholder '{name}' has no matching parameter") value = _path_value(parameter, arguments) check_segments(value, f"path parameter '{parameter.arg_name}'") - safe = "/" if name in SLASH_PATH_PARAMS else "" + safe = "/" if allows_slash(operation.path, name) else "" display[name] = value return quote(value, safe=safe) @@ -247,8 +280,14 @@ def _has_control(value: str) -> bool: return any(unicodedata.category(char) in _BAD_CATEGORIES for char in value) +def _decoded_size_bound(encoded: str) -> int: + """Decoded byte count implied by a base64 string's length (before decoding it).""" + padding = len(encoded) - len(encoded.rstrip("=")) + return max(0, len(encoded) * 3 // 4 - min(padding, 2)) + + def _file_part( - parameter: Parameter, value: Any + parameter: Parameter, value: Any, max_file_bytes: int | None ) -> tuple[str, tuple[str | None, bytes, str | None]]: label = f"file argument '{parameter.arg_name}'" if not isinstance(value, Mapping): @@ -263,6 +302,8 @@ def _file_part( filename = _single_line(filename, f"{label} filename", _MAX_FILENAME_CHARS) if '"' in filename: raise ArgumentError(f"{label} filename must not contain a double quote") + if max_file_bytes is not None and _decoded_size_bound(encoded) > max_file_bytes: + raise PayloadTooLargeError(f"{label} is larger than the limit of {max_file_bytes} bytes") try: content = base64.b64decode(encoded, validate=True) except (binascii.Error, ValueError) as exc: @@ -275,14 +316,23 @@ def _file_part( return parameter.name, (filename, content, content_type or "application/octet-stream") -def build_body(operation: Operation, arguments: Mapping[str, Any]) -> RequestBody: +def build_body( + operation: Operation, arguments: Mapping[str, Any], max_file_bytes: int | None = None +) -> RequestBody: """Build the request body from the operation's body and formData parameters. ``formData`` parameters produce a multipart body (files are decoded from base64 strictly). A ``body`` parameter is sent as JSON, or as a plain string when the operation only consumes ``text/plain``. + Args: + operation: Catalog operation. + arguments: Tool arguments keyed by ``arg_name``. + max_file_bytes: Largest decoded size accepted for one file argument; the + base64 length is checked before anything is decoded. None disables it. + Raises: + PayloadTooLargeError: A file argument is larger than ``max_file_bytes``. ArgumentError: A file argument or text body is malformed. """ form_parts: list[tuple[str, tuple[str | None, bytes, str | None]]] = [] @@ -293,7 +343,7 @@ def build_body(operation: Operation, arguments: Mapping[str, Any]) -> RequestBod if value is None: continue if parameter.is_file: - form_parts.append(_file_part(parameter, value)) + form_parts.append(_file_part(parameter, value, max_file_bytes)) else: form_parts.append( (parameter.name, (None, _scalar(value, parameter.arg_name).encode(), None)) diff --git a/src/aegis_gitea_mcp/gitea/executor.py b/src/aegis_gitea_mcp/gitea/executor.py index 61e7dc5..23e0d74 100644 --- a/src/aegis_gitea_mcp/gitea/executor.py +++ b/src/aegis_gitea_mcp/gitea/executor.py @@ -39,6 +39,7 @@ from aegis_gitea_mcp.errors import kind_for_status, sanitize_message from aegis_gitea_mcp.gitea.encoding import ( RAW_METHODS, ArgumentError, + PayloadTooLargeError, RequestBody, build_body, build_query, @@ -115,6 +116,10 @@ _MAX_TEXT_BYTES = 16 * 1024 * 1024 #: Cap on the error body inspected for a message. _MAX_ERROR_BODY_BYTES = 64 * 1024 _RAW_ACCEPT = "application/json, text/plain;q=0.9, */*;q=0.5" +#: Deepest JSON nesting accepted from Gitea. Deeper bodies are returned as text: +#: parsing and the recursive redaction/masking passes must never hit the +#: interpreter recursion limit (a hostile or broken upstream could try). +_MAX_JSON_DEPTH = 100 @dataclass(frozen=True) @@ -184,6 +189,34 @@ def _decode(body: bytes) -> str: return body.decode("utf-8", errors="replace") +def _too_deep(value: Any, limit: int = _MAX_JSON_DEPTH) -> bool: + """True when ``value`` nests lists/dicts deeper than ``limit`` (iterative walk).""" + stack: list[tuple[Any, int]] = [(value, 1)] + while stack: + item, depth = stack.pop() + if isinstance(item, dict): + children: Iterable[Any] = item.values() + elif isinstance(item, list): + children = item + else: + continue + if depth > limit: + return True + stack.extend((child, depth + 1) for child in children) + return False + + +def _parse_json(text: str) -> Any: + """Parse JSON; raises ``ValueError`` for invalid, or too deeply nested, documents.""" + try: + parsed = json.loads(text) + except RecursionError as exc: + raise ValueError("JSON nesting too deep") from exc + if _too_deep(parsed): + raise ValueError("JSON nesting too deep") + return parsed + + def _declared_size(response: httpx.Response) -> int | None: """Content-Length when it describes the decoded body (no content-encoding).""" if response.headers.get("content-encoding", "identity").lower() != "identity": @@ -213,7 +246,7 @@ def _error_message(body: bytes, status: int) -> str: text = _decode(body).strip() message = text try: - parsed = json.loads(text) + parsed = _parse_json(text) except ValueError: parsed = None if isinstance(parsed, dict): @@ -289,7 +322,7 @@ class GiteaExecutor: sudo = validate_sudo(arguments.get("sudo")) encoded_path, display_path = render_path(operation, arguments) query = build_query(operation, arguments) - body = build_body(operation, arguments) + body = build_body(operation, arguments, self._settings.max_binary_bytes) except ArgumentError as exc: call = _Call( method=operation.method, @@ -299,6 +332,8 @@ class GiteaExecutor: accept="", sudo=sudo, ) + if isinstance(exc, PayloadTooLargeError): + return self._error(call, identity, None, "payload_too_large", str(exc)) return self._invalid(call, identity, str(exc)) body_param = next((p for p in operation.parameters if p.location == "body"), None) @@ -409,6 +444,8 @@ class GiteaExecutor: return await self._handle(response, call, identity) except httpx.RequestError as exc: return self._transport_error(exc, call, identity) + except RecursionError: + return self._invalid(call, identity, "request body is nested too deeply") finally: if response is not None: await response.aclose() @@ -509,7 +546,7 @@ class GiteaExecutor: def _decode_body(body: bytes, kind: str) -> Any: if kind == "json": try: - return json.loads(body.decode("utf-8")) + return _parse_json(body.decode("utf-8")) except ValueError: pass return _decode(body) diff --git a/tests/test_executor_hardening.py b/tests/test_executor_hardening.py new file mode 100644 index 0000000..c0a2c44 --- /dev/null +++ b/tests/test_executor_hardening.py @@ -0,0 +1,286 @@ +"""Hardening tests for the executor: slash routing, secret fields, nesting and upload caps.""" + +from __future__ import annotations + +import base64 +import json +from collections.abc import AsyncIterator + +import pytest +from pytest_httpx import HTTPXMock + +from aegis_gitea_mcp.catalog import load_catalog +from aegis_gitea_mcp.gitea import GiteaExecutor +from aegis_gitea_mcp.gitea.encoding import SLASH_ROUTES, allows_slash, render_path +from tests.test_executor_helpers import ( + UPLOAD_ATTACHMENT, + admin_identity, + make_settings, +) + +#: Wildcard routes of Gitea 28.1.0, verified against a live instance (raw slash +#: resolves). Written out independently of the implementation on purpose. +VERIFIED_WILDCARD = { + ("/repos/{owner}/{repo}/branches/{branch}", "branch"), + ("/repos/{owner}/{repo}/tags/{tag}", "tag"), + ("/repos/{owner}/{repo}/contents/{filepath}", "filepath"), + ("/repos/{owner}/{repo}/contents-ext/{filepath}", "filepath"), + ("/repos/{owner}/{repo}/raw/{filepath}", "filepath"), + ("/repos/{owner}/{repo}/media/{filepath}", "filepath"), + ("/repos/{owner}/{repo}/compare/{basehead}", "basehead"), + ("/repos/{owner}/{repo}/git/refs/{ref}", "ref"), + ("/repos/{owner}/{repo}/commits/{ref}/status", "ref"), + ("/repos/{owner}/{repo}/commits/{ref}/statuses", "ref"), + ("/repos/{owner}/{repo}/pulls/{base}/{head}", "head"), +} + + +@pytest.fixture +async def executor() -> AsyncIterator[GiteaExecutor]: + instance = GiteaExecutor(make_settings()) + yield instance + await instance.aclose() + + +def test_slash_decision_matches_verified_routes_for_every_catalog_parameter() -> None: + operations = load_catalog().operations + seen_routes: set[tuple[str, str]] = set() + for operation in operations: + path_params = [p for p in operation.parameters if p.location == "path"] + for target in path_params: + arguments = {p.arg_name: "x" for p in path_params} + arguments[target.arg_name] = "a/b" + encoded, display = render_path(operation, arguments) + key = (operation.path, target.name) + keeps_slash = "a/b" in encoded + assert keeps_slash == (key in VERIFIED_WILDCARD), (operation.operation_id, key) + assert ("a%2Fb" in encoded) == (not keeps_slash) + assert "a/b" in display + if keeps_slash: + seen_routes.add(key) + assert seen_routes == VERIFIED_WILDCARD + assert SLASH_ROUTES == VERIFIED_WILDCARD + + +@pytest.mark.parametrize( + ("template", "name"), + [ + ("/repos/{owner}/{repo}/releases/tags/{tag}", "tag"), + ("/repos/{owner}/{repo}/editorconfig/{filepath}", "filepath"), + ("/repos/{owner}/{repo}/archive/{archive}", "archive"), + ("/repos/{owner}/{repo}/branch_protections/{name}", "name"), + ("/repos/{owner}/{repo}/pulls/{base}/{head}", "base"), + ], +) +def test_same_named_or_single_segment_parameters_are_encoded(template: str, name: str) -> None: + assert not allows_slash(template, name) + + +def test_releases_tag_and_editorconfig_send_percent_2f() -> None: + catalog = load_catalog() + by_path = {(op.method, op.path): op for op in catalog.operations} + release = by_path[("GET", "/repos/{owner}/{repo}/releases/tags/{tag}")] + editorconfig = by_path[("GET", "/repos/{owner}/{repo}/editorconfig/{filepath}")] + wildcard_tag = by_path[("GET", "/repos/{owner}/{repo}/tags/{tag}")] + + base = {"owner": "o", "repo": "r"} + assert ( + render_path(release, {**base, "tag": "rel/1.0"})[0] == "/repos/o/r/releases/tags/rel%2F1.0" + ) + assert ( + render_path(editorconfig, {**base, "filepath": "sub/dir/a.py"})[0] + == "/repos/o/r/editorconfig/sub%2Fdir%2Fa.py" + ) + assert render_path(wildcard_tag, {**base, "tag": "rel/1.0"})[0] == "/repos/o/r/tags/rel/1.0" + + +async def test_release_tag_request_goes_out_with_encoded_slash( + executor: GiteaExecutor, httpx_mock: HTTPXMock +) -> None: + operation = next( + op + for op in load_catalog().operations + if op.path.endswith("/releases/tags/{tag}") and op.method == "GET" + ) + httpx_mock.add_response(json={"id": 1}) + + await executor.execute( + operation, {"owner": "o", "repo": "r", "tag": "rel/1.0"}, admin_identity() + ) + + assert httpx_mock.get_requests()[0].url.raw_path == b"/api/v1/repos/o/r/releases/tags/rel%2F1.0" + + +# Secret-bearing request fields + + +@pytest.mark.parametrize( + "field", + [ + "authorization_header", + "remote_password", + "mirror_password", + "mirror_token", + "aws_secret_access_key", + "aws_access_key_id", + "auth_password", + "auth_token", + ], +) +async def test_secret_request_fields_are_not_echoed( + executor: GiteaExecutor, httpx_mock: HTTPXMock, field: str +) -> None: + value = "Bearer super-secret-value-123" + httpx_mock.add_response(status_code=201, json={"id": 1, field: value, "url": "https://x"}) + operation = next(op for op in load_catalog().operations if op.operation_id == "adminCreateHook") + + outcome = await executor.execute( + operation, + {"body": {"type": "gitea", "config": {"url": "https://x"}, field: value}}, + admin_identity(), + ) + + dumped = json.dumps(outcome.payload) + assert value not in dumped + assert "[REDACTED]" in dumped + + +async def test_hook_authorization_header_is_not_returned( + executor: GiteaExecutor, httpx_mock: HTTPXMock +) -> None: + header = "Basic dXNlcjpwYXNzd29yZA==" + httpx_mock.add_response( + status_code=201, + json={"id": 3, "type": "gitea", "authorization_header": header, "active": True}, + ) + operation = next(op for op in load_catalog().operations if op.operation_id == "orgCreateHook") + + outcome = await executor.execute( + operation, + { + "org": "acme", + "body": { + "type": "gitea", + "config": {"url": "https://hooks.example.com", "content_type": "json"}, + "authorization_header": header, + }, + }, + admin_identity(), + ) + + assert not outcome.is_error + assert header not in json.dumps(outcome.payload) + assert outcome.payload["data"]["authorization_header"] == "[REDACTED]" + assert outcome.payload["data"]["id"] == 3 + + +# Deeply nested JSON + + +def _nested(depth: int) -> str: + return "[" * depth + "]" * depth + + +@pytest.mark.parametrize("depth", [150, 100_000]) +async def test_deeply_nested_json_success_body_is_returned_as_text( + executor: GiteaExecutor, httpx_mock: HTTPXMock, depth: int +) -> None: + operation = next(op for op in load_catalog().operations if op.operation_id == "adminGetHook") + httpx_mock.add_response( + content=_nested(depth).encode(), headers={"content-type": "application/json"} + ) + + outcome = await executor.execute(operation, {"id": 1}, admin_identity()) + + assert not outcome.is_error + assert isinstance(outcome.payload["data"], str) + + +async def test_deeply_nested_json_error_body_still_gives_an_error( + executor: GiteaExecutor, httpx_mock: HTTPXMock +) -> None: + operation = next(op for op in load_catalog().operations if op.operation_id == "adminGetHook") + httpx_mock.add_response( + status_code=500, + content=_nested(100_000).encode(), + headers={"content-type": "application/json"}, + ) + + outcome = await executor.execute(operation, {"id": 1}, admin_identity()) + + assert outcome.is_error + assert outcome.payload["error"]["kind"] == "server_error" + + +async def test_moderately_nested_json_still_parses( + executor: GiteaExecutor, httpx_mock: HTTPXMock +) -> None: + operation = next(op for op in load_catalog().operations if op.operation_id == "adminGetHook") + httpx_mock.add_response( + content=_nested(50).encode(), headers={"content-type": "application/json"} + ) + + outcome = await executor.execute(operation, {"id": 1}, admin_identity()) + + assert isinstance(outcome.payload["data"], list) + + +# Upload size cap + + +async def test_oversized_base64_upload_is_rejected_before_decoding( + httpx_mock: HTTPXMock, monkeypatch: pytest.MonkeyPatch +) -> None: + executor = GiteaExecutor(make_settings(max_binary_bytes=10)) + decoded: list[str] = [] + import base64 as b64module + + real = b64module.b64decode + monkeypatch.setattr( + "aegis_gitea_mcp.gitea.encoding.base64.b64decode", + lambda value, *a, **k: decoded.append(value) or real(value, *a, **k), + ) + try: + outcome = await executor.execute( + UPLOAD_ATTACHMENT, + { + "owner": "o", + "repo": "r", + "id": 1, + "attachment": { + "filename": "big.bin", + "content_base64": base64.b64encode(b"x" * 11).decode(), + }, + }, + admin_identity(), + ) + finally: + await executor.aclose() + + assert outcome.is_error + assert outcome.payload["error"]["kind"] == "payload_too_large" + assert decoded == [] + assert httpx_mock.get_requests() == [] + + +async def test_upload_exactly_at_the_limit_is_sent(httpx_mock: HTTPXMock) -> None: + executor = GiteaExecutor(make_settings(max_binary_bytes=10)) + httpx_mock.add_response(status_code=201, json={"id": 1}) + try: + outcome = await executor.execute( + UPLOAD_ATTACHMENT, + { + "owner": "o", + "repo": "r", + "id": 1, + "attachment": { + "filename": "ok.bin", + "content_base64": base64.b64encode(b"x" * 10).decode(), + }, + }, + admin_identity(), + ) + finally: + await executor.aclose() + + assert not outcome.is_error diff --git a/tests/test_executor_units.py b/tests/test_executor_units.py index 0fc1ebc..bb9b89d 100644 --- a/tests/test_executor_units.py +++ b/tests/test_executor_units.py @@ -5,7 +5,7 @@ from __future__ import annotations import pytest from aegis_gitea_mcp.gitea.encoding import ( - SLASH_PATH_PARAMS, + SLASH_ROUTES, ArgumentError, check_segments, render_path, @@ -20,8 +20,16 @@ from tests.test_executor_helpers import GET_BRANCH, GET_RAW_FILE, PULL_DIFF # Encoding -def test_slash_parameter_set_is_documented_and_minimal() -> None: - assert {"filepath", "branch", "ref", "basehead", "head", "tag"} == SLASH_PATH_PARAMS +def test_slash_routes_are_the_verified_wildcard_routes() -> None: + assert {name for _, name in SLASH_ROUTES} == { + "filepath", + "branch", + "ref", + "basehead", + "head", + "tag", + } + assert len(SLASH_ROUTES) == 11 def test_render_path_returns_encoded_and_display_forms() -> None: