AegisGitea-MCP/tests/security/test_auth_required.py

"""The MCP endpoint rejects every request that does not carry a valid credential."""

from __future__ import annotations

import hashlib
import hmac
from typing import Any

import pytest
from pytest_httpx import HTTPXMock
from starlette.applications import Starlette
from starlette.routing import Route
from starlette.testclient import TestClient

from aegis_gitea_mcp import auth as auth_module
from aegis_gitea_mcp.auth import StaticBearerVerifier
from aegis_gitea_mcp.oauth.provider import TOKEN_PREFIX
from aegis_gitea_mcp.transports import build_http_app
from tests.test_oauth_helpers import (
    API,
    GITEA_TOKEN,
    USER_TOKEN,
    FakeClock,
    browser,
    build_oauth_app,
    mcp_client,
    mcp_status_lifespan,
    sign_in,
)
from tests.test_server_helpers import BEARER_TOKEN, server_settings

MCP_BODY = {"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}}
HOST = "http://127.0.0.1:8080"


@pytest.fixture(autouse=True)
def _env(mock_env: None) -> None:
    """The audit logger and the rate limiter read the process-wide settings."""


def bearer_app() -> Starlette:
    return build_http_app(server_settings(mcp_bearer_token=BEARER_TOKEN))


def post_mcp(client: TestClient, authorization: str | None) -> Any:
    headers = {"Content-Type": "application/json", "Accept": "application/json, text/event-stream"}
    if authorization is not None:
        headers["Authorization"] = authorization
    return client.post("/mcp", json=MCP_BODY, headers=headers)


# No credential, wrong credential


def test_mcp_without_a_bearer_is_401_with_a_challenge_on_the_bearer_app() -> None:
    client = TestClient(bearer_app(), base_url=HOST)

    response = post_mcp(client, None)

    assert response.status_code == 401
    assert response.headers["www-authenticate"].lower().startswith("bearer")


def test_mcp_without_a_bearer_is_401_with_a_challenge_on_the_oauth_app() -> None:
    client = browser(build_oauth_app().app)

    response = post_mcp(client, None)

    assert response.status_code == 401
    challenge = response.headers["www-authenticate"]
    assert challenge.lower().startswith("bearer")
    assert "resource_metadata" in challenge  # tells clients where to find the OAuth server


@pytest.mark.parametrize("method", ["GET", "POST", "DELETE"])
def test_every_method_on_the_endpoint_needs_authentication(method: str) -> None:
    client = TestClient(bearer_app(), base_url=HOST)

    assert client.request(method, "/mcp").status_code == 401


WRONG_BEARERS = {
    "same length": "x" * len(BEARER_TOKEN),
    "one character off": BEARER_TOKEN[:-1] + ("A" if BEARER_TOKEN[-1] != "A" else "B"),
    "shorter": BEARER_TOKEN[:-1],
    "longer": BEARER_TOKEN + "x",
    "prefix only": BEARER_TOKEN[:8],
    "doubled": BEARER_TOKEN * 2,
    "upper case": BEARER_TOKEN.upper(),
    "trailing space": BEARER_TOKEN + " ",
    "trailing newline": BEARER_TOKEN + "\n",
    "another deployment": "another-deployments-bearer-0123456789abcdef",
    "admin gitea token": GITEA_TOKEN,
    "null byte": BEARER_TOKEN + "\x00",
}


@pytest.mark.parametrize("name", WRONG_BEARERS)
def test_a_wrong_bearer_is_401(name: str) -> None:
    client = TestClient(bearer_app(), base_url=HOST)

    try:
        response = post_mcp(client, f"Bearer {WRONG_BEARERS[name]}")
    except Exception as exc:  # an invalid header value is refused before it leaves the client
        assert "header" in str(exc).lower() or isinstance(exc, (ValueError, UnicodeError))
        return

    assert response.status_code == 401


@pytest.mark.parametrize(
    "header",
    [
        "",
        "Bearer",
        "Bearer ",
        f"Basic {BEARER_TOKEN}",
        f"token {BEARER_TOKEN}",
        f"Token {BEARER_TOKEN}",
        BEARER_TOKEN,
        f"Bearer  {BEARER_TOKEN} extra",
        f"Digest {BEARER_TOKEN}",
    ],
)
def test_wrong_scheme_or_shape_of_the_authorization_header_is_401(header: str) -> None:
    client = TestClient(bearer_app(), base_url=HOST)

    assert post_mcp(client, header).status_code == 401


def test_the_right_bearer_is_not_401() -> None:
    assert mcp_status_lifespan(bearer_app(), BEARER_TOKEN) != 401


def test_the_bearer_is_never_accepted_from_a_query_string_or_a_cookie() -> None:
    client = TestClient(bearer_app(), base_url=HOST)

    by_query = client.post(f"/mcp?access_token={BEARER_TOKEN}&token={BEARER_TOKEN}", json=MCP_BODY)
    client.cookies.set("token", BEARER_TOKEN)
    by_cookie = client.post("/mcp", json=MCP_BODY)

    assert by_query.status_code == 401 and by_cookie.status_code == 401


# Constant-time comparison


async def test_the_static_bearer_is_compared_in_constant_time_on_digests(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    seen: list[tuple[Any, Any]] = []
    real = hmac.compare_digest

    def spy(a: Any, b: Any) -> bool:
        seen.append((a, b))
        return real(a, b)

    monkeypatch.setattr(auth_module.hmac, "compare_digest", spy)
    verifier = StaticBearerVerifier(BEARER_TOKEN)

    for candidate in ("", "x", BEARER_TOKEN[:-1], BEARER_TOKEN, BEARER_TOKEN + "x" * 500):
        accepted = await verifier.verify_token(candidate)
        assert (accepted is not None) == (candidate == BEARER_TOKEN)

    assert len(seen) == 5
    # Digests of equal length whatever the candidate length: nothing about the length leaks.
    assert {len(a) for a, _ in seen} == {len(b) for _, b in seen} == {32}
    assert all(isinstance(a, bytes) and isinstance(b, bytes) for a, b in seen)
    assert seen[3][0] == hashlib.sha256(BEARER_TOKEN.encode()).digest()


def test_the_source_has_no_plain_equality_on_the_secret() -> None:
    import inspect

    source = inspect.getsource(StaticBearerVerifier)

    assert "compare_digest" in source
    assert "== self._token" not in source and "self._token ==" not in source


async def test_the_access_token_of_the_static_bearer_never_carries_the_secret() -> None:
    access = await StaticBearerVerifier(BEARER_TOKEN).verify_token(BEARER_TOKEN)

    assert access is not None
    assert BEARER_TOKEN not in repr(access) and BEARER_TOKEN not in access.model_dump_json()


# OAuth tokens


def test_unknown_expired_revoked_and_foreign_oauth_tokens_are_401(httpx_mock: HTTPXMock) -> None:
    clock = FakeClock()
    oauth = build_oauth_app(clock=clock)
    web = browser(oauth.app)
    flow = sign_in(web, httpx_mock)
    access = flow.tokens["access_token"]
    assert access.startswith(TOKEN_PREFIX)
    assert mcp_status_lifespan(oauth.app, access) != 401

    unknown = TOKEN_PREFIX + "A" * 43
    assert post_mcp(web, f"Bearer {unknown}").status_code == 401
    assert post_mcp(web, f"Bearer {TOKEN_PREFIX}").status_code == 401
    assert post_mcp(web, f"Bearer {access}x").status_code == 401
    assert post_mcp(web, f"Bearer {access[:-1]}").status_code == 401

    # Bound to another resource: the token does not work here.
    (session,) = oauth.store.sessions.values()
    real_resource = session.resource
    session.resource = "https://other.example.com/mcp"
    assert post_mcp(web, f"Bearer {access}").status_code == 401
    session.resource = real_resource

    # Revoked by its client.
    revoked = web.post(
        "/revoke",
        data={"token": access, "client_id": flow.client_id, "client_secret": flow.client_secret},
    )
    assert revoked.status_code == 200
    assert post_mcp(web, f"Bearer {access}").status_code == 401


def test_an_expired_oauth_token_is_401(httpx_mock: HTTPXMock) -> None:
    clock = FakeClock()
    oauth = build_oauth_app(clock=clock)
    web = browser(oauth.app)
    flow = sign_in(web, httpx_mock)
    access = flow.tokens["access_token"]

    clock.advance(3600 + 1)

    assert post_mcp(web, f"Bearer {access}").status_code == 401


async def test_an_oauth_token_never_resolves_to_the_admin_identity(
    httpx_mock: HTTPXMock,
) -> None:
    oauth = build_oauth_app()
    flow = sign_in(browser(oauth.app), httpx_mock)
    httpx_mock.add_response(url=f"{API}/repos/octo/demo", json={})
    httpx_mock.add_response(url=f"{API}/admin/users", status_code=403, json={"message": "no"})

    async with mcp_client(oauth.app, flow.tokens["access_token"]) as mcp:
        await mcp.call_tool("gitea_repo_get", {"owner": "octo", "repo": "demo"})
        await mcp.call_tool("gitea_admin_search_users", {})

    calls = [
        r for r in httpx_mock.get_requests() if "/repos/" in str(r.url) or "/admin/" in str(r.url)
    ]
    assert len(calls) == 2
    for request in calls:
        assert request.headers["Authorization"] == f"Bearer {USER_TOKEN}"
        assert GITEA_TOKEN not in request.headers["Authorization"]


def test_an_oauth_only_deployment_has_no_static_bearer_to_guess(httpx_mock: HTTPXMock) -> None:
    from tests.test_oauth_helpers import oauth_settings

    oauth = build_oauth_app(oauth_settings(gitea_token=None, mcp_bearer_token=None))
    web = browser(oauth.app)

    assert post_mcp(web, f"Bearer {BEARER_TOKEN}").status_code == 401
    assert post_mcp(web, None).status_code == 401


# Route inventory

#: Routes reachable without authentication, with the reason each one exists.
PUBLIC_ROUTES = {
    "/": "service banner",
    "/health": "container and client health check",
    "/.well-known/oauth-protected-resource": "RFC 9728 discovery (bare path, OAuth only)",
    "/.well-known/oauth-protected-resource/mcp": "RFC 9728 discovery",
    "/.well-known/oauth-authorization-server": "RFC 8414 discovery (OAuth only)",
    "/authorize": "OAuth authorization endpoint (OAuth only)",
    "/token": "OAuth token endpoint (OAuth only)",
    "/register": "OAuth dynamic client registration (OAuth only)",
    "/revoke": "OAuth revocation (OAuth only)",
    "/oauth/consent": "consent page (OAuth only)",
    "/oauth/callback": "Gitea callback (OAuth only)",
}


def inventory(app: Starlette) -> tuple[set[str], set[str]]:
    """Paths without authentication, and paths guarded by the SDK's bearer middleware."""
    open_paths: set[str] = set()
    guarded: set[str] = set()
    for route in app.routes:
        assert isinstance(route, Route)
        endpoint = type(route.endpoint).__name__
        (guarded if endpoint == "RequireAuthMiddleware" else open_paths).add(route.path)
    return open_paths, guarded


def test_the_routes_reachable_without_authentication_are_exactly_the_known_ones() -> None:
    oauth_open, oauth_guarded = inventory(build_oauth_app().app)
    bearer_open, bearer_guarded = inventory(bearer_app())

    assert oauth_open == set(PUBLIC_ROUTES)
    assert bearer_open <= set(PUBLIC_ROUTES)
    assert {"/", "/health"} <= bearer_open
    # The MCP endpoint is the only guarded route, and it is guarded in both modes.
    assert oauth_guarded == bearer_guarded == {"/mcp"}