AegisGitea-MCP/tests/test_auth.py

"""Tests for the static bearer verifier, composite verifier and identity resolution."""

from __future__ import annotations

import hmac
from typing import Any

import pytest
from mcp.server.auth.provider import AccessToken
from pytest_httpx import HTTPXMock

from aegis_gitea_mcp import auth as auth_module
from aegis_gitea_mcp.auth import (
    STATIC_CLIENT_ID,
    AdminIdentityResolver,
    CompositeVerifier,
    StaticBearerVerifier,
)
from aegis_gitea_mcp.errors import ToolError
from aegis_gitea_mcp.gitea import GiteaExecutor
from tests.test_server_helpers import API, BEARER_TOKEN, GITEA_TOKEN, server_settings

# Static bearer verifier


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

    assert access is not None
    assert access.client_id == STATIC_CLIENT_ID == "aegis-static-bearer"
    assert access.scopes == ["gitea"]
    assert access.subject == "admin"
    assert access.token == "static"
    assert BEARER_TOKEN not in repr(access)


@pytest.mark.parametrize(
    "presented",
    [
        "",
        "x",
        BEARER_TOKEN + "x",
        BEARER_TOKEN[:-1],
        BEARER_TOKEN.upper(),
        "Bearer " + BEARER_TOKEN,
    ],
)
async def test_wrong_tokens_are_rejected(presented: str) -> None:
    assert await StaticBearerVerifier(BEARER_TOKEN).verify_token(presented) is None


async def test_comparison_is_constant_time_even_for_different_lengths(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    calls: list[tuple[bytes, bytes]] = []
    real = hmac.compare_digest

    def spy(left: Any, right: Any) -> bool:
        calls.append((left, right))
        return real(left, right)

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

    await verifier.verify_token("short")
    await verifier.verify_token(BEARER_TOKEN + "-much-longer-than-expected")
    await verifier.verify_token(BEARER_TOKEN)

    assert len(calls) == 3
    # Digests of equal length are compared, never the raw secrets.
    assert all(len(left) == len(right) == 32 for left, right in calls)
    assert all(BEARER_TOKEN.encode() not in pair for pair in calls)


async def test_composite_verifier_returns_first_match() -> None:
    other = AccessToken(token="t", client_id="oauth-client", scopes=["x"])

    class Accepts:
        async def verify_token(self, token: str) -> AccessToken | None:
            return other if token == "oauth-token" else None

    composite = CompositeVerifier(StaticBearerVerifier(BEARER_TOKEN), Accepts())

    static = await composite.verify_token(BEARER_TOKEN)
    oauth = await composite.verify_token("oauth-token")

    assert static is not None and static.client_id == STATIC_CLIENT_ID
    assert oauth is other
    assert await composite.verify_token("nothing") is None
    assert await CompositeVerifier().verify_token(BEARER_TOKEN) is None


# Identity resolver


async def test_no_access_token_resolves_to_the_admin_identity(httpx_mock: HTTPXMock) -> None:
    httpx_mock.add_response(url=f"{API}/user", json={"login": "root"})
    settings = server_settings()
    resolver = AdminIdentityResolver(settings, GiteaExecutor(settings))

    identity = await resolver.resolve(None)

    assert (identity.login, identity.mode, identity.token) == ("root", "admin-token", GITEA_TOKEN)
    assert GITEA_TOKEN not in repr(identity)
    assert httpx_mock.get_requests()[0].headers["Authorization"] == f"token {GITEA_TOKEN}"


async def test_login_is_looked_up_once_and_cached(httpx_mock: HTTPXMock) -> None:
    httpx_mock.add_response(url=f"{API}/user", json={"login": "root"})
    settings = server_settings()
    resolver = AdminIdentityResolver(settings, GiteaExecutor(settings))
    access = AccessToken(token="static", client_id=STATIC_CLIENT_ID, scopes=["gitea"])

    first = await resolver.resolve(access)
    second = await resolver.resolve(access)

    assert first.login == second.login == "root"
    assert len(httpx_mock.get_requests()) == 1


@pytest.mark.parametrize(
    "response",
    [
        {"status_code": 401, "json": {"message": "bad token"}},
        {"status_code": 200, "json": ["not", "an", "object"]},
        {"status_code": 200, "json": {"login": ""}},
        {"status_code": 200, "json": {"login": 5}},
    ],
)
async def test_failed_lookup_leaves_the_login_unknown_and_is_throttled(
    httpx_mock: HTTPXMock, response: dict[str, Any]
) -> None:
    httpx_mock.add_response(url=f"{API}/user", **response)
    settings = server_settings()
    resolver = AdminIdentityResolver(settings, GiteaExecutor(settings))

    first = await resolver.resolve(None)
    second = await resolver.resolve(None)

    assert first.login is None and second.login is None
    assert first.mode == "admin-token" and first.token == GITEA_TOKEN
    assert len(httpx_mock.get_requests()) == 1  # no retry inside the back-off window


async def test_lookup_is_retried_after_the_backoff(
    httpx_mock: HTTPXMock, monkeypatch: pytest.MonkeyPatch
) -> None:
    httpx_mock.add_response(url=f"{API}/user", status_code=500)
    httpx_mock.add_response(url=f"{API}/user", json={"login": "root"})
    settings = server_settings()
    resolver = AdminIdentityResolver(settings, GiteaExecutor(settings))
    clock = [1000.0]
    monkeypatch.setattr(auth_module.time, "monotonic", lambda: clock[0])

    assert (await resolver.resolve(None)).login is None
    clock[0] += auth_module._LOGIN_RETRY_SECONDS + 1
    assert (await resolver.resolve(None)).login == "root"


async def test_other_client_ids_are_refused() -> None:
    settings = server_settings()
    resolver = AdminIdentityResolver(settings, GiteaExecutor(settings))

    with pytest.raises(ToolError, match="unknown identity"):
        await resolver.resolve(AccessToken(token="x", client_id="oauth-client", scopes=[]))


async def test_missing_gitea_token_is_refused() -> None:
    settings = server_settings(gitea_token=None)
    resolver = AdminIdentityResolver(settings, GiteaExecutor(settings))

    with pytest.raises(ToolError, match="GITEA_TOKEN"):
        await resolver.resolve(None)