{_INVALID_MESSAGE}
", 400) + + +def _redirect(location: str, status_code: int) -> RedirectResponse: + return RedirectResponse(location, status_code=status_code, headers=_SECURITY_HEADERS) + + +def _text(value: Any, limit: int = 200) -> str: + return value[:limit] if isinstance(value, str) else "" + + +def build_browser_routes(provider: GiteaOAuthProvider) -> list[Route]: + """Return the consent and callback routes for ``provider``.""" + cookies = _Cookies(provider.base_url) + origin_parts = urlsplit(provider.base_url) + expected_origin = f"{origin_parts.scheme}://{origin_parts.netloc}" + + async def consent_page(request: Request) -> Response: + request_id = _text(request.query_params.get("request")) + pending = provider.consent_request(request_id) + csrf = provider.begin_consent(request_id) + if pending is None or csrf is None: + return _invalid() + client = html.escape(pending.client_name or "An unnamed client") + destination = urlsplit(pending.redirect_uri) + host = html.escape(destination.netloc) + body = ( + "{client} wants to use Gitea as you through this MCP server. " + "It can do everything your Gitea account can do within the configured scopes.
" + f"The result goes to {host} "
+ f"({html.escape(pending.redirect_uri)}).
Only continue if you started this sign-in yourself from your ' + "MCP client. If a link sent by someone else brought you here, deny.
" + f'" + ) + response = _page("Allow access to Gitea?", body) + cookies.set(response, cookies.csrf, csrf, same_site="strict") + return response + + async def consent_submit(request: Request) -> Response: + origin = request.headers.get("origin") + if origin is not None and origin != expected_origin: + return _invalid() + form = await request.form() + decision = provider.finish_consent( + _text(form.get("request")), + _text(form.get("csrf")), + _text(request.cookies.get(cookies.csrf)), + approve=form.get("decision") == "approve", + ) + if decision is None: + return _invalid() + response = _redirect(decision.redirect, 303) + cookies.clear(response, cookies.csrf, same_site="strict") + if decision.binding_token is not None: + # Lax: the cookie must come back on Gitea's top-level redirect to us. + cookies.set(response, cookies.binding, decision.binding_token, same_site="lax") + return response + + async def callback(request: Request) -> Response: + params = request.query_params + location = await provider.complete_callback( + _text(params.get("state")), + _text(request.cookies.get(cookies.binding)), + _text(params.get("code"), 4096), + _text(params.get("error")), + ) + response = _invalid() if location is None else _redirect(location, 302) + cookies.clear(response, cookies.binding, same_site="lax") + return response + + return [ + Route(CONSENT_PATH, consent_page, methods=["GET"]), + Route(CONSENT_PATH, consent_submit, methods=["POST"]), + Route(CALLBACK_PATH, callback, methods=["GET"]), + ] diff --git a/src/aegis_gitea_mcp/oauth/components.py b/src/aegis_gitea_mcp/oauth/components.py new file mode 100644 index 0000000..2d01aa2 --- /dev/null +++ b/src/aegis_gitea_mcp/oauth/components.py @@ -0,0 +1,107 @@ +"""Assembles the per-user OAuth pieces for the HTTP transport.""" + +from __future__ import annotations + +import time +from collections.abc import Callable +from typing import cast + +import httpx +from mcp.server.auth.routes import create_protected_resource_routes +from mcp.server.auth.settings import AuthSettings, ClientRegistrationOptions, RevocationOptions +from pydantic import AnyHttpUrl, SecretStr +from starlette.routing import Route + +from aegis_gitea_mcp.audit import AuditLogger +from aegis_gitea_mcp.auth import MCP_SCOPE, AdminIdentityResolver +from aegis_gitea_mcp.config import Settings +from aegis_gitea_mcp.oauth import OAuthSetupError +from aegis_gitea_mcp.oauth.callback import build_browser_routes +from aegis_gitea_mcp.oauth.provider import CONSENT_PATH, GiteaOAuthProvider, OAuthTokenVerifier +from aegis_gitea_mcp.oauth.resolver import OAuthGiteaExecutor, OAuthIdentityResolver +from aegis_gitea_mcp.oauth.store import OAuthStore, OAuthStoreError +from aegis_gitea_mcp.oauth.upstream import CALLBACK_PATH, GiteaUpstream +from aegis_gitea_mcp.transports import HttpAuthExtras + +#: Unauthenticated OAuth endpoints that get the per-IP rate limit. +RATE_LIMITED_PATHS = ("/register", "/authorize", CONSENT_PATH, CALLBACK_PATH) +PROTECTED_RESOURCE_PATH = "/.well-known/oauth-protected-resource" + + +def bare_protected_resource_route(settings: Settings) -> Route: + """``/.well-known/oauth-protected-resource`` serving the same document as the suffixed path. + + The SDK only mounts the RFC 9728 path-suffixed URL; some clients probe the + bare path too. + """ + base = settings.public_base_url or "" + auth = AuthSettings( + issuer_url=base, # type: ignore[arg-type] + resource_server_url=f"{base}{settings.mcp_path}", # type: ignore[arg-type] + validate_token_resource=True, + ) + (suffixed,) = create_protected_resource_routes( + resource_url=cast(AnyHttpUrl, auth.resource_server_url), + authorization_servers=[auth.issuer_url], + scopes_supported=[MCP_SCOPE], + ) + return Route(PROTECTED_RESOURCE_PATH, endpoint=suffixed.endpoint, methods=["GET", "OPTIONS"]) + + +def build_oauth_components( + settings: Settings, + *, + clock: Callable[[], float] = time.time, + client_factory: Callable[[], httpx.AsyncClient] | None = None, + executor_client: httpx.AsyncClient | None = None, + audit: AuditLogger | None = None, +) -> HttpAuthExtras: + """Build the provider, verifier, resolver, executor and routes for per-user OAuth. + + An OAuth-only deployment (no ``GITEA_TOKEN``) is supported: users bring + their own Gitea tokens, and anything that would need admin-token mode + fails closed. + + Args: + settings: Application settings with a complete OAuth configuration. + clock: Wall clock (epoch seconds); tests inject a fake one. + client_factory: Creates the HTTP client for Gitea OAuth exchanges. + executor_client: HTTP client for the tool executor (tests). + audit: Audit logger for OAuth security events. + + Raises: + OAuthSetupError: The configuration is incomplete or the token store + cannot be opened. + """ + problems = settings.oauth_config_errors() + if not settings.oauth_enabled or problems: + raise OAuthSetupError("; ".join(problems) or "OAUTH_ENABLED is not set") + key = cast(SecretStr, settings.token_encryption_key) # checked by oauth_config_errors + try: + store = OAuthStore.load(settings.data_dir, key.get_secret_value(), clock) + except OAuthStoreError as exc: + raise OAuthSetupError(str(exc)) from None + except OSError as exc: + raise OAuthSetupError( + f"cannot read the OAuth token store in {settings.data_dir} ({type(exc).__name__})" + ) from None + + upstream = GiteaUpstream(settings, client_factory=client_factory, clock=clock) + provider = GiteaOAuthProvider(settings, store, upstream, clock=clock, audit=audit) + executor = OAuthGiteaExecutor( + settings, client=executor_client, extra_secrets=[key.get_secret_value()] + ) + admin = AdminIdentityResolver(settings, executor) if settings.gitea_token is not None else None + resolver = OAuthIdentityResolver(store, upstream, admin, clock=clock, audit=audit) + return HttpAuthExtras( + token_verifier=OAuthTokenVerifier(provider), + auth_server_provider=provider, + resolver=resolver, + executor=executor, + extra_routes=(*build_browser_routes(provider), bare_protected_resource_route(settings)), + client_registration_options=ClientRegistrationOptions( + enabled=True, valid_scopes=[MCP_SCOPE], default_scopes=[MCP_SCOPE] + ), + revocation_options=RevocationOptions(enabled=True), + rate_limited_paths=RATE_LIMITED_PATHS, + ) diff --git a/src/aegis_gitea_mcp/oauth/provider.py b/src/aegis_gitea_mcp/oauth/provider.py new file mode 100644 index 0000000..cfce6bf --- /dev/null +++ b/src/aegis_gitea_mcp/oauth/provider.py @@ -0,0 +1,670 @@ +"""OAuth 2.1 authorization server for MCP clients, proxying sign-in to Gitea. + +Implements the SDK's ``OAuthAuthorizationServerProvider``. The SDK mounts +``/authorize``, ``/token``, ``/register``, ``/revoke`` and the metadata, and +does PKCE, redirect URI matching and client authentication; this module holds +the state behind them. + +Sign-in flow (each step single use, every value 256-bit random): + +1. ``/authorize`` (SDK) calls :meth:`GiteaOAuthProvider.authorize`, which parks + the client's request under a random request id and sends the browser to our + consent page. +2. The consent page (``callback.py``) shows which client asks for access and + where the result goes. Approving needs a CSRF token that is bound to the + browser by a ``SameSite=Strict`` cookie. On approval a fresh Gitea ``state`` + and a browser-binding secret are created; the binding secret goes into a + cookie and the browser is sent to Gitea with our own PKCE challenge. +3. ``/oauth/callback`` pops the state, checks the binding cookie (so a Gitea + redirect URL obtained by someone else cannot be completed in a victim's + browser), exchanges Gitea's code, looks up the user and mints our + authorization code for the client's registered redirect URI. +4. ``/token`` (SDK, PKCE checked) calls :meth:`exchange_authorization_code`, + which issues opaque ``agm_`` access and refresh tokens. Only their SHA-256 + hashes are stored. + +The consent step exists because Gitea remembers consent for our (single, +static) Gitea application: without it, anyone could register a client with +their own redirect URI and turn a victim's click on a link into a token for +the victim's account (the confused-deputy problem of OAuth proxies). +""" + +from __future__ import annotations + +import base64 +import hashlib +import json +import logging +import secrets +import time +import uuid +from collections.abc import Callable +from dataclasses import dataclass, field +from typing import Any, Generic, TypeVar +from urllib.parse import urlencode, urlsplit + +from mcp.server.auth.provider import ( + AccessToken, + AuthorizationCode, + AuthorizationParams, + AuthorizeError, + OAuthAuthorizationServerProvider, + RefreshToken, + RegistrationError, + TokenError, + construct_redirect_uri, +) +from mcp.shared.auth import OAuthClientInformationFull, OAuthToken +from pydantic import AnyHttpUrl, ValidationError + +from aegis_gitea_mcp.audit import AuditLogger, get_audit_logger +from aegis_gitea_mcp.auth import MCP_SCOPE +from aegis_gitea_mcp.config import Settings +from aegis_gitea_mcp.oauth.store import ( + ClientRecord, + OAuthStore, + OAuthStoreError, + SessionRecord, + TokenRecord, + token_hash, +) +from aegis_gitea_mcp.oauth.upstream import GiteaUpstream, UpstreamError + +logger = logging.getLogger(__name__) + +TOKEN_PREFIX = "agm_" # noqa: S105 (public token prefix, not a secret) +ACCESS_TOKEN_TTL = 3600 +REFRESH_TOKEN_TTL = 30 * 24 * 3600 +AUTHORIZATION_CODE_TTL = 300 +PENDING_TTL = 600 +MAX_PENDING = 1024 +MAX_CODES = 1024 +MAX_CLIENTS = 500 +MAX_CLIENT_METADATA_BYTES = 8192 +MAX_REDIRECT_URIS = 10 +MAX_SESSIONS_PER_LOGIN = 50 +CONSENT_PATH = "/oauth/consent" + +_LOOPBACK_REDIRECT_HOSTS = frozenset({"localhost", "127.0.0.1", "::1"}) +_CLIENT_AUTH_METHODS = frozenset({"none", "client_secret_post", "client_secret_basic"}) +_REDIRECT_RULE = ( + "redirect URIs must use https://, or http:// on localhost, 127.0.0.1 or [::1], " + "without credentials or a fragment" +) + +V = TypeVar("V") +Clock = Callable[[], float] + + +def redirect_uri_problem(uri: str) -> str | None: + """Return why ``uri`` is not an acceptable redirect URI, or None when it is.""" + try: + parts = urlsplit(uri) + parts.port # noqa: B018 (accessing .port validates it) + except ValueError: + return _REDIRECT_RULE + if parts.fragment or "#" in uri or parts.username is not None or parts.password is not None: + return _REDIRECT_RULE + host = (parts.hostname or "").lower() + if not host: + return _REDIRECT_RULE + if parts.scheme == "https": + return None + if parts.scheme == "http" and host in _LOOPBACK_REDIRECT_HOSTS: + return None + return _REDIRECT_RULE + + +def same_resource(left: str, right: str) -> bool: + """Compare two resource URLs the way the SDK's bearer middleware does.""" + try: + normalized = [str(AnyHttpUrl(value)).removesuffix("/") for value in (left, right)] + except ValidationError: + return False + return normalized[0] == normalized[1] + + +def s256(verifier: str) -> str: + """PKCE S256 challenge for ``verifier``.""" + digest = hashlib.sha256(verifier.encode("ascii")).digest() + return base64.urlsafe_b64encode(digest).decode("ascii").rstrip("=") + + +def _matches(value: str | None, expected_hash: str | None) -> bool: + if not value or not expected_hash: + return False + return secrets.compare_digest(token_hash(value), expected_hash) + + +class _ExpiringMap(Generic[V]): + """Bounded in-memory map with per-entry expiry; the oldest entry is evicted when full.""" + + def __init__(self, capacity: int, clock: Clock) -> None: + self._capacity = capacity + self._clock = clock + self._entries: dict[str, tuple[float, V]] = {} + + def __len__(self) -> int: + return len(self._entries) + + def _purge(self) -> None: + now = self._clock() + for key in [key for key, (expires, _) in self._entries.items() if expires <= now]: + del self._entries[key] + + def put(self, key: str, value: V, ttl: float) -> None: + self._purge() + while len(self._entries) >= self._capacity: + del self._entries[next(iter(self._entries))] + self._entries[key] = (self._clock() + ttl, value) + + def get(self, key: str) -> V | None: + entry = self._entries.get(key) + if entry is None: + return None + if entry[0] <= self._clock(): + del self._entries[key] + return None + return entry[1] + + def pop(self, key: str) -> V | None: + value = self.get(key) + self._entries.pop(key, None) + return value + + +@dataclass +class PendingAuthorization: + """A client's authorization request waiting for consent and the Gitea sign-in.""" + + client_id: str + client_name: str | None + redirect_uri: str + redirect_uri_provided_explicitly: bool + code_challenge: str + client_state: str | None = field(repr=False) + scopes: list[str] + resource: str + code_verifier: str = field(repr=False) + csrf_hash: str | None = field(default=None, repr=False) + binding_hash: str | None = field(default=None, repr=False) + + +class StoredAuthorizationCode(AuthorizationCode): + """Our authorization code plus the session it grants (``code`` holds only its hash).""" + + session_id: str + + +class StoredRefreshToken(RefreshToken): + """A loaded refresh token (``token`` holds only its hash).""" + + session_id: str + token_hash: str + + +@dataclass(frozen=True) +class ConsentDecision: + """Where to send the browser after the consent form, plus the binding cookie value.""" + + redirect: str = field(repr=False) + binding_token: str | None = field(default=None, repr=False) + + +class GiteaOAuthProvider( + OAuthAuthorizationServerProvider[StoredAuthorizationCode, StoredRefreshToken, AccessToken] +): + """The SDK authorization server provider backed by Gitea sign-in. + + Args: + settings: Application settings (``PUBLIC_BASE_URL``, ``MCP_PATH``, scopes). + store: Persistent store for clients, sessions and token hashes. + upstream: Gitea OAuth client. + clock: Wall clock (epoch seconds). + audit: Audit logger for sign-in and revocation events (defaults to the + process-wide one). + """ + + def __init__( + self, + settings: Settings, + store: OAuthStore, + upstream: GiteaUpstream, + *, + clock: Clock = time.time, + audit: AuditLogger | None = None, + ) -> None: + self._settings = settings + self._store = store + self._upstream = upstream + self._clock = clock + self._audit_logger = audit + base = settings.public_base_url or "" + self.base_url = base + self.resource = f"{base}{settings.mcp_path}" + self._requests: _ExpiringMap[PendingAuthorization] = _ExpiringMap(MAX_PENDING, clock) + self._states: _ExpiringMap[PendingAuthorization] = _ExpiringMap(MAX_PENDING, clock) + self._codes: _ExpiringMap[StoredAuthorizationCode] = _ExpiringMap(MAX_CODES, clock) + # Hashes of exchanged codes -> session id, to detect replays (RFC 6749 4.1.2). + self._used_codes: _ExpiringMap[str] = _ExpiringMap(MAX_CODES, clock) + + # Helpers + + def _audit(self, event: str, description: str, severity: str, **metadata: Any) -> None: + try: + audit = self._audit_logger or get_audit_logger() + audit.log_security_event(event, description, severity=severity, metadata=metadata) + except Exception as exc: + logger.error("audit log write failed: %s", type(exc).__name__) + + def _end_session(self, session_id: str, event: str, description: str) -> None: + session = self._store.sessions.get(session_id) + if self._store.end_session(session_id) and session is not None: + self._audit( + event, description, "medium", login=session.login, client_id=session.client_id + ) + + async def _issue_tokens( + self, session: SessionRecord, access_scopes: list[str], refresh_scopes: list[str] + ) -> OAuthToken: + now = self._clock() + access = TOKEN_PREFIX + secrets.token_urlsafe(32) + refresh = TOKEN_PREFIX + secrets.token_urlsafe(32) + self._store.tokens[token_hash(access)] = TokenRecord( + kind="access", + session_id=session.session_id, + client_id=session.client_id, + scopes=list(access_scopes), + expires_at=now + ACCESS_TOKEN_TTL, + ) + self._store.tokens[token_hash(refresh)] = TokenRecord( + kind="refresh", + session_id=session.session_id, + client_id=session.client_id, + scopes=list(refresh_scopes), + expires_at=now + REFRESH_TOKEN_TTL, + ) + await self._store.save() + return OAuthToken( + access_token=access, + token_type="Bearer", # noqa: S106 (token type name) + expires_in=ACCESS_TOKEN_TTL, + refresh_token=refresh, + scope=" ".join(access_scopes), + ) + + # Client registration + + async def get_client(self, client_id: str) -> OAuthClientInformationFull | None: + """Return a registered client (and note that it was used).""" + record = self._store.clients.get(client_id) + if record is None: + return None + record.last_used_at = self._clock() + return OAuthClientInformationFull.model_validate(record.info) + + async def register_client(self, client_info: OAuthClientInformationFull) -> None: + """Validate and persist a dynamically registered client. + + Redirect URIs must be https, or http on a loopback host. When + :data:`MAX_CLIENTS` is reached, the least recently used client without + any session is evicted; when every client has a session, registration + is refused. + + Raises: + RegistrationError: Invalid redirect URIs, metadata or the client cap. + """ + uris = [str(uri) for uri in client_info.redirect_uris or []] + if not uris or len(uris) > MAX_REDIRECT_URIS: + raise RegistrationError( + "invalid_redirect_uri", f"between 1 and {MAX_REDIRECT_URIS} redirect URIs required" + ) + for uri in uris: + problem = redirect_uri_problem(uri) + if problem is not None: + raise RegistrationError("invalid_redirect_uri", problem) + if client_info.token_endpoint_auth_method not in _CLIENT_AUTH_METHODS: + raise RegistrationError( + "invalid_client_metadata", "unsupported token_endpoint_auth_method" + ) + info = client_info.model_dump(mode="json", exclude_none=True) + if len(json.dumps(info)) > MAX_CLIENT_METADATA_BYTES: + raise RegistrationError("invalid_client_metadata", "client metadata is too large") + if len(self._store.clients) >= MAX_CLIENTS and not self._evict_one_client(): + logger.warning("oauth client registration refused: client limit reached") + raise RegistrationError( + "invalid_client_metadata", "client registration limit reached; try again later" + ) + now = self._clock() + self._store.clients[client_info.client_id] = ClientRecord( + info=info, registered_at=now, last_used_at=now + ) + await self._store.save() + logger.info("oauth client registered: %s", client_info.client_id) + + def _evict_one_client(self) -> bool: + candidates = [ + (record.last_used_at, client_id) + for client_id, record in self._store.clients.items() + if not self._store.client_has_sessions(client_id) + ] + if not candidates: + return False + _, client_id = min(candidates) + del self._store.clients[client_id] + logger.info("oauth client evicted (least recently used, no sessions): %s", client_id) + return True + + # Authorization + + async def authorize( + self, client: OAuthClientInformationFull, params: AuthorizationParams + ) -> str: + """Park the request and send the browser to our consent page. + + Raises: + AuthorizeError: ``invalid_target`` when the client asks for a token + for another resource than this server's MCP endpoint. + """ + if params.resource is not None and not same_resource(params.resource, self.resource): + raise AuthorizeError("invalid_target", "unknown resource") + request_id = secrets.token_urlsafe(32) + pending = PendingAuthorization( + client_id=client.client_id, + client_name=client.client_name, + redirect_uri=str(params.redirect_uri), + redirect_uri_provided_explicitly=params.redirect_uri_provided_explicitly, + code_challenge=params.code_challenge, + client_state=params.state, + scopes=list(params.scopes or [MCP_SCOPE]), + resource=self.resource, + code_verifier=secrets.token_urlsafe(64), + ) + self._requests.put(token_hash(request_id), pending, PENDING_TTL) + return f"{self.base_url}{CONSENT_PATH}?{urlencode({'request': request_id})}" + + def consent_request(self, request_id: str) -> PendingAuthorization | None: + """The pending request shown on the consent page, or None when unknown or expired.""" + return self._requests.get(token_hash(request_id)) if request_id else None + + def begin_consent(self, request_id: str) -> str | None: + """Mint the CSRF token for the consent form (the newest page wins).""" + pending = self.consent_request(request_id) + if pending is None: + return None + csrf = secrets.token_urlsafe(32) + pending.csrf_hash = token_hash(csrf) + return csrf + + def finish_consent( + self, request_id: str, csrf_form: str, csrf_cookie: str, *, approve: bool + ) -> ConsentDecision | None: + """Apply the consent decision; None when the request or CSRF proof is invalid.""" + key = token_hash(request_id) if request_id else "" + pending = self._requests.get(key) if key else None + if pending is None: + return None + # Double submit: the form value and the SameSite=Strict cookie must both + # carry the token minted for this request. + if not ( + _matches(csrf_form, pending.csrf_hash) and _matches(csrf_cookie, pending.csrf_hash) + ): + logger.warning("oauth consent rejected: CSRF check failed") + return None + self._requests.pop(key) + if not approve: + logger.info("oauth consent denied for client %s", pending.client_id) + return ConsentDecision( + redirect=construct_redirect_uri( + pending.redirect_uri, error="access_denied", state=pending.client_state + ) + ) + state = secrets.token_urlsafe(32) + binding = secrets.token_urlsafe(32) + pending.binding_hash = token_hash(binding) + self._states.put(token_hash(state), pending, PENDING_TTL) + location = self._upstream.authorize_url( + state=state, code_challenge=s256(pending.code_verifier) + ) + return ConsentDecision(redirect=location, binding_token=binding) + + async def complete_callback( + self, state: str, binding: str, code: str, error: str + ) -> str | None: + """Finish the Gitea sign-in; return the client redirect, or None for a bad request. + + A bad request (unknown, expired or replayed state, or a browser that + did not start the flow) never redirects anywhere. + """ + pending = self._states.pop(token_hash(state)) if state else None + if pending is None: + logger.warning("oauth callback rejected: unknown, expired or replayed state") + return None + if not _matches(binding, pending.binding_hash): + logger.warning("oauth callback rejected: the browser did not start this sign-in") + self._audit( + "oauth_callback_binding_mismatch", + "OAuth callback from a browser that did not start the sign-in", + "high", + client_id=pending.client_id, + ) + return None + if pending.client_id not in self._store.clients: + logger.warning("oauth callback rejected: client no longer registered") + return None + + def back(**params: str | None) -> str: + return construct_redirect_uri( + pending.redirect_uri, **params, state=pending.client_state + ) + + if error: + logger.info("oauth sign-in cancelled or refused at Gitea") + return back(error="access_denied") + if not code: + logger.warning("oauth sign-in failed: Gitea sent no code") + return back(error="server_error") + try: + tokens = await self._upstream.exchange_code(code, pending.code_verifier) + user = await self._upstream.fetch_user(tokens.access_token) + except UpstreamError as exc: + logger.warning("oauth sign-in failed: %s", exc) + return back(error="server_error") + + now = self._clock() + session = SessionRecord( + session_id=uuid.uuid4().hex, + client_id=pending.client_id, + login=user.login, + is_admin=user.is_admin, + gitea_access_token=tokens.access_token, + gitea_refresh_token=tokens.refresh_token, + gitea_expires_at=tokens.expires_at, + scopes=list(pending.scopes), + resource=pending.resource, + created_at=now, + ) + existing = sorted(self._store.sessions_of(user.login), key=lambda item: item.created_at) + for old in existing[: max(0, len(existing) - MAX_SESSIONS_PER_LOGIN + 1)]: + self._end_session( + old.session_id, "oauth_session_evicted", "Oldest OAuth session of the user ended" + ) + self._store.sessions[session.session_id] = session + try: + await self._store.save() + except (OSError, OAuthStoreError) as exc: + self._store.sessions.pop(session.session_id, None) + logger.error( + "oauth sign-in failed: could not save the session (%s)", type(exc).__name__ + ) + return back(error="server_error") + + our_code = secrets.token_urlsafe(32) + self._codes.put( + token_hash(our_code), + StoredAuthorizationCode( + code=token_hash(our_code), + scopes=list(pending.scopes), + expires_at=now + AUTHORIZATION_CODE_TTL, + client_id=pending.client_id, + code_challenge=pending.code_challenge, + redirect_uri=pending.redirect_uri, # type: ignore[arg-type] + redirect_uri_provided_explicitly=pending.redirect_uri_provided_explicitly, + resource=pending.resource, + subject=user.login, + session_id=session.session_id, + ), + AUTHORIZATION_CODE_TTL, + ) + logger.info("oauth sign-in: Gitea user %s via client %s", user.login, pending.client_id) + self._audit( + "oauth_login", + "User signed in with Gitea OAuth", + "low", + login=user.login, + client_id=pending.client_id, + gitea_admin=user.is_admin, + ) + return back(code=our_code) + + # Authorization codes + + async def load_authorization_code( + self, client: OAuthClientInformationFull, authorization_code: str + ) -> StoredAuthorizationCode | None: + """Look up our code; a replayed code revokes the grant it produced.""" + key = token_hash(authorization_code) + used_by = self._used_codes.get(key) + if used_by is not None: + logger.warning("oauth authorization code replayed; revoking the grant") + self._end_session( + used_by, "oauth_code_replay", "Authorization code replayed; grant revoked" + ) + await self._store.save() + return None + code = self._codes.get(key) + if code is None or code.client_id != client.client_id: + return None + return code + + async def exchange_authorization_code( + self, client: OAuthClientInformationFull, authorization_code: StoredAuthorizationCode + ) -> OAuthToken: + """Consume the code (single use) and issue our access and refresh tokens.""" + code = self._codes.pop(authorization_code.code) + if code is None or code.client_id != client.client_id: + raise TokenError("invalid_grant", "authorization code does not exist") + self._used_codes.put(authorization_code.code, code.session_id, AUTHORIZATION_CODE_TTL) + session = self._store.sessions.get(code.session_id) + if session is None or session.client_id != client.client_id: + raise TokenError("invalid_grant", "authorization code does not exist") + return await self._issue_tokens(session, code.scopes, code.scopes) + + # Refresh tokens + + async def load_refresh_token( + self, client: OAuthClientInformationFull, refresh_token: str + ) -> StoredRefreshToken | None: + """Look up a refresh token; reuse of a rotated-out token revokes the grant.""" + if not refresh_token.startswith(TOKEN_PREFIX): + return None + key = token_hash(refresh_token) + record = self._store.tokens.get(key) + if record is None: + reused = self._store.session_by_previous_refresh(key) + if reused is not None: + logger.warning("oauth refresh token reused; revoking the grant") + self._end_session( + reused.session_id, + "oauth_refresh_reuse", + "Rotated refresh token reused; grant revoked", + ) + await self._store.save() + return None + session = self._store.sessions.get(record.session_id) + if record.kind != "refresh" or session is None or record.client_id != client.client_id: + return None + if record.expires_at <= self._clock(): + return None + return StoredRefreshToken( + token=f"sha256:{key}", + client_id=record.client_id, + scopes=list(record.scopes), + expires_at=int(record.expires_at), + resource=session.resource, + subject=session.login, + session_id=record.session_id, + token_hash=key, + ) + + async def exchange_refresh_token( + self, + client: OAuthClientInformationFull, + refresh_token: StoredRefreshToken, + scopes: list[str], + ) -> OAuthToken: + """Rotate: revoke the old pair and issue a new pair for the same session.""" + record = self._store.tokens.get(refresh_token.token_hash) + session = self._store.sessions.get(refresh_token.session_id) + if ( + record is None + or record.kind != "refresh" + or record.client_id != client.client_id + or session is None + ): + raise TokenError("invalid_grant", "refresh token does not exist") + for key in self._store.session_token_hashes(session.session_id): + del self._store.tokens[key] + session.previous_refresh_hash = refresh_token.token_hash + logger.info("oauth tokens refreshed for %s", session.login) + return await self._issue_tokens(session, scopes, record.scopes) + + # Access tokens and revocation + + async def load_access_token(self, token: str) -> AccessToken | None: + """Verify one of our access tokens (``token`` on the result is only a hash).""" + if not token.startswith(TOKEN_PREFIX): + return None + key = token_hash(token) + record = self._store.tokens.get(key) + if record is None or record.kind != "access" or record.expires_at <= self._clock(): + return None + session = self._store.sessions.get(record.session_id) + if session is None: + return None + return AccessToken( + token=f"sha256:{key}", + client_id=record.client_id, + scopes=list(record.scopes), + expires_at=int(record.expires_at), + resource=session.resource, + subject=session.login, + claims={"auth": "oauth", "session_id": session.session_id}, + ) + + async def revoke_token(self, token: AccessToken | RefreshToken) -> None: + """Revoke the whole grant (access and refresh tokens and the Gitea session).""" + if isinstance(token, StoredRefreshToken): + session_id: Any = token.session_id + elif isinstance(token, AccessToken): + session_id = (token.claims or {}).get("session_id") + else: + return + if not isinstance(session_id, str): + return + self._end_session(session_id, "oauth_revoke", "OAuth grant revoked by the client") + await self._store.save() + logger.info("oauth grant revoked") + + +class OAuthTokenVerifier: + """``TokenVerifier`` for our issued access tokens (anything else is not ours).""" + + def __init__(self, provider: GiteaOAuthProvider) -> None: + self._provider = provider + + async def verify_token(self, token: str) -> AccessToken | None: + """Return the access token for an issued ``agm_`` token, else None.""" + if not token.startswith(TOKEN_PREFIX): + return None + return await self._provider.load_access_token(token) diff --git a/src/aegis_gitea_mcp/oauth/resolver.py b/src/aegis_gitea_mcp/oauth/resolver.py new file mode 100644 index 0000000..fc52b3a --- /dev/null +++ b/src/aegis_gitea_mcp/oauth/resolver.py @@ -0,0 +1,188 @@ +"""Identity resolution for per-user OAuth, with automatic upstream refresh. + +Static and OAuth identities never mix: + +* an access token carrying our OAuth marker (``claims["auth"] == "oauth"``) + always resolves to that session's Gitea user, or fails; it never falls back + to the admin token; +* only the static bearer (``client_id == STATIC_CLIENT_ID`` and no OAuth + marker) or no token at all reaches the admin resolver, and in an OAuth-only + deployment (no ``GITEA_TOKEN``) that fails closed; +* any other access token is refused. +""" + +from __future__ import annotations + +import asyncio +import dataclasses +import logging +import time +from collections.abc import Callable +from dataclasses import dataclass, field +from typing import TYPE_CHECKING + +from mcp.server.auth.provider import AccessToken + +from aegis_gitea_mcp.audit import AuditLogger, get_audit_logger +from aegis_gitea_mcp.auth import STATIC_CLIENT_ID, IdentityResolver +from aegis_gitea_mcp.errors import ToolError +from aegis_gitea_mcp.gitea import GiteaExecutor +from aegis_gitea_mcp.gitea.redaction import normalize_secrets +from aegis_gitea_mcp.identity import Identity +from aegis_gitea_mcp.oauth.store import OAuthStore, OAuthStoreError, SessionRecord +from aegis_gitea_mcp.oauth.upstream import GiteaUpstream, UpstreamError, UpstreamInvalidGrant + +if TYPE_CHECKING: + from aegis_gitea_mcp.gitea.executor import _Call + +logger = logging.getLogger(__name__) + +#: Refresh the user's Gitea token when it expires within this many seconds. +REFRESH_MARGIN_SECONDS = 60.0 + +SIGN_IN_AGAIN = "your Gitea sign-in has ended; sign in again through your MCP client" +OAUTH_ONLY = ( + "this server accepts only per-user OAuth sign-in (no GITEA_TOKEN is configured for " + "admin-token mode)" +) + + +@dataclass(frozen=True) +class SessionIdentity(Identity): + """An OAuth identity that also names session secrets to redact from results. + + Attributes: + redact: Further secrets of the session (the Gitea refresh token) that must + never appear in a result. Excluded from ``repr`` and comparisons. + """ + + redact: tuple[str, ...] = field(default=(), repr=False, compare=False) + + +class OAuthGiteaExecutor(GiteaExecutor): + """Executor that also redacts the session secrets carried by a :class:`SessionIdentity`.""" + + def _secrets(self, identity: Identity, call: _Call) -> tuple[str, ...]: + extra = identity.redact if isinstance(identity, SessionIdentity) else () + return normalize_secrets([*super()._secrets(identity, call), *extra]) + + +class OAuthIdentityResolver: + """Maps a verified access token to the identity a tool call runs as. + + Args: + store: OAuth store (sessions and Gitea tokens). + upstream: Gitea OAuth client used to refresh the user's Gitea token. + admin: Resolver for the static bearer (admin-token mode), or None in an + OAuth-only deployment. + clock: Wall clock (epoch seconds). + audit: Audit logger for ended sessions (defaults to the process-wide one). + """ + + def __init__( + self, + store: OAuthStore, + upstream: GiteaUpstream, + admin: IdentityResolver | None, + *, + clock: Callable[[], float] = time.time, + audit: AuditLogger | None = None, + ) -> None: + self._store = store + self._upstream = upstream + self._admin = admin + self._clock = clock + self._audit_logger = audit + self._locks: dict[str, asyncio.Lock] = {} + + async def resolve(self, access: AccessToken | None) -> Identity: + """Return the user's identity for an OAuth token, the admin's for the static bearer. + + Raises: + ToolError: Unknown token kind, ended session, failed upstream refresh, + or admin-token mode requested while it is not configured. + """ + claims = (access.claims or {}) if access is not None else {} + if access is not None and claims.get("auth") == "oauth": + return await self._resolve_session(access, claims.get("session_id")) + if access is None or access.client_id == STATIC_CLIENT_ID: + if self._admin is None: + raise ToolError(OAUTH_ONLY, status_code=401) + return await self._admin.resolve(access) + raise ToolError("unknown identity", status_code=401) + + async def _resolve_session(self, access: AccessToken, session_id: object) -> Identity: + if not isinstance(session_id, str): + raise ToolError("unknown identity", status_code=401) + session = self._store.sessions.get(session_id) + if session is None or session.login != access.subject: + raise ToolError(SIGN_IN_AGAIN, status_code=401) + if self._needs_refresh(session): + session = await self._refresh(session_id) + redact = (session.gitea_refresh_token,) if session.gitea_refresh_token else () + return SessionIdentity( + login=session.login, mode="oauth", token=session.gitea_access_token, redact=redact + ) + + def _needs_refresh(self, session: SessionRecord) -> bool: + expires = session.gitea_expires_at + return expires is not None and expires - self._clock() <= REFRESH_MARGIN_SECONDS + + async def _refresh(self, session_id: str) -> SessionRecord: + lock = self._locks.setdefault(session_id, asyncio.Lock()) + async with lock: + session = self._store.sessions.get(session_id) + if session is None: + raise ToolError(SIGN_IN_AGAIN, status_code=401) + if not self._needs_refresh(session): + return session # refreshed by a concurrent call + still_valid = (session.gitea_expires_at or 0.0) > self._clock() + if session.gitea_refresh_token is None: + if still_valid: + return session + await self._end(session, "Gitea token expired and no refresh token was issued") + raise ToolError(SIGN_IN_AGAIN, status_code=401) + try: + tokens = await self._upstream.refresh(session.gitea_refresh_token) + except UpstreamInvalidGrant: + await self._end(session, "Gitea refused the refresh token") + raise ToolError(SIGN_IN_AGAIN, status_code=401) from None + except UpstreamError as exc: + logger.warning("gitea token refresh failed for %s: %s", session.login, exc) + if still_valid: + return session + raise ToolError( + "could not refresh your Gitea sign-in; try again shortly", status_code=503 + ) from None + updated = dataclasses.replace( + session, + gitea_access_token=tokens.access_token, + gitea_refresh_token=tokens.refresh_token or session.gitea_refresh_token, + gitea_expires_at=tokens.expires_at, + ) + self._store.sessions[session_id] = updated + try: + await self._store.save() + except (OSError, OAuthStoreError) as exc: + logger.error("could not save the refreshed Gitea token (%s)", type(exc).__name__) + logger.info("gitea token refreshed for %s", session.login) + return updated + + async def _end(self, session: SessionRecord, reason: str) -> None: + self._store.end_session(session.session_id) + self._locks.pop(session.session_id, None) + logger.warning("oauth session ended for %s: %s", session.login, reason) + try: + audit = self._audit_logger or get_audit_logger() + audit.log_security_event( + "oauth_session_ended", + reason, + severity="low", + metadata={"login": session.login, "client_id": session.client_id}, + ) + except Exception as exc: + logger.error("audit log write failed: %s", type(exc).__name__) + try: + await self._store.save() + except (OSError, OAuthStoreError) as exc: + logger.error("could not save the OAuth store (%s)", type(exc).__name__) diff --git a/src/aegis_gitea_mcp/oauth/store.py b/src/aegis_gitea_mcp/oauth/store.py new file mode 100644 index 0000000..9be412a --- /dev/null +++ b/src/aegis_gitea_mcp/oauth/store.py @@ -0,0 +1,318 @@ +"""Encrypted on-disk store for OAuth clients, sessions and token hashes. + +One file, ``DATA_DIR/oauth-store.json``, holds the whole state as a single +Fernet-encrypted JSON document. The Fernet key is derived from +``TOKEN_ENCRYPTION_KEY`` with HKDF-SHA256, so the setting can be any long random +string (the setup command writes a ``secrets.token_urlsafe`` value, not a Fernet +key). + +What is stored: + +* registered MCP clients, including their client secrets (the SDK compares the + presented secret with the stored one at ``/token``); +* sessions: the Gitea login and the user's Gitea access and refresh tokens; +* our own access and refresh tokens, only as SHA-256 hashes. + +File safety: the directory is created with mode 0700 and the file is written +with mode 0600 through a fresh ``O_EXCL`` temporary file and an atomic +``os.replace``. A symlink at the store path is refused. A file that cannot be +decrypted or parsed stops startup with a clear message instead of being wiped, +because wiping would silently sign every user out and hide a wrong key. + +Mutations happen synchronously in memory (so a check-and-delete is atomic under +asyncio) and are persisted with :meth:`OAuthStore.save`. +""" + +from __future__ import annotations + +import asyncio +import base64 +import contextlib +import hashlib +import json +import logging +import os +import secrets +import stat +import time +from collections.abc import Callable, Iterator +from dataclasses import asdict, dataclass, field +from pathlib import Path +from typing import Any, Literal + +from cryptography.fernet import Fernet, InvalidToken +from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives.kdf.hkdf import HKDF + +logger = logging.getLogger(__name__) + +STORE_FILENAME = "oauth-store.json" +STORE_VERSION = 1 +HKDF_INFO = b"aegis-gitea-mcp oauth store" + +#: A session that never received tokens (an abandoned sign-in) is dropped after this long. +ABANDONED_SESSION_SECONDS = 600.0 + +Clock = Callable[[], float] +TokenKind = Literal["access", "refresh"] + + +class OAuthStoreError(Exception): + """The OAuth store cannot be used (startup error; the message is safe to print).""" + + +def derive_fernet_key(secret: str) -> bytes: + """Derive the Fernet key for the store from ``TOKEN_ENCRYPTION_KEY`` (HKDF-SHA256).""" + hkdf = HKDF(algorithm=hashes.SHA256(), length=32, salt=None, info=HKDF_INFO) + return base64.urlsafe_b64encode(hkdf.derive(secret.encode("utf-8"))) + + +def token_hash(value: str) -> str: + """SHA-256 hex digest used as the stored form of our tokens, codes and states.""" + return hashlib.sha256(value.encode("utf-8")).hexdigest() + + +@dataclass +class ClientRecord: + """A dynamically registered MCP client. + + Attributes: + info: ``OAuthClientInformationFull`` as JSON (includes the client secret). + registered_at: Registration time (epoch seconds). + last_used_at: Last time the client was looked up (epoch seconds). + """ + + info: dict[str, Any] = field(repr=False) + registered_at: float + last_used_at: float + + +@dataclass +class SessionRecord: + """One signed-in grant: a Gitea user, the client it was granted to and the Gitea tokens. + + Attributes: + session_id: Random identifier (never sent to clients). + client_id: MCP client the grant belongs to. + login: Gitea login of the user. + is_admin: Whether Gitea reported the user as site admin at sign-in + (informational only; Gitea enforces permissions). + gitea_access_token: The user's Gitea access token. + gitea_refresh_token: The user's Gitea refresh token, if Gitea issued one. + gitea_expires_at: Expiry of the Gitea access token (epoch seconds), if known. + scopes: Our scopes granted to the client. + resource: The resource (MCP endpoint URL) the grant is bound to. + created_at: Sign-in time (epoch seconds). + previous_refresh_hash: Hash of the last rotated-out refresh token (reuse detection). + """ + + session_id: str + client_id: str + login: str + is_admin: bool + gitea_access_token: str = field(repr=False) + gitea_refresh_token: str | None = field(repr=False) + gitea_expires_at: float | None + scopes: list[str] + resource: str + created_at: float + previous_refresh_hash: str | None = field(default=None, repr=False) + + +@dataclass +class TokenRecord: + """One of our tokens, stored under its SHA-256 hash.""" + + kind: TokenKind + session_id: str + client_id: str + scopes: list[str] + expires_at: float + + +def _restrict_directory(directory: Path) -> None: + """Create the state directory with mode 0700, or tighten an existing one we own.""" + if not directory.exists(): + directory.mkdir(mode=0o700, parents=True, exist_ok=True) + return + mode = stat.S_IMODE(directory.stat().st_mode) + if mode & 0o077: + try: + directory.chmod(0o700) + except OSError: + logger.warning("could not restrict the permissions of the state directory to 0700") + + +def _parse_document(document: Any) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]: + if not isinstance(document, dict) or document.get("version") != STORE_VERSION: + raise ValueError("unsupported store version") + clients_raw = document.get("clients", {}) + sessions_raw = document.get("sessions", {}) + tokens_raw = document.get("tokens", {}) + if not all(isinstance(part, dict) for part in (clients_raw, sessions_raw, tokens_raw)): + raise ValueError("malformed store sections") + clients = {str(key): ClientRecord(**value) for key, value in clients_raw.items()} + sessions = {str(key): SessionRecord(**value) for key, value in sessions_raw.items()} + tokens = {str(key): TokenRecord(**value) for key, value in tokens_raw.items()} + for record in clients.values(): + if not isinstance(record.info, dict): + raise ValueError("malformed client record") + for token in tokens.values(): + if token.kind not in ("access", "refresh"): + raise ValueError("malformed token record") + return clients, sessions, tokens + + +class OAuthStore: + """In-memory OAuth state backed by one encrypted file. + + Use :meth:`load` to create it. All getters and mutators are synchronous; + :meth:`save` persists the current state (pruning expired entries first). + + Args: + directory: State directory (``DATA_DIR``). + secret: ``TOKEN_ENCRYPTION_KEY``. + clock: Wall clock (epoch seconds); tests inject a fake one. + """ + + def __init__(self, directory: Path, secret: str, clock: Clock = time.time) -> None: + self._directory = directory + self._path = directory / STORE_FILENAME + self._fernet = Fernet(derive_fernet_key(secret)) + self._clock = clock + self._lock = asyncio.Lock() + self.clients: dict[str, ClientRecord] = {} + self.sessions: dict[str, SessionRecord] = {} + self.tokens: dict[str, TokenRecord] = {} + + @classmethod + def load(cls, directory: Path, secret: str, clock: Clock = time.time) -> OAuthStore: + """Open the store, reading and decrypting the file when it exists. + + Raises: + OAuthStoreError: The path is a symlink or not a regular file, or the + file cannot be decrypted or parsed. + """ + store = cls(directory, secret, clock) + store._read() + store.prune() + return store + + @property + def path(self) -> Path: + """Location of the store file.""" + return self._path + + # Reading and writing the file + + def _read(self) -> None: + try: + info = os.lstat(self._path) + except FileNotFoundError: + return + if stat.S_ISLNK(info.st_mode): + raise OAuthStoreError( + f"{self._path} is a symlink; refusing to use it as the OAuth token store" + ) + if not stat.S_ISREG(info.st_mode): + raise OAuthStoreError(f"{self._path} is not a regular file") + fd = os.open(self._path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)) + with os.fdopen(fd, "rb") as handle: + raw = handle.read() + if stat.S_IMODE(info.st_mode) & 0o077: + logger.warning("OAuth store file had loose permissions; restricting it to 0600") + os.chmod(self._path, 0o600) + try: + plain = self._fernet.decrypt(raw) + except InvalidToken: + raise OAuthStoreError( + f"cannot decrypt {self._path}: TOKEN_ENCRYPTION_KEY does not match the key the " + "store was written with, or the file is damaged. Restore the original key, or " + "move the file away to start over (this signs every OAuth user out)." + ) from None + try: + self.clients, self.sessions, self.tokens = _parse_document(json.loads(plain)) + except (ValueError, TypeError, KeyError): + raise OAuthStoreError( + f"{self._path} is not a valid OAuth store document; move the file away to " + "start over (this signs every OAuth user out)" + ) from None + + def _document(self) -> bytes: + document = { + "version": STORE_VERSION, + "clients": {key: asdict(value) for key, value in self.clients.items()}, + "sessions": {key: asdict(value) for key, value in self.sessions.items()}, + "tokens": {key: asdict(value) for key, value in self.tokens.items()}, + } + return json.dumps(document, separators=(",", ":")).encode("utf-8") + + def _write(self, data: bytes) -> None: + _restrict_directory(self._directory) + with contextlib.suppress(FileNotFoundError): + if stat.S_ISLNK(os.lstat(self._path).st_mode): + raise OAuthStoreError( + f"{self._path} is a symlink; refusing to write the OAuth token store" + ) + temporary = self._directory / f".{STORE_FILENAME}.{secrets.token_hex(8)}.tmp" + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0) + fd = os.open(temporary, flags, 0o600) + try: + with os.fdopen(fd, "wb") as handle: + handle.write(data) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary, self._path) + except BaseException: + with contextlib.suppress(FileNotFoundError): + os.unlink(temporary) + raise + + async def save(self) -> None: + """Prune expired entries and persist the state atomically (encrypted, mode 0600).""" + async with self._lock: + self.prune() + self._write(self._fernet.encrypt(self._document())) + + # Maintenance + + def prune(self) -> None: + """Drop expired tokens, and sessions that have no tokens left.""" + now = self._clock() + for key in [key for key, token in self.tokens.items() if token.expires_at <= now]: + del self.tokens[key] + live = {token.session_id for token in self.tokens.values()} + for session_id, session in list(self.sessions.items()): + if session_id in live: + continue + if session.created_at + ABANDONED_SESSION_SECONDS < now: + del self.sessions[session_id] + + # Sessions and tokens + + def session_token_hashes(self, session_id: str) -> list[str]: + """Hashes of every token issued for ``session_id``.""" + return [key for key, token in self.tokens.items() if token.session_id == session_id] + + def end_session(self, session_id: str) -> bool: + """Delete a session and all of its tokens; return whether it existed.""" + for key in self.session_token_hashes(session_id): + del self.tokens[key] + return self.sessions.pop(session_id, None) is not None + + def sessions_of(self, login: str) -> Iterator[SessionRecord]: + """Every session of one Gitea login.""" + return (session for session in self.sessions.values() if session.login == login) + + def session_by_previous_refresh(self, refresh_hash: str) -> SessionRecord | None: + """The session whose last rotated-out refresh token has ``refresh_hash``.""" + for session in self.sessions.values(): + if session.previous_refresh_hash is not None and secrets.compare_digest( + session.previous_refresh_hash, refresh_hash + ): + return session + return None + + def client_has_sessions(self, client_id: str) -> bool: + """Whether any session belongs to ``client_id``.""" + return any(session.client_id == client_id for session in self.sessions.values()) diff --git a/src/aegis_gitea_mcp/oauth/upstream.py b/src/aegis_gitea_mcp/oauth/upstream.py new file mode 100644 index 0000000..ed8efa5 --- /dev/null +++ b/src/aegis_gitea_mcp/oauth/upstream.py @@ -0,0 +1,191 @@ +"""Gitea as the upstream OAuth2 provider: code exchange, refresh and the user lookup. + +We are a confidential client of Gitea (``GITEA_OAUTH_CLIENT_ID`` and +``GITEA_OAUTH_CLIENT_SECRET``, sent form-encoded). Errors never carry Gitea's +response body or any token: callers only learn a coarse kind. +""" + +from __future__ import annotations + +import re +import time +from collections.abc import Callable +from dataclasses import dataclass, field +from typing import Any +from urllib.parse import urlencode + +import httpx + +from aegis_gitea_mcp import __version__ +from aegis_gitea_mcp.config import Settings + +#: Callback path on our server that Gitea redirects to (registered on the Gitea app). +CALLBACK_PATH = "/oauth/callback" + +_MAX_TOKEN_CHARS = 32_767 # Gitea's MAX_TOKEN_LENGTH +# Gitea user names: alphanumerics, dash, underscore and dot (also keeps logs clean). +_LOGIN = re.compile(r"^[A-Za-z0-9_.-]{1,255}$") + + +class UpstreamError(Exception): + """A request to Gitea's OAuth endpoints failed (message is a coarse, safe kind).""" + + +class UpstreamInvalidGrant(UpstreamError): + """Gitea rejected the code or refresh token (``invalid_grant``): the user must sign in again.""" + + +@dataclass(frozen=True) +class UpstreamTokens: + """Tokens returned by Gitea's token endpoint (excluded from ``repr``).""" + + access_token: str = field(repr=False) + refresh_token: str | None = field(repr=False) + expires_at: float | None + + +@dataclass(frozen=True) +class GiteaUser: + """The signed-in Gitea user.""" + + login: str + is_admin: bool + + +def _default_client(settings: Settings) -> httpx.AsyncClient: + return httpx.AsyncClient( + timeout=httpx.Timeout(float(settings.request_timeout_seconds)), + follow_redirects=False, + ) + + +def _valid_token(value: Any) -> bool: + return isinstance(value, str) and 0 < len(value) <= _MAX_TOKEN_CHARS and value.isprintable() + + +class GiteaUpstream: + """Client for Gitea's ``/login/oauth/access_token`` and ``/api/v1/user``. + + Args: + settings: Application settings (Gitea URL, OAuth app credentials, timeout). + client_factory: Creates the ``httpx.AsyncClient`` for one exchange. A new + client per exchange means no cookies or connections are shared. + clock: Wall clock used to turn ``expires_in`` into an expiry time. + """ + + def __init__( + self, + settings: Settings, + *, + client_factory: Callable[[], httpx.AsyncClient] | None = None, + clock: Callable[[], float] = time.time, + ) -> None: + self._settings = settings + self._client_factory = client_factory or (lambda: _default_client(settings)) + self._clock = clock + self._token_url = f"{settings.gitea_url}/login/oauth/access_token" + self._user_url = f"{settings.gitea_api_url}/user" + secret = settings.gitea_oauth_client_secret + self._client_id = settings.gitea_oauth_client_id or "" + self._client_secret = secret.get_secret_value() if secret is not None else "" + self._headers = { + "Accept": "application/json", + "User-Agent": f"aegis-gitea-mcp/{__version__}", + } + + @property + def redirect_uri(self) -> str: + """Our callback URL as registered on the Gitea OAuth application.""" + return f"{self._settings.public_base_url}{CALLBACK_PATH}" + + def authorize_url(self, *, state: str, code_challenge: str) -> str: + """Gitea's authorization URL for one sign-in (PKCE S256, fixed scope list).""" + query = { + "client_id": self._client_id, + "redirect_uri": self.redirect_uri, + "response_type": "code", + "state": state, + "scope": " ".join(self._settings.oauth_scopes), + "code_challenge": code_challenge, + "code_challenge_method": "S256", + } + return f"{self._settings.gitea_url}/login/oauth/authorize?{urlencode(query)}" + + async def exchange_code(self, code: str, code_verifier: str) -> UpstreamTokens: + """Exchange Gitea's authorization code (with our PKCE verifier) for tokens.""" + return await self._token_request( + { + "grant_type": "authorization_code", + "code": code, + "redirect_uri": self.redirect_uri, + "code_verifier": code_verifier, + } + ) + + async def refresh(self, refresh_token: str) -> UpstreamTokens: + """Refresh the user's Gitea access token. + + Raises: + UpstreamInvalidGrant: Gitea no longer accepts the refresh token. + UpstreamError: Any other failure (network, 5xx, malformed answer). + """ + return await self._token_request( + {"grant_type": "refresh_token", "refresh_token": refresh_token} + ) + + async def fetch_user(self, access_token: str) -> GiteaUser: + """Look up the user that owns ``access_token`` (``GET /api/v1/user``).""" + headers = {**self._headers, "Authorization": f"Bearer {access_token}"} + try: + async with self._client_factory() as client: + response = await client.get(self._user_url, headers=headers) + except httpx.HTTPError: + raise UpstreamError("user lookup failed: Gitea unreachable") from None + if response.status_code != 200: + raise UpstreamError(f"user lookup failed: HTTP {response.status_code}") + body = _json(response) or {} + login = body.get("login") + if not isinstance(login, str) or not _LOGIN.match(login): + raise UpstreamError("user lookup failed: malformed answer") + return GiteaUser(login=login, is_admin=body.get("is_admin") is True) + + async def _token_request(self, form: dict[str, str]) -> UpstreamTokens: + payload = {**form, "client_id": self._client_id, "client_secret": self._client_secret} + try: + async with self._client_factory() as client: + response = await client.post(self._token_url, data=payload, headers=self._headers) + except httpx.HTTPError: + raise UpstreamError("token request failed: Gitea unreachable") from None + body = _json(response) + if response.status_code != 200: + if response.status_code in (400, 401) and body and body.get("error") == "invalid_grant": + raise UpstreamInvalidGrant("Gitea rejected the grant") + raise UpstreamError(f"token request failed: HTTP {response.status_code}") + if body is None: + raise UpstreamError("token request failed: malformed answer") + access = body.get("access_token") + refresh = body.get("refresh_token") + token_type = body.get("token_type", "bearer") + expires_in = body.get("expires_in") + if not isinstance(access, str) or not _valid_token(access): + raise UpstreamError("token request failed: malformed answer") + if not isinstance(token_type, str): + raise UpstreamError("token request failed: malformed answer") + if token_type.lower() != "bearer": + raise UpstreamError("token request failed: unexpected token type") + if refresh is not None and not _valid_token(refresh): + raise UpstreamError("token request failed: malformed answer") + expires_at: float | None = None + if isinstance(expires_in, bool): + expires_in = None + if isinstance(expires_in, (int, float)) and expires_in > 0: + expires_at = self._clock() + float(expires_in) + return UpstreamTokens(access_token=access, refresh_token=refresh, expires_at=expires_at) + + +def _json(response: httpx.Response) -> dict[str, Any] | None: + try: + body = response.json() + except ValueError: + return None + return body if isinstance(body, dict) else None diff --git a/src/aegis_gitea_mcp/transports.py b/src/aegis_gitea_mcp/transports.py index 71bb9e6..8506737 100644 --- a/src/aegis_gitea_mcp/transports.py +++ b/src/aegis_gitea_mcp/transports.py @@ -32,8 +32,15 @@ from starlette.types import ASGIApp, Receive, Scope, Send from aegis_gitea_mcp import __version__ from aegis_gitea_mcp.audit import get_audit_logger -from aegis_gitea_mcp.auth import CompositeVerifier, IdentityResolver, StaticBearerVerifier -from aegis_gitea_mcp.config import Settings +from aegis_gitea_mcp.auth import ( + MCP_SCOPE, + CompositeVerifier, + IdentityResolver, + StaticBearerVerifier, +) +from aegis_gitea_mcp.config import Settings, is_loopback_host +from aegis_gitea_mcp.gitea import GiteaExecutor +from aegis_gitea_mcp.oauth import OAuthSetupError from aegis_gitea_mcp.rate_limit import RequestRateLimiter from aegis_gitea_mcp.server import ( SERVER_NAME, @@ -69,6 +76,8 @@ class HttpAuthExtras: extra_routes: Extra unauthenticated routes (for example ``/oauth/callback``). client_registration_options: Dynamic client registration settings. revocation_options: Token revocation settings. + executor: Gitea executor shared by the server and the resolver. + rate_limited_paths: Unauthenticated paths that get the per-IP rate limit. """ token_verifier: TokenVerifier | None = None @@ -77,12 +86,19 @@ class HttpAuthExtras: extra_routes: tuple[Route, ...] = () client_registration_options: ClientRegistrationOptions | None = None revocation_options: RevocationOptions | None = None + executor: GiteaExecutor | None = None + rate_limited_paths: tuple[str, ...] = () -def load_oauth_components(settings: Settings) -> HttpAuthExtras | None: - """Return the per-user OAuth components for ``OAUTH_ENABLED``, or None when unavailable.""" - del settings - return None +def load_oauth_components(settings: Settings) -> HttpAuthExtras: + """Build the per-user OAuth components for ``OAUTH_ENABLED``. + + Raises: + OAuthSetupError: Incomplete configuration or an unusable token store. + """ + from aegis_gitea_mcp.oauth.components import build_oauth_components + + return build_oauth_components(settings) # Health probe @@ -133,15 +149,24 @@ class GiteaProbe: class RateLimitMiddleware: - """ASGI middleware applying the per-IP and per-token limits to the MCP path.""" + """ASGI middleware applying the per-IP and per-token limits to the MCP path. - def __init__(self, app: ASGIApp, mcp_path: str, limiter: RequestRateLimiter) -> None: + ``extra_paths`` (for example the unauthenticated OAuth endpoints) share the limits. + """ + + def __init__( + self, + app: ASGIApp, + mcp_path: str, + limiter: RequestRateLimiter, + extra_paths: Sequence[str] = (), + ) -> None: self.app = app - self._mcp_path = mcp_path + self._paths = frozenset({mcp_path, *(path.rstrip("/") for path in extra_paths)}) self._limiter = limiter async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: - if scope["type"] != "http" or scope["path"].rstrip("/") != self._mcp_path: + if scope["type"] != "http" or scope["path"].rstrip("/") not in self._paths: await self.app(scope, receive, send) return client = scope.get("client") @@ -191,19 +216,27 @@ def transport_security(settings: Settings) -> TransportSecuritySettings: ) +def _resource_url(settings: Settings) -> str: + return f"{_base_url(settings)}{settings.mcp_path}" + + def _auth_settings( settings: Settings, client_registration_options: ClientRegistrationOptions | None, revocation_options: RevocationOptions | None, + *, + authorization_server: bool, ) -> AuthSettings: issuer = _base_url(settings) return AuthSettings( issuer_url=issuer, # type: ignore[arg-type] - resource_server_url=f"{issuer}{settings.mcp_path}", # type: ignore[arg-type] - required_scopes=None, - # The static bearer carries no RFC 8707 resource; a per-user verifier - # that binds tokens to this resource checks the audience itself. - validate_token_resource=False, + resource_server_url=_resource_url(settings), # type: ignore[arg-type] + # With our own authorization server every token (OAuth-issued and the + # static bearer) names this endpoint as its RFC 8707 resource and carries + # the one scope we grant, so the SDK enforces both. Without it, extra + # verifiers are trusted to check the audience themselves. + required_scopes=[MCP_SCOPE] if authorization_server else None, + validate_token_resource=authorization_server, client_registration_options=client_registration_options, revocation_options=revocation_options, ) @@ -219,6 +252,7 @@ def build_http_app( client_registration_options: ClientRegistrationOptions | None = None, revocation_options: RevocationOptions | None = None, probe: GiteaProbe | None = None, + rate_limited_paths: Sequence[str] = (), ) -> Starlette: """Build the streamable HTTP ASGI app (MCP endpoint, ``/health``, ``/``). @@ -236,6 +270,8 @@ def build_http_app( client_registration_options: Dynamic client registration settings. revocation_options: Token revocation settings. probe: Gitea reachability probe for ``/health`` (defaults to a new one). + rate_limited_paths: Further paths (besides the MCP endpoint) that get the + rate limit, for example the OAuth endpoints. Raises: ValueError: ``auth_server_provider`` given without any token verifier. @@ -245,7 +281,11 @@ def build_http_app( verifiers: list[TokenVerifier] = [] if settings.mcp_bearer_token is not None: - verifiers.append(StaticBearerVerifier(settings.mcp_bearer_token.get_secret_value())) + verifiers.append( + StaticBearerVerifier( + settings.mcp_bearer_token.get_secret_value(), resource=_resource_url(settings) + ) + ) if token_verifier is not None: verifiers.append(token_verifier) if auth_server_provider is not None and not verifiers: @@ -289,7 +329,12 @@ def build_http_app( transport_security=transport_security(settings), host=settings.mcp_host, auth=( - _auth_settings(settings, client_registration_options, revocation_options) + _auth_settings( + settings, + client_registration_options, + revocation_options, + authorization_server=auth_server_provider is not None, + ) if verifier is not None else None ), @@ -310,7 +355,10 @@ def build_http_app( app.router.lifespan_context = lifespan app.add_middleware( - RateLimitMiddleware, mcp_path=settings.mcp_path, limiter=RequestRateLimiter(settings) + RateLimitMiddleware, + mcp_path=settings.mcp_path, + limiter=RequestRateLimiter(settings), + extra_paths=tuple(rate_limited_paths), ) return app @@ -319,13 +367,22 @@ def build_http_app( def http_startup_errors(settings: Settings, extras: HttpAuthExtras | None) -> list[str]: - """List reasons the HTTP server must not start (empty when startup may proceed).""" + """List reasons the HTTP server must not start (empty when startup may proceed). + + An OAuth-only deployment (OAuth on, no ``MCP_BEARER_TOKEN``, no + ``GITEA_TOKEN``) may start: every user brings their own Gitea token and + admin-token mode is simply unavailable. ``GITEA_TOKEN`` stays required + whenever admin-token mode is reachable (a static bearer, or no HTTP auth). + """ errors: list[str] = [] bearer = settings.mcp_bearer_token is not None if settings.oauth_enabled: errors.extend(settings.oauth_config_errors()) - if not settings.oauth_config_errors() and extras is None and not bearer: - errors.append("OAUTH_ENABLED=true but per-user OAuth is not available in this build") + if not settings.oauth_config_errors() and extras is None: + errors.append("OAUTH_ENABLED=true but per-user OAuth could not be set up") + public = urlsplit(settings.public_base_url or "") + if public.scheme == "http" and not is_loopback_host(public.hostname or ""): + errors.append("PUBLIC_BASE_URL must use https unless the host is loopback") authenticated = bearer or extras is not None if not settings.is_loopback_bind() and not authenticated: errors.append( @@ -333,7 +390,10 @@ def http_startup_errors(settings: Settings, extras: HttpAuthExtras | None) -> li "MCP_BEARER_TOKEN (or enable OAuth), or bind to 127.0.0.1" ) if (bearer or extras is None) and settings.gitea_token is None: - errors.append("GITEA_TOKEN is required for admin-token mode (static bearer or no auth)") + message = "GITEA_TOKEN is required for admin-token mode (static bearer or no auth)" + if settings.oauth_enabled and bearer: + message += "; for an OAuth-only deployment unset MCP_BEARER_TOKEN" + errors.append(message) return errors @@ -351,11 +411,6 @@ def http_startup_warnings(settings: Settings, extras: HttpAuthExtras | None) -> "bind address are rejected with 421. Set PUBLIC_BASE_URL (the public https URL) " "or ALLOWED_HOSTS (comma separated host names clients use)" ) - if settings.oauth_enabled and extras is None and settings.mcp_bearer_token is not None: - warnings.append( - "OAUTH_ENABLED=true but per-user OAuth is not available in this build: only " - "MCP_BEARER_TOKEN is accepted and every call runs as the GITEA_TOKEN owner" - ) if settings.forwarded_allow_ips == "*": warnings.append( "FORWARDED_ALLOW_IPS=*: X-Forwarded-For is trusted from any peer, so clients can " @@ -382,13 +437,35 @@ def _audit_startup_error(settings: Settings) -> str | None: return None +class AccessLogQueryFilter(logging.Filter): + """Drops query strings from uvicorn access log lines. + + OAuth redirects carry one-time codes and states in the query string + (``/oauth/callback?code=...&state=...``); they must not reach the logs. + """ + + def filter(self, record: logging.LogRecord) -> bool: + """Rewrite the path argument of an access log record; never drops the record.""" + args = record.args + if isinstance(args, tuple) and len(args) >= 3 and isinstance(args[2], str): + path, separator, _ = args[2].partition("?") + if separator: + record.args = (*args[:2], f"{path}?[redacted]", *args[3:]) + return True + + def run_http(settings: Settings) -> int: """Run the streamable HTTP server; return a process exit code. Returns 2 without starting when a startup check fails (including an audit log that cannot be opened for append). """ - extras = load_oauth_components(settings) if settings.oauth_enabled else None + extras: HttpAuthExtras | None = None + if settings.oauth_enabled and not settings.oauth_config_errors(): + try: + extras = load_oauth_components(settings) + except OAuthSetupError as exc: + return _fail(f"per-user OAuth cannot start: {exc}") errors = http_startup_errors(settings, extras) if errors: for message in errors: @@ -400,7 +477,11 @@ def run_http(settings: Settings) -> int: for message in http_startup_warnings(settings, extras): print(f"aegis-gitea-mcp: warning: {message}", file=sys.stderr) try: - server = build_server(settings, resolver=extras.resolver if extras else None) + server = build_server( + settings, + resolver=extras.resolver if extras else None, + executor=extras.executor if extras else None, + ) except ServerConfigError as exc: return _fail(str(exc)) app = build_http_app( @@ -411,7 +492,11 @@ def run_http(settings: Settings) -> int: extra_routes=extras.extra_routes if extras else (), client_registration_options=extras.client_registration_options if extras else None, revocation_options=extras.revocation_options if extras else None, + rate_limited_paths=extras.rate_limited_paths if extras else (), ) + access_logger = logging.getLogger("uvicorn.access") + if not any(isinstance(item, AccessLogQueryFilter) for item in access_logger.filters): + access_logger.addFilter(AccessLogQueryFilter()) uvicorn.run( app, host=settings.mcp_host,