AegisGitea-MCP/tests/test_executor_hardening.py

"""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