AegisGitea-MCP/src/aegis_gitea_mcp/auth.py

"""HTTP authentication: static bearer verification and identity resolution.

The SDK validates the ``Authorization: Bearer`` header with a
:class:`~mcp.server.auth.provider.TokenVerifier`. This module provides the
static-token verifier (admin-token mode), a composite verifier so per-user OAuth
can be added next to it, and the :class:`IdentityResolver` that maps the verified
token to the :class:`~aegis_gitea_mcp.identity.Identity` a tool call runs as.
"""

from __future__ import annotations

import asyncio
import hashlib
import hmac
import time
from typing import TYPE_CHECKING, Protocol

from mcp.server.auth.provider import AccessToken, TokenVerifier

from aegis_gitea_mcp.errors import ToolError
from aegis_gitea_mcp.identity import Identity

if TYPE_CHECKING:
    from aegis_gitea_mcp.config import Settings
    from aegis_gitea_mcp.gitea import GiteaExecutor

#: ``client_id`` of the access token the static bearer verifier returns.
STATIC_CLIENT_ID = "aegis-static-bearer"

#: The one OAuth scope this server grants and requires on the MCP endpoint.
MCP_SCOPE = "gitea"

# After a failed ``GET /user`` the admin login lookup is not retried for this long.
_LOGIN_RETRY_SECONDS = 30.0


def _digest(value: str) -> bytes:
    return hashlib.sha256(value.encode("utf-8")).digest()


class StaticBearerVerifier:
    """Accepts exactly one shared secret (``MCP_BEARER_TOKEN``).

    The comparison is constant time and does not leak the length of the secret:
    both values are hashed first and the digests are compared with
    :func:`hmac.compare_digest`. The returned access token never carries the real
    secret.

    Args:
        token: The expected bearer token.
        resource: Resource (MCP endpoint URL) reported on the access token, so
            the SDK's RFC 8707 audience check accepts the static bearer when
            per-user OAuth enables it.
    """

    def __init__(self, token: str, *, resource: str | None = None) -> None:
        self._expected = _digest(token)
        self._resource = resource

    async def verify_token(self, token: str) -> AccessToken | None:
        """Return an admin access token when ``token`` matches, else ``None``."""
        if not hmac.compare_digest(_digest(token), self._expected):
            return None
        return AccessToken(
            token="static",  # noqa: S106 (placeholder, never the real secret)
            client_id=STATIC_CLIENT_ID,
            scopes=[MCP_SCOPE],
            resource=self._resource,
            subject="admin",
        )


class CompositeVerifier:
    """Tries several verifiers in order; the first one that accepts wins."""

    def __init__(self, *verifiers: TokenVerifier) -> None:
        self._verifiers = tuple(verifiers)

    async def verify_token(self, token: str) -> AccessToken | None:
        """Return the first non-``None`` result of the wrapped verifiers."""
        for verifier in self._verifiers:
            access = await verifier.verify_token(token)
            if access is not None:
                return access
        return None


class IdentityResolver(Protocol):
    """Maps the verified access token of a request to the identity a call runs as."""

    async def resolve(self, access: AccessToken | None) -> Identity:
        """Return the identity for ``access`` (``None`` on stdio and in-memory).

        Raises:
            ToolError: When the access token maps to no known identity.
        """
        ...  # pragma: no cover (protocol)


class AdminIdentityResolver:
    """Admin-token mode: every call runs with ``GITEA_TOKEN``.

    Applies to stdio and in-memory use (no access token) and to the static
    bearer. Any other client id is refused so a per-user resolver can be plugged
    in instead. The login of the token owner is looked up once with
    ``GET /user`` and cached; when the lookup fails the call still proceeds with
    an unknown login.
    """

    def __init__(self, settings: Settings, executor: GiteaExecutor) -> None:
        self._settings = settings
        self._executor = executor
        self._login: str | None = None
        self._retry_at = 0.0
        self._lookup_lock = asyncio.Lock()

    async def resolve(self, access: AccessToken | None) -> Identity:
        """Return the admin identity, or raise for an access token of another kind."""
        if access is not None and access.client_id != STATIC_CLIENT_ID:
            raise ToolError("unknown identity", status_code=401)
        secret = self._settings.gitea_token
        if secret is None:
            raise ToolError("admin-token mode needs GITEA_TOKEN, which is not configured")
        token = secret.get_secret_value()
        if self._login is None and time.monotonic() >= self._retry_at:
            # One lookup for concurrent first calls: later callers wait, then see the result.
            async with self._lookup_lock:
                if self._login is None and time.monotonic() >= self._retry_at:
                    self._login = await self._lookup_login(token)
                    if self._login is None:
                        self._retry_at = time.monotonic() + _LOGIN_RETRY_SECONDS
        return Identity(login=self._login, mode="admin-token", token=token)

    async def _lookup_login(self, token: str) -> str | None:
        probe = Identity(login=None, mode="admin-token", token=token)
        outcome = await self._executor.request("GET", "/user", probe)
        if outcome.is_error:
            return None
        data = outcome.payload.get("data")
        login = data.get("login") if isinstance(data, dict) else None
        return login if isinstance(login, str) and login else None