From 5730313574acfcab9676e3f99ceb556c861ace18 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Fri, 3 Apr 2026 00:22:20 +0200 Subject: [PATCH] refactor: remove RFC 8693 token exchange and Keycloak OAuth implementation Nextcloud doesn't support OAuth bearer tokens without upstream patches, making the RFC 8693 token exchange path untestable and dead code. Removed: - nextcloud_mcp_server/auth/token_exchange.py (597 lines) - nextcloud_mcp_server/auth/keycloak_oauth.py (586 lines) - OAUTH_TOKEN_EXCHANGE deployment mode from AuthMode enum - get_session_client_from_context() from context_helper.py - get_session_token() from token_broker.py - enable_token_exchange / token_exchange_cache_ttl config fields - oauth_token_exchange_total Prometheus metric - Keycloak fixture block from tests/conftest.py (~408 lines) - Token exchange unit tests from test_config_validators.py, test_unified_verifier.py, test_management_status_endpoint.py Preserved: - Multi-audience OAuth mode (OAUTH_SINGLE_AUDIENCE) - Login Flow v2 provisioning with elicitation support - Token broker background token management - All existing test coverage for non-exchange paths Co-Authored-By: Claude Opus 4.6 (1M context) --- nextcloud_mcp_server/api/management.py | 2 +- nextcloud_mcp_server/app.py | 81 +-- nextcloud_mcp_server/auth/context_helper.py | 141 +---- nextcloud_mcp_server/auth/keycloak_oauth.py | 585 ----------------- .../auth/provisioning_decorator.py | 9 - nextcloud_mcp_server/auth/token_broker.py | 52 +- nextcloud_mcp_server/auth/token_exchange.py | 596 ------------------ nextcloud_mcp_server/auth/unified_verifier.py | 2 +- nextcloud_mcp_server/config.py | 12 +- nextcloud_mcp_server/config_validators.py | 61 +- nextcloud_mcp_server/context.py | 37 +- nextcloud_mcp_server/observability/metrics.py | 6 - nextcloud_mcp_server/observability/tracing.py | 2 +- tests/conftest.py | 428 +------------ tests/unit/test_config_validators.py | 107 ---- tests/unit/test_management_status_endpoint.py | 2 - tests/unit/test_unified_verifier.py | 49 +- 17 files changed, 48 insertions(+), 2124 deletions(-) delete mode 100644 nextcloud_mcp_server/auth/keycloak_oauth.py delete mode 100644 nextcloud_mcp_server/auth/token_exchange.py diff --git a/nextcloud_mcp_server/api/management.py b/nextcloud_mcp_server/api/management.py index 9fe022fa..0f10d6b9 100644 --- a/nextcloud_mcp_server/api/management.py +++ b/nextcloud_mcp_server/api/management.py @@ -212,7 +212,7 @@ async def get_server_status(request: Request) -> JSONResponse: # Map deployment mode to auth_mode for API response # This helps clients (like Astrolabe) determine which auth flow to use - if mode == AuthMode.OAUTH_SINGLE_AUDIENCE or mode == AuthMode.OAUTH_TOKEN_EXCHANGE: + if mode == AuthMode.OAUTH_SINGLE_AUDIENCE: auth_mode = "oauth" elif mode == AuthMode.MULTI_USER_BASIC: auth_mode = "multi_user_basic" diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index d3835f1f..b58b2849 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -63,7 +63,6 @@ from nextcloud_mcp_server.auth.browser_oauth_routes import ( oauth_logout, ) from nextcloud_mcp_server.auth.client_registration import ensure_oauth_client -from nextcloud_mcp_server.auth.keycloak_oauth import KeycloakOAuthClient from nextcloud_mcp_server.auth.oauth_routes import ( oauth_as_metadata, oauth_authorize, @@ -353,7 +352,7 @@ class OAuthAppContext: nextcloud_host: str token_verifier: object # UnifiedTokenVerifier (ADR-005 compliant) refresh_token_storage: Optional["RefreshTokenStorage"] = None - oauth_client: Optional[object] = None # NextcloudOAuthClient or KeycloakOAuthClient + oauth_client: Optional[object] = None oauth_provider: str = "nextcloud" # "nextcloud" or "keycloak" server_client_id: Optional[str] = ( None # MCP server's OAuth client ID (static or DCR) @@ -772,21 +771,11 @@ async def setup_oauth_config(): token_verifier = UnifiedTokenVerifier(settings) # Log the mode - enable_token_exchange = ( - os.getenv("ENABLE_TOKEN_EXCHANGE", "false").lower() == "true" + logger.info( + "✓ Multi-audience mode enabled (ADR-005) - tokens must contain both MCP and Nextcloud audiences" ) - if enable_token_exchange: - logger.info( - "✓ Token Exchange mode enabled (ADR-005) - exchanging MCP tokens for Nextcloud tokens via RFC 8693" - ) - logger.info(f" MCP audience: {client_id} or {mcp_server_url}") - logger.info(f" Nextcloud audience: {nextcloud_resource_uri}") - else: - logger.info( - "✓ Multi-audience mode enabled (ADR-005) - tokens must contain both MCP and Nextcloud audiences" - ) - logger.info(f" Required MCP audience: {client_id} or {mcp_server_url}") - logger.info(f" Required Nextcloud audience: {nextcloud_resource_uri}") + logger.info(f" Required MCP audience: {client_id} or {mcp_server_url}") + logger.info(f" Required Nextcloud audience: {nextcloud_resource_uri}") if introspection_uri: logger.info("✓ Opaque token introspection enabled (RFC 7662)") @@ -803,45 +792,7 @@ async def setup_oauth_config(): # that are separate from the real-time token exchange flow logger.debug("Token broker available for future offline access features") - # Create OAuth client for server-initiated flows (e.g., token exchange, background workers) oauth_client = None - if enable_offline_access and refresh_token_storage and is_external_idp: - # For external IdP mode, create generic OIDC client for token operations - mcp_server_url = os.getenv("NEXTCLOUD_MCP_SERVER_URL", "http://localhost:8000") - # Note: This redirect_uri is for OAuth client initialization, not used for actual redirects - # since this client is used for backend token operations (exchange, refresh) - redirect_uri = f"{mcp_server_url}/oauth/callback" - - # Extract base URL and realm from discovery URL - # Format: http://keycloak:8080/realms/nextcloud-mcp/.well-known/openid-configuration - # → base_url: http://keycloak:8080, realm: nextcloud-mcp - if "/realms/" in discovery_url: - base_url = discovery_url.split("/realms/")[0] - realm = discovery_url.split("/realms/")[1].split("/")[0] - else: - # Fallback: use issuer to extract base URL - base_url = ( - issuer.rsplit("/realms/", 1)[0] if "/realms/" in issuer else issuer - ) - realm = issuer.split("/realms/")[1] if "/realms/" in issuer else "" - - oauth_client = KeycloakOAuthClient( - keycloak_url=base_url, - realm=realm, - client_id=client_id, - client_secret=client_secret, - redirect_uri=redirect_uri, - ) - await oauth_client.discover() - logger.info( - "✓ OIDC client initialized for token operations (token exchange, refresh)" - ) - elif enable_offline_access and refresh_token_storage: - # For integrated mode, OAuth client could be added later - # For now, token refresh can use httpx directly with discovered endpoints - logger.info( - "OAuth client for token refresh not yet implemented for integrated mode" - ) # Create auth settings mcp_server_url = os.getenv("NEXTCLOUD_MCP_SERVER_URL", "http://localhost:8000") @@ -1049,10 +1000,7 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = logger.debug(f"Mode details:\n{get_mode_summary(mode)}") # Derive helper variables for backward compatibility with existing code - oauth_enabled = mode in ( - AuthMode.OAUTH_SINGLE_AUDIENCE, - AuthMode.OAUTH_TOKEN_EXCHANGE, - ) + oauth_enabled = mode == AuthMode.OAUTH_SINGLE_AUDIENCE # Log hybrid authentication status for multi-user BasicAuth with offline access if mode == AuthMode.MULTI_USER_BASIC and settings.enable_offline_access: logger.info( @@ -1202,7 +1150,7 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = raise # Create MCP server based on detected mode - if mode in (AuthMode.OAUTH_SINGLE_AUDIENCE, AuthMode.OAUTH_TOKEN_EXCHANGE): + if mode == AuthMode.OAUTH_SINGLE_AUDIENCE: logger.info("Configuring MCP server for OAuth mode") # Asynchronously get the OAuth configuration @@ -1320,18 +1268,10 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = logger.info("Skipping semantic search tools (VECTOR_SYNC_ENABLED not set)") # Register OAuth provisioning tools (only when offline access is enabled) - # With token exchange enabled (external IdP), provisioning is not needed for MCP operations - enable_token_exchange = ( - os.getenv("ENABLE_TOKEN_EXCHANGE", "false").lower() == "true" - ) - # Use settings.enable_offline_access which handles both ENABLE_BACKGROUND_OPERATIONS (new) - # and ENABLE_OFFLINE_ACCESS (deprecated) environment variables enable_offline_access_for_tools = settings.enable_offline_access - if oauth_enabled and enable_offline_access_for_tools and not enable_token_exchange: + if oauth_enabled and enable_offline_access_for_tools: logger.info("Registering OAuth provisioning tools for offline access") register_oauth_tools(mcp) - elif oauth_enabled and enable_token_exchange: - logger.info("Skipping provisioning tools registration (token exchange enabled)") elif oauth_enabled and not enable_offline_access_for_tools: logger.info( "Skipping provisioning tools registration (offline access not enabled)" @@ -1965,10 +1905,7 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = # Check authentication configuration # Report the deployment mode, not just whether OAuth is enabled # This helps clients (like Astrolabe) determine which auth flow to use - if ( - mode == AuthMode.OAUTH_SINGLE_AUDIENCE - or mode == AuthMode.OAUTH_TOKEN_EXCHANGE - ): + if mode == AuthMode.OAUTH_SINGLE_AUDIENCE: checks["auth_mode"] = "oauth" checks["auth_configured"] = "ok" elif mode == AuthMode.MULTI_USER_BASIC: diff --git a/nextcloud_mcp_server/auth/context_helper.py b/nextcloud_mcp_server/auth/context_helper.py index 83a22d08..480be8ba 100644 --- a/nextcloud_mcp_server/auth/context_helper.py +++ b/nextcloud_mcp_server/auth/context_helper.py @@ -1,28 +1,17 @@ """Helper functions for extracting OAuth context from MCP requests. -ADR-005 compliant implementation with token exchange caching. +ADR-005 compliant implementation for multi-audience token mode. """ -import hashlib import logging -import time from mcp.server.auth.provider import AccessToken from mcp.server.fastmcp import Context from ..client import NextcloudClient -from ..config import get_settings -from ..observability.metrics import ( - oauth_token_cache_hits_total, - oauth_token_exchange_total, -) -from .token_exchange import exchange_token_for_audience logger = logging.getLogger(__name__) -# Token exchange cache: token_hash -> (exchanged_token, expiry_timestamp) -_exchange_cache: dict[str, tuple[str, float]] = {} - def get_client_from_context(ctx: Context, base_url: str) -> NextcloudClient: """ @@ -79,131 +68,3 @@ def get_client_from_context(ctx: Context, base_url: str) -> NextcloudClient: logger.error(f"Failed to extract OAuth context: {e}") logger.error("This may indicate the server is not running in OAuth mode") raise - - -async def get_session_client_from_context( - ctx: Context, base_url: str -) -> NextcloudClient: - """ - Create NextcloudClient using RFC 8693 token exchange with caching. - - ADR-005 Mode 2: Exchange MCP token for Nextcloud token via RFC 8693. - - This implements the token exchange pattern where: - 1. Extract MCP token from context (validated by UnifiedTokenVerifier) - 2. Check cache for existing exchanged token - 3. If not cached or expired, exchange via RFC 8693 - 4. Cache the exchanged token to minimize exchange frequency - 5. Create client with exchanged token - - CRITICAL: This is where token exchange happens, NOT in the verifier. - The verifier already validated the MCP audience; now we exchange for Nextcloud. - - Note: Nextcloud doesn't support OAuth scopes natively. Scopes are enforced - by the MCP server via @require_scopes decorator, not by the IdP. Therefore, - we don't pass scopes to the token exchange - the MCP server already validated - permissions before calling this function. - - Args: - ctx: MCP request context containing session info - base_url: Nextcloud base URL - - Returns: - NextcloudClient configured with ephemeral exchanged token - - Raises: - AttributeError: If context doesn't contain expected OAuth session data - RuntimeError: If token exchange fails - """ - settings = get_settings() - - try: - # Extract MCP token from context - if hasattr(ctx.request_context.request, "user") and hasattr( - ctx.request_context.request.user, "access_token" - ): - access_token: AccessToken = ctx.request_context.request.user.access_token - mcp_token = access_token.token - username = access_token.resource # Username from UnifiedTokenVerifier - logger.debug(f"Retrieved MCP token for user: {username}") - else: - logger.error("No MCP token found in request context") - raise AttributeError("No access token found in OAuth request context") - - if not username: - logger.error("No username found in access token resource field") - raise ValueError("Username not available in OAuth token context") - - # Check cache for existing exchanged token - cache_key = hashlib.sha256(mcp_token.encode()).hexdigest() - if cache_key in _exchange_cache: - cached_token, expiry = _exchange_cache[cache_key] - if time.time() < expiry: - logger.debug( - f"Using cached exchanged token (expires in {expiry - time.time():.1f}s)" - ) - oauth_token_cache_hits_total.labels(hit="true").inc() - return NextcloudClient.from_token( - base_url=base_url, token=cached_token, username=username - ) - else: - logger.debug("Cached token expired, removing from cache") - del _exchange_cache[cache_key] - - oauth_token_cache_hits_total.labels(hit="false").inc() - - # Perform RFC 8693 token exchange - logger.info(f"Exchanging MCP token for Nextcloud API token (user: {username})") - - try: - # Exchange for Nextcloud resource URI audience - exchanged_token, expires_in = await exchange_token_for_audience( - subject_token=mcp_token, - requested_audience=settings.nextcloud_resource_uri or "nextcloud", - requested_scopes=None, # Nextcloud doesn't support scopes - ) - oauth_token_exchange_total.labels(status="success").inc() - - logger.info(f"Token exchange successful. Token expires in {expires_in}s") - except Exception: - oauth_token_exchange_total.labels(status="error").inc() - raise - - # Cache the exchanged token - # Use the minimum of exchange TTL and configured cache TTL - cache_ttl = min(expires_in, settings.token_exchange_cache_ttl) - _exchange_cache[cache_key] = (exchanged_token, time.time() + cache_ttl) - logger.debug(f"Cached exchanged token for {cache_ttl}s") - - # Clean up expired cache entries - _cleanup_exchange_cache() - - # Create client with exchanged token - return NextcloudClient.from_token( - base_url=base_url, token=exchanged_token, username=username - ) - - except AttributeError as e: - logger.error(f"Failed to extract OAuth context: {e}") - raise - except Exception as e: - logger.error(f"Token exchange failed: {e}") - raise RuntimeError(f"Token exchange required but failed: {e}") from e - - -def _cleanup_exchange_cache(): - """Remove expired entries from the token exchange cache.""" - global _exchange_cache - now = time.time() - expired_keys = [k for k, (_, expiry) in _exchange_cache.items() if expiry <= now] - for key in expired_keys: - del _exchange_cache[key] - if expired_keys: - logger.debug(f"Cleaned up {len(expired_keys)} expired cache entries") - - -def clear_exchange_cache(): - """Clear the entire token exchange cache. Useful for testing.""" - global _exchange_cache - _exchange_cache.clear() - logger.debug("Token exchange cache cleared") diff --git a/nextcloud_mcp_server/auth/keycloak_oauth.py b/nextcloud_mcp_server/auth/keycloak_oauth.py deleted file mode 100644 index 5e840d8f..00000000 --- a/nextcloud_mcp_server/auth/keycloak_oauth.py +++ /dev/null @@ -1,585 +0,0 @@ -""" -Keycloak OAuth 2.0 / OIDC Client - -Handles OAuth flows with Keycloak as the identity provider, including: -- OIDC Discovery -- Authorization Code Flow with PKCE -- Token refresh using refresh tokens (ADR-002 Tier 1) -- Integration with RefreshTokenStorage -""" - -import base64 -import hashlib -import logging -import os -import secrets -from typing import Optional -from urllib.parse import urlencode, urlparse - -import httpx - -from ..http import nextcloud_httpx_client - -logger = logging.getLogger(__name__) - - -class KeycloakOAuthClient: - """OAuth 2.0 client for Keycloak integration""" - - def __init__( - self, - keycloak_url: str, - realm: str, - client_id: str, - client_secret: str, - redirect_uri: str, - scopes: Optional[list[str]] = None, - ): - """ - Initialize Keycloak OAuth client. - - Args: - keycloak_url: Base URL of Keycloak (e.g., http://keycloak:8080) - realm: Keycloak realm name - client_id: OAuth client ID - client_secret: OAuth client secret - redirect_uri: OAuth redirect URI - scopes: List of scopes to request (default: openid, profile, email, offline_access) - """ - self.keycloak_url = keycloak_url.rstrip("/") - self.realm = realm - self.client_id = client_id - self.client_secret = client_secret - self.redirect_uri = redirect_uri - self.scopes = scopes or ["openid", "profile", "email", "offline_access"] - - # Discovered endpoints (populated by discover()) - self.authorization_endpoint: Optional[str] = None - self.token_endpoint: Optional[str] = None - self.userinfo_endpoint: Optional[str] = None - self.jwks_uri: Optional[str] = None - self.end_session_endpoint: Optional[str] = None - - self._http_client: Optional[httpx.AsyncClient] = None - - @classmethod - def from_env(cls) -> "KeycloakOAuthClient": - """ - Create client from environment variables. - - Environment variables: - KEYCLOAK_URL: Keycloak base URL - KEYCLOAK_REALM: Realm name - KEYCLOAK_CLIENT_ID: Client ID - KEYCLOAK_CLIENT_SECRET: Client secret - NEXTCLOUD_MCP_SERVER_URL: MCP server URL (for redirect URI) - - Returns: - KeycloakOAuthClient instance - - Raises: - ValueError: If required environment variables are missing - """ - keycloak_url = os.getenv("KEYCLOAK_URL") - realm = os.getenv("KEYCLOAK_REALM") - client_id = os.getenv("KEYCLOAK_CLIENT_ID") - client_secret = os.getenv("KEYCLOAK_CLIENT_SECRET") - server_url = os.getenv("NEXTCLOUD_MCP_SERVER_URL", "http://localhost:8000") - - if not all([keycloak_url, realm, client_id, client_secret]): - raise ValueError( - "Missing required environment variables: " - "KEYCLOAK_URL, KEYCLOAK_REALM, KEYCLOAK_CLIENT_ID, KEYCLOAK_CLIENT_SECRET" - ) - - # Parse server URL to construct redirect URI - # Note: This is for OAuth client initialization, not used for actual redirects - # since this client is used for backend token operations (exchange, refresh) - parsed_url = urlparse(server_url) - redirect_uri = f"{parsed_url.scheme}://{parsed_url.netloc}/oauth/callback" - - return cls( - keycloak_url=keycloak_url, - realm=realm, - client_id=client_id, - client_secret=client_secret, - redirect_uri=redirect_uri, - ) - - async def _get_http_client(self) -> httpx.AsyncClient: - """Get or create HTTP client""" - if self._http_client is None: - self._http_client = nextcloud_httpx_client(timeout=30.0) - return self._http_client - - async def close(self) -> None: - """Close HTTP client""" - if self._http_client: - await self._http_client.aclose() - self._http_client = None - - async def discover(self) -> None: - """ - Perform OIDC discovery to get endpoint URLs. - - Raises: - httpx.HTTPError: If discovery fails - """ - discovery_url = ( - f"{self.keycloak_url}/realms/{self.realm}/.well-known/openid-configuration" - ) - - logger.info(f"Discovering Keycloak endpoints at {discovery_url}") - - client = await self._get_http_client() - response = await client.get(discovery_url) - response.raise_for_status() - - discovery_data = response.json() - - self.authorization_endpoint = discovery_data["authorization_endpoint"] - self.token_endpoint = discovery_data["token_endpoint"] - self.userinfo_endpoint = discovery_data["userinfo_endpoint"] - self.jwks_uri = discovery_data.get("jwks_uri") - self.end_session_endpoint = discovery_data.get("end_session_endpoint") - - logger.info( - f"✓ Discovered Keycloak endpoints:\n" - f" Authorization: {self.authorization_endpoint}\n" - f" Token: {self.token_endpoint}\n" - f" Userinfo: {self.userinfo_endpoint}\n" - f" JWKS: {self.jwks_uri}" - ) - - def generate_pkce_challenge(self) -> tuple[str, str]: - """ - Generate PKCE code verifier and challenge. - - Returns: - Tuple of (code_verifier, code_challenge) - """ - - # Generate code verifier (43-128 characters) - code_verifier = secrets.token_urlsafe(32) - - # Generate code challenge using S256 method (base64url-encoded SHA256) - digest = hashlib.sha256(code_verifier.encode()).digest() - code_challenge = base64.urlsafe_b64encode(digest).decode().rstrip("=") - - return code_verifier, code_challenge - - async def get_authorization_url( - self, - state: str, - code_challenge: str, - extra_params: Optional[dict[str, str]] = None, - ) -> str: - """ - Build authorization URL for OAuth flow. - - Args: - state: CSRF protection state parameter - code_challenge: PKCE code challenge - extra_params: Additional query parameters - - Returns: - Authorization URL - - Raises: - RuntimeError: If discover() hasn't been called - """ - if not self.authorization_endpoint: - await self.discover() - - if not self.authorization_endpoint: - raise RuntimeError("Authorization endpoint not discovered") - - params = { - "client_id": self.client_id, - "response_type": "code", - "redirect_uri": self.redirect_uri, - "scope": " ".join(self.scopes), - "state": state, - "code_challenge": code_challenge, - "code_challenge_method": "S256", - } - - if extra_params: - params.update(extra_params) - - return f"{self.authorization_endpoint}?{urlencode(params)}" - - async def exchange_authorization_code( - self, - code: str, - code_verifier: str, - ) -> dict: - """ - Exchange authorization code for tokens. - - Args: - code: Authorization code from OAuth callback - code_verifier: PKCE code verifier - - Returns: - Token response dictionary with keys: - - access_token: Access token - - refresh_token: Refresh token (if offline_access scope requested) - - id_token: ID token (JWT) - - expires_in: Access token lifetime in seconds - - refresh_expires_in: Refresh token lifetime in seconds (optional) - - token_type: Token type (Bearer) - - Raises: - httpx.HTTPError: If token exchange fails - """ - if not self.token_endpoint: - await self.discover() - - if not self.token_endpoint: - raise RuntimeError("Token endpoint not discovered") - - logger.debug( - f"Exchanging authorization code for tokens at {self.token_endpoint}" - ) - - client = await self._get_http_client() - response = await client.post( - self.token_endpoint, - data={ - "grant_type": "authorization_code", - "code": code, - "redirect_uri": self.redirect_uri, - "code_verifier": code_verifier, - }, - auth=(self.client_id, self.client_secret), - ) - - response.raise_for_status() - token_data = response.json() - - logger.info("✓ Successfully exchanged authorization code for tokens") - - if "refresh_token" in token_data: - logger.info(" Received refresh token (offline_access granted)") - - return token_data - - async def refresh_access_token(self, refresh_token: str) -> dict: - """ - Refresh access token using refresh token. - - Args: - refresh_token: Refresh token - - Returns: - Token response dictionary (same format as exchange_authorization_code) - - Raises: - httpx.HTTPError: If token refresh fails - """ - if not self.token_endpoint: - await self.discover() - - if not self.token_endpoint: - raise RuntimeError("Token endpoint not discovered") - - logger.debug("Refreshing access token") - - client = await self._get_http_client() - response = await client.post( - self.token_endpoint, - data={ - "grant_type": "refresh_token", - "refresh_token": refresh_token, - }, - auth=(self.client_id, self.client_secret), - ) - - response.raise_for_status() - token_data = response.json() - - logger.debug("✓ Successfully refreshed access token") - - return token_data - - async def get_userinfo(self, access_token: str) -> dict: - """ - Get user information using access token. - - Args: - access_token: Access token - - Returns: - Userinfo response dictionary with claims like: - - sub: Subject (user ID) - - name: Full name - - preferred_username: Username - - email: Email address - - email_verified: Email verification status - - Raises: - httpx.HTTPError: If userinfo request fails - """ - if not self.userinfo_endpoint: - await self.discover() - - if not self.userinfo_endpoint: - raise RuntimeError("Userinfo endpoint not discovered") - - logger.debug("Fetching user info") - - client = await self._get_http_client() - response = await client.get( - self.userinfo_endpoint, - headers={"Authorization": f"Bearer {access_token}"}, - ) - - response.raise_for_status() - userinfo = response.json() - - logger.debug(f"✓ Retrieved user info for subject: {userinfo.get('sub')}") - - return userinfo - - async def get_service_account_token(self, scopes: list[str] | None = None) -> dict: - """ - Get a service account token using client_credentials grant. - - ⚠️ **WARNING: DO NOT USE FOR DIRECT API ACCESS IN OAUTH MODE** ⚠️ - - This method creates a service account user in Nextcloud which VIOLATES - OAuth "act on-behalf-of" principles. Using this token directly for API - access will: - - Create a Nextcloud user: `service-account-{client_id}` - - Attribute all actions to service account instead of real user - - Break audit trail and user attribution - - Create stateful server identity in Nextcloud - - Violate OAuth security model - - **Valid Use Case**: ONLY as subject_token for RFC 8693 token exchange - (ADR-002 Tier 2) where it's immediately exchanged for a user token. - - **Invalid Use Case**: Direct API access with this token (ADR-002 rejected - this as "Tier 1" - see docs/ADR-002-vector-sync-authentication.md). - - **Alternative**: Use token exchange (impersonation/delegation) for - background operations, or use BasicAuth mode if truly need service account. - - This requires the client to have serviceAccountsEnabled=true in provider. - - Args: - scopes: Optional list of scopes to request (default: openid profile email) - - Returns: - Token response dictionary with: - - access_token: Service account access token - - token_type: Bearer - - expires_in: Token lifetime in seconds - - scope: Granted scopes - - Raises: - httpx.HTTPError: If token request fails - - See Also: - - ADR-002 "Will Not Implement" section for detailed critique - - exchange_token_for_user() for proper token exchange usage - """ - if not self.token_endpoint: - await self.discover() - - if not self.token_endpoint: - raise RuntimeError("Token endpoint not discovered") - - # Default scopes - if scopes is None: - scopes = ["openid", "profile", "email"] - - scope_str = " ".join(scopes) - - logger.info(f"Requesting service account token with scopes: {scope_str}") - - client = await self._get_http_client() - response = await client.post( - self.token_endpoint, - data={ - "grant_type": "client_credentials", - "scope": scope_str, - }, - auth=(self.client_id, self.client_secret), - ) - - response.raise_for_status() - token_data = response.json() - - logger.info("✓ Service account token acquired") - - return token_data - - async def exchange_token_for_user( - self, - subject_token: str, - target_user_id: str | None = None, - audience: str | None = None, - scopes: list[str] | None = None, - ) -> dict: - """ - Exchange a token for a user-scoped token using RFC 8693 Token Exchange. - - This allows the MCP server (with a service account token) to obtain - user-scoped access tokens for background operations without needing - refresh tokens. - - Args: - subject_token: The token being exchanged (service account or user token) - target_user_id: Optional user ID to impersonate/exchange for - audience: Optional target audience (client ID) - scopes: Optional list of scopes for the new token - - Returns: - Token response dictionary with: - - access_token: User-scoped access token - - issued_token_type: urn:ietf:params:oauth:token-type:access_token - - token_type: Bearer - - expires_in: Token lifetime in seconds - - Raises: - httpx.HTTPError: If token exchange fails (403 if not authorized) - - Example: - # Get service account token - service_token = await client.get_service_account_token() - - # Exchange for user-scoped token - user_token = await client.exchange_token_for_user( - subject_token=service_token["access_token"], - target_user_id="admin", # Username or sub claim - audience="nextcloud", - scopes=["notes:read", "files:read"] - ) - - Note: - This implements BOTH ADR-002 tiers: - - **Tier 2 (Delegation - Recommended)**: When target_user_id is None - - Uses Keycloak Standard V2 (production-ready) - - Service account maintains its identity (sub claim unchanged) - - No special permissions required - - **Tier 1 (Impersonation - Advanced)**: When target_user_id is provided - - Requires Keycloak Legacy V1 (--features=preview) - - Subject claim changes to target user - - Requires impersonation role granted via Keycloak CLI: - ``` - kcadm.sh add-roles -r \ - --uusername service-account- \ - --cclientid realm-management \ - --rolename impersonation - ``` - - Both tiers require: - - Client has token.exchange.grant.enabled=true - - Client has serviceAccountsEnabled=true - """ - if not self.token_endpoint: - await self.discover() - - if not self.token_endpoint: - raise RuntimeError("Token endpoint not discovered") - - # Build token exchange request - data = { - "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange", - "subject_token": subject_token, - "subject_token_type": "urn:ietf:params:oauth:token-type:access_token", - "requested_token_type": "urn:ietf:params:oauth:token-type:access_token", - } - - # Add optional parameters - if audience: - data["audience"] = audience - - if scopes: - data["scope"] = " ".join(scopes) - - if target_user_id: - # Tier 1: Impersonation (Legacy V1) - # Use requested_subject for user impersonation - data["requested_subject"] = target_user_id - logger.info( - f"Exchanging token with impersonation (Tier 1): target_user={target_user_id}" - ) - else: - # Tier 2: Delegation (Standard V2) - logger.info( - "Exchanging token with delegation (Tier 2): service account identity preserved" - ) - - client = await self._get_http_client() - response = await client.post( - self.token_endpoint, - data=data, - auth=(self.client_id, self.client_secret), - ) - - if response.status_code != 200: - error_data = ( - response.json() - if response.headers.get("content-type", "").startswith( - "application/json" - ) - else {"error": "unknown"} - ) - logger.error(f"Token exchange failed: {response.status_code}") - logger.error(f"Error response: {error_data}") - - response.raise_for_status() - token_data = response.json() - - logger.info( - f"✓ Token exchange successful, issued_token_type: {token_data.get('issued_token_type')}" - ) - - return token_data - - async def check_token_exchange_support(self) -> bool: - """ - Check if Keycloak supports RFC 8693 token exchange. - - Returns: - True if token exchange is supported - - Note: - This is ADR-002 Tier 2. Most Keycloak installations don't - have token exchange enabled by default. - """ - if not self.token_endpoint: - await self.discover() - - # Try to get discovery document and check for token exchange grant - discovery_url = ( - f"{self.keycloak_url}/realms/{self.realm}/.well-known/openid-configuration" - ) - - try: - client = await self._get_http_client() - response = await client.get(discovery_url) - response.raise_for_status() - discovery_data = response.json() - - grant_types = discovery_data.get("grant_types_supported", []) - supported = "urn:ietf:params:oauth:grant-type:token-exchange" in grant_types - - if supported: - logger.info("✓ Token exchange (RFC 8693) is supported") - else: - logger.info("Token exchange (RFC 8693) is not supported") - - return supported - - except Exception as e: - logger.warning(f"Failed to check token exchange support: {e}") - return False - - -__all__ = ["KeycloakOAuthClient"] diff --git a/nextcloud_mcp_server/auth/provisioning_decorator.py b/nextcloud_mcp_server/auth/provisioning_decorator.py index d2d98b66..585b6ff2 100644 --- a/nextcloud_mcp_server/auth/provisioning_decorator.py +++ b/nextcloud_mcp_server/auth/provisioning_decorator.py @@ -15,7 +15,6 @@ from mcp.shared.exceptions import McpError from mcp.types import ErrorData from nextcloud_mcp_server.auth.storage import RefreshTokenStorage -from nextcloud_mcp_server.config import get_settings logger = logging.getLogger(__name__) @@ -65,14 +64,6 @@ def require_provisioning(func: Callable) -> Callable: logger.debug("BasicAuth mode detected - skipping provisioning check") return await func(*args, **kwargs) - # Check if we're in token exchange mode - if so, skip provisioning check - # In token exchange mode, tokens are exchanged per-request (no stored refresh tokens) - settings = get_settings() - if hasattr(lifespan_ctx, "nextcloud_host") and settings.enable_token_exchange: - # Token exchange mode - per-request exchange, no provisioning needed - logger.debug("Token exchange mode detected - skipping provisioning check") - return await func(*args, **kwargs) - # Offline access mode - check if user has completed Flow 2 provisioning # Get user_id from authorization token user_id = None diff --git a/nextcloud_mcp_server/auth/token_broker.py b/nextcloud_mcp_server/auth/token_broker.py index b14478bc..7e11c198 100644 --- a/nextcloud_mcp_server/auth/token_broker.py +++ b/nextcloud_mcp_server/auth/token_broker.py @@ -11,7 +11,7 @@ The Token Broker provides: - Short-lived token caching (5-minute TTL) - Master refresh token rotation - Audience-specific token validation -- Session vs background token separation (RFC 8693) +- Background token management """ import logging @@ -23,7 +23,6 @@ import httpx import jwt from nextcloud_mcp_server.auth.storage import RefreshTokenStorage -from nextcloud_mcp_server.auth.token_exchange import exchange_token_for_delegation from ..http import nextcloud_httpx_client @@ -219,55 +218,6 @@ class TokenBrokerService: await self.cache.invalidate(user_id) return None - async def get_session_token( - self, - flow1_token: str, - required_scopes: list[str], - requested_audience: str = "nextcloud", - ) -> Optional[str]: - """ - Get ephemeral token for MCP session operations (on-demand). - - This implements the correct Progressive Consent pattern where: - 1. Client provides Flow 1 token (aud: "mcp-server") - 2. Server exchanges it for ephemeral Nextcloud token - 3. Token is NOT stored, only used for current operation - - Key properties: - - On-demand generation during tool execution - - Ephemeral (not stored, discarded after use) - - Limited scopes (only what tool needs) - - Short-lived (5 minutes) - - Args: - flow1_token: The MCP session token (aud: "mcp-server") - required_scopes: Minimal scopes needed for this operation - requested_audience: Target audience (usually "nextcloud") - - Returns: - Ephemeral Nextcloud access token or None if exchange fails - """ - try: - # Perform RFC 8693 token exchange - delegated_token, expires_in = await exchange_token_for_delegation( - flow1_token=flow1_token, - requested_scopes=required_scopes, - requested_audience=requested_audience, - ) - - # NOTE: We intentionally do NOT cache session tokens - # They are ephemeral and should be discarded after use - logger.info( - f"Generated ephemeral session token with scopes: {required_scopes}, " - f"expires in {expires_in}s" - ) - - return delegated_token - - except Exception as e: - logger.error(f"Failed to get session token: {e}") - return None - async def get_background_token( self, user_id: str, required_scopes: list[str] ) -> Optional[str]: diff --git a/nextcloud_mcp_server/auth/token_exchange.py b/nextcloud_mcp_server/auth/token_exchange.py deleted file mode 100644 index 52be1ab7..00000000 --- a/nextcloud_mcp_server/auth/token_exchange.py +++ /dev/null @@ -1,596 +0,0 @@ -"""RFC 8693 Token Exchange implementation for ADR-004 Progressive Consent. - -This module implements the token exchange pattern to convert Flow 1 MCP tokens -(aud: "mcp-server") into ephemeral delegated Nextcloud tokens (aud: "nextcloud") -for session operations. - -Key Properties: -- On-demand generation during tool execution -- Ephemeral tokens (NOT stored, discarded after use) -- Limited scopes (only what tool needs) -- Short-lived (5 minutes default) -""" - -import logging -import time -from typing import Any, Dict, Optional, Tuple -from urllib.parse import urljoin - -import httpx -import jwt - -from ..config import get_settings -from ..http import nextcloud_httpx_client -from .storage import RefreshTokenStorage - -logger = logging.getLogger(__name__) - - -class TokenExchangeService: - """Implements RFC 8693 OAuth 2.0 Token Exchange.""" - - # RFC 8693 Grant Type - TOKEN_EXCHANGE_GRANT = "urn:ietf:params:oauth:grant-type:token-exchange" - - # RFC 8693 Token Type Identifiers - TOKEN_TYPE_ACCESS_TOKEN = "urn:ietf:params:oauth:token-type:access_token" - TOKEN_TYPE_JWT = "urn:ietf:params:oauth:token-type:jwt" - TOKEN_TYPE_ID_TOKEN = "urn:ietf:params:oauth:token-type:id_token" - - def __init__( - self, - oidc_discovery_url: Optional[str] = None, - client_id: Optional[str] = None, - client_secret: Optional[str] = None, - nextcloud_host: Optional[str] = None, - ): - """Initialize token exchange service. - - Args: - oidc_discovery_url: OIDC discovery endpoint URL - client_id: OAuth client ID for token exchange - client_secret: OAuth client secret - nextcloud_host: Nextcloud instance URL - """ - settings = get_settings() - self.oidc_discovery_url = oidc_discovery_url or settings.oidc_discovery_url - self.client_id = client_id or settings.oidc_client_id - self.client_secret = client_secret or settings.oidc_client_secret - self.nextcloud_host = nextcloud_host or settings.nextcloud_host - - self._token_endpoint: Optional[str] = None - self._jwks_uri: Optional[str] = None - self._discovery_cache: Optional[Dict[str, Any]] = None - self._discovery_cache_time: float = 0 - self._discovery_cache_ttl: float = 3600 # 1 hour - - # Storage for Progressive Consent (refresh tokens) - only needed for delegation - # NOT needed for pure RFC 8693 exchange (MCP tools) - self.storage: Optional[RefreshTokenStorage] = None - - # Create HTTP client - self.http_client = nextcloud_httpx_client( - timeout=30.0, - follow_redirects=True, - ) - - async def __aenter__(self): - """Async context manager entry.""" - if self.storage: - await self.storage.initialize() - return self - - async def __aexit__(self, exc_type, exc_val, exc_tb): - """Async context manager exit.""" - await self.close() - - async def close(self): - """Close HTTP client and storage.""" - await self.http_client.aclose() - # RefreshTokenStorage doesn't have a close method - - async def _ensure_storage(self): - """Lazily initialize storage for Progressive Consent operations. - - Only needed for delegation operations that use refresh tokens. - NOT needed for pure RFC 8693 exchange (MCP tools). - """ - if self.storage is None: - self.storage = RefreshTokenStorage.from_env() - await self.storage.initialize() - - async def _discover_endpoints(self) -> Dict[str, Any]: - """Discover OIDC endpoints from discovery URL. - - Returns: - Discovery document containing endpoint URLs - """ - # Check cache - if ( - self._discovery_cache - and (time.time() - self._discovery_cache_time) < self._discovery_cache_ttl - ): - return self._discovery_cache - - if not self.oidc_discovery_url: - # Fallback to Nextcloud OIDC if no discovery URL - self.oidc_discovery_url = urljoin( - self.nextcloud_host, # type: ignore[arg-type] - "/.well-known/openid-configuration", - ) - - try: - response = await self.http_client.get(self.oidc_discovery_url) - response.raise_for_status() - - self._discovery_cache = response.json() - self._discovery_cache_time = time.time() - - # Cache frequently used endpoints - self._token_endpoint = self._discovery_cache.get("token_endpoint") - self._jwks_uri = self._discovery_cache.get("jwks_uri") - - return self._discovery_cache - - except Exception as e: - logger.error(f"Failed to discover OIDC endpoints: {e}") - raise - - async def exchange_token_for_delegation( - self, - flow1_token: str, - requested_scopes: list[str], - requested_audience: str = "nextcloud", - ) -> Tuple[str, int]: - """Exchange Flow 1 MCP token for delegated Nextcloud token. - - This implements RFC 8693 Token Exchange for on-behalf-of delegation. - - Args: - flow1_token: The MCP session token (aud: "mcp-server") - requested_scopes: Scopes needed for this operation - requested_audience: Target audience (usually "nextcloud") - - Returns: - Tuple of (delegated_token, expires_in) - - Raises: - ValueError: If token validation fails - RuntimeError: If provisioning not completed or exchange fails - """ - # 1. Validate Flow 1 token audience - await self._validate_flow1_token(flow1_token) - - # 2. Extract user ID from token - user_id = self._extract_user_id(flow1_token) - - # 3. Check user has provisioned Nextcloud access (Flow 2) - if not await self._check_provisioning(user_id): - raise RuntimeError( - "Nextcloud access not provisioned. " - "User must complete Flow 2 provisioning first." - ) - - # 4. Get stored refresh token for user (from Flow 2) - refresh_token = await self._get_user_refresh_token(user_id) - if not refresh_token: - raise RuntimeError( - "No refresh token found. User must complete provisioning." - ) - - # 5. Perform token exchange with IdP - delegated_token, expires_in = await self._perform_token_exchange( - subject_token=flow1_token, - refresh_token=refresh_token, - requested_scopes=requested_scopes, - requested_audience=requested_audience, - ) - - # 6. Log the exchange for audit trail - logger.info( - f"Token exchange completed for user {user_id}: " - f"scopes={requested_scopes}, audience={requested_audience}, " - f"expires_in={expires_in}s" - ) - - return delegated_token, expires_in - - async def exchange_token_for_audience( - self, - subject_token: str, - requested_audience: str = "nextcloud", - requested_scopes: list[str] | None = None, - ) -> Tuple[str, int]: - """ - Pure RFC 8693 token exchange (no refresh tokens required). - - This implements stateless per-request token exchange where: - 1. Client token has aud: (e.g., "nextcloud-mcp-server") - 2. Exchange for token with aud: "nextcloud" (for API access) - 3. NO refresh tokens or provisioning required - - Use case: All MCP tool calls (request-time operations). - NOT for background jobs (which use refresh tokens separately). - - Args: - subject_token: Token being exchanged (from MCP client) - requested_audience: Target audience (usually "nextcloud") - requested_scopes: Optional scopes (may not be supported by all IdPs) - - Returns: - Tuple of (access_token, expires_in) - - Raises: - ValueError: If token validation fails - RuntimeError: If exchange fails - """ - # 1. Validate subject token (accepts both "mcp-server" and client_id) - await self._validate_flow1_token(subject_token) - - # 2. Extract user ID for logging - user_id = self._extract_user_id(subject_token) - - # 3. Discover token endpoint - discovery = await self._discover_endpoints() - token_endpoint = discovery.get("token_endpoint") - - if not token_endpoint: - raise RuntimeError("No token endpoint found in discovery") - - # 4. Build pure RFC 8693 exchange request (subject_token ONLY) - data = { - "grant_type": self.TOKEN_EXCHANGE_GRANT, - "subject_token": subject_token, - "subject_token_type": self.TOKEN_TYPE_ACCESS_TOKEN, - "requested_token_type": self.TOKEN_TYPE_ACCESS_TOKEN, - "audience": requested_audience, - } - - # Add scopes if provided (may not be supported by all providers) - if requested_scopes: - data["scope"] = " ".join(requested_scopes) - - # Add client credentials - if self.client_id and self.client_secret: - data["client_id"] = self.client_id - data["client_secret"] = self.client_secret - - try: - # Perform exchange - logger.debug(f"Exchanging token for audience={requested_audience}") - response = await self.http_client.post( - token_endpoint, - data=data, - headers={"Content-Type": "application/x-www-form-urlencoded"}, - ) - response.raise_for_status() - result = response.json() - - access_token = result.get("access_token") - expires_in = result.get("expires_in", 300) - - if not access_token: - raise RuntimeError("No access token in exchange response") - - logger.info( - f"Pure RFC 8693 token exchange successful for user {user_id}: " - f"audience={requested_audience}, expires_in={expires_in}s" - ) - - return access_token, expires_in - - except httpx.HTTPStatusError as e: - logger.error(f"Token exchange failed: {e.response.text}") - raise RuntimeError(f"Token exchange failed: {e}") - except Exception as e: - logger.error(f"Token exchange error: {e}") - raise - - async def _validate_flow1_token(self, token: str): - """Validate that token has correct audience for MCP server. - - Accepts either: - - "mcp-server" (Progressive Consent legacy) - - self.client_id (external IdP, e.g., "nextcloud-mcp-server") - - Args: - token: JWT token to validate - - Raises: - ValueError: If token is invalid or has wrong audience - """ - try: - # Decode without verification first to check audience - # In production, should verify signature against JWKS - payload = jwt.decode(token, options={"verify_signature": False}) - - # Check audience - audience = payload.get("aud", []) - if isinstance(audience, str): - audience = [audience] - - # Accept either "mcp-server" (Progressive Consent) or client_id (external IdP) - valid_audiences = ["mcp-server"] - if self.client_id: - valid_audiences.append(self.client_id) - - if not any(aud in audience for aud in valid_audiences): - raise ValueError( - f"Invalid token audience. Expected one of {valid_audiences}, got {audience}" - ) - - # Check expiration - exp = payload.get("exp", 0) - if exp < time.time(): - raise ValueError("Token has expired") - - except jwt.DecodeError as e: - raise ValueError(f"Invalid JWT token: {e}") - - def _extract_user_id(self, token: str) -> str: - """Extract user ID from JWT token. - - Args: - token: JWT token - - Returns: - User ID from token - """ - try: - payload = jwt.decode(token, options={"verify_signature": False}) - - # Try standard claims in order of preference - user_id = ( - payload.get("sub") - or payload.get("preferred_username") - or payload.get("email") - or payload.get("name") - ) - - if not user_id: - raise ValueError("No user identifier in token") - - return user_id - - except jwt.DecodeError as e: - raise ValueError(f"Failed to extract user ID: {e}") - - async def _check_provisioning(self, user_id: str) -> bool: - """Check if user has completed Flow 2 provisioning. - - Args: - user_id: User identifier - - Returns: - True if provisioned, False otherwise - """ - await self._ensure_storage() - assert self.storage is not None # _ensure_storage() ensures this - token_data = await self.storage.get_refresh_token(user_id) - return token_data is not None - - async def _get_user_refresh_token(self, user_id: str) -> Optional[str]: - """Get stored refresh token for user from Flow 2 provisioning. - - Args: - user_id: User identifier - - Returns: - Refresh token if found, None otherwise - """ - await self._ensure_storage() - assert self.storage is not None # _ensure_storage() ensures this - token_data = await self.storage.get_refresh_token(user_id) - if token_data: - return token_data.get("refresh_token") - return None - - async def _perform_token_exchange( - self, - subject_token: str, - refresh_token: str, - requested_scopes: list[str], - requested_audience: str, - ) -> Tuple[str, int]: - """Perform RFC 8693 token exchange with IdP. - - Args: - subject_token: The token being exchanged (Flow 1 token) - refresh_token: User's stored refresh token for delegation - requested_scopes: Minimal scopes for this operation - requested_audience: Target audience - - Returns: - Tuple of (access_token, expires_in) - """ - # Discover token endpoint - discovery = await self._discover_endpoints() - token_endpoint = discovery.get("token_endpoint") - - if not token_endpoint: - raise RuntimeError("No token endpoint found in discovery") - - # Build token exchange request per RFC 8693 - data = { - # Token exchange grant type - "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange", - # The token we're exchanging (Flow 1 MCP token) - "subject_token": subject_token, - "subject_token_type": self.TOKEN_TYPE_ACCESS_TOKEN, - # Use refresh token as actor token (proves we have delegation rights) - "actor_token": refresh_token, - "actor_token_type": self.TOKEN_TYPE_ACCESS_TOKEN, - # Requested token properties - "requested_token_type": self.TOKEN_TYPE_ACCESS_TOKEN, - "audience": requested_audience, - "scope": " ".join(requested_scopes), - } - - # Add client credentials if configured - if self.client_id and self.client_secret: - data["client_id"] = self.client_id - data["client_secret"] = self.client_secret - - try: - # Attempt RFC 8693 token exchange - response = await self.http_client.post( - token_endpoint, - data=data, - headers={"Content-Type": "application/x-www-form-urlencoded"}, - ) - - if response.status_code == 400: - # Token exchange might not be supported, fall back to refresh grant - logger.info( - "Token exchange not supported, falling back to refresh grant" - ) - return await self._fallback_refresh_grant( - refresh_token=refresh_token, - requested_scopes=requested_scopes, - token_endpoint=token_endpoint, - ) - - response.raise_for_status() - result = response.json() - - access_token = result.get("access_token") - expires_in = result.get("expires_in", 300) # Default 5 minutes - - if not access_token: - raise RuntimeError("No access token in exchange response") - - return access_token, expires_in - - except httpx.HTTPStatusError as e: - logger.error(f"Token exchange failed: {e.response.text}") - raise RuntimeError(f"Token exchange failed: {e}") - except Exception as e: - logger.error(f"Token exchange error: {e}") - raise - - async def _fallback_refresh_grant( - self, refresh_token: str, requested_scopes: list[str], token_endpoint: str - ) -> Tuple[str, int]: - """Fallback to standard refresh token grant if token exchange not supported. - - This is less secure than token exchange but provides compatibility. - - Args: - refresh_token: User's stored refresh token - requested_scopes: Minimal scopes for this operation - token_endpoint: Token endpoint URL - - Returns: - Tuple of (access_token, expires_in) - """ - data = { - "grant_type": "refresh_token", - "refresh_token": refresh_token, - "scope": " ".join(requested_scopes), # Request minimal scopes - } - - # Add client credentials if configured - if self.client_id and self.client_secret: - data["client_id"] = self.client_id - data["client_secret"] = self.client_secret - - try: - response = await self.http_client.post( - token_endpoint, - data=data, - headers={"Content-Type": "application/x-www-form-urlencoded"}, - ) - response.raise_for_status() - - result = response.json() - - access_token = result.get("access_token") - expires_in = result.get("expires_in", 300) # Default 5 minutes - - if not access_token: - raise RuntimeError("No access token in refresh response") - - # Log that we're using fallback - logger.warning( - f"Using refresh grant fallback for token exchange. " - f"Scopes: {requested_scopes}" - ) - - return access_token, expires_in - - except httpx.HTTPStatusError as e: - logger.error(f"Refresh grant failed: {e.response.text}") - raise RuntimeError(f"Refresh grant failed: {e}") - except Exception as e: - logger.error(f"Refresh grant error: {e}") - raise - - -# Singleton instance -_token_exchange_service: Optional[TokenExchangeService] = None - - -async def get_token_exchange_service() -> TokenExchangeService: - """Get or create the singleton token exchange service. - - Note: Storage is initialized lazily only when needed for delegation operations. - Pure RFC 8693 exchange (MCP tools) doesn't require storage. - - Returns: - TokenExchangeService instance - """ - global _token_exchange_service - - if _token_exchange_service is None: - _token_exchange_service = TokenExchangeService() - # Storage is initialized lazily via _ensure_storage() when needed - - return _token_exchange_service - - -async def exchange_token_for_delegation( - flow1_token: str, requested_scopes: list[str], requested_audience: str = "nextcloud" -) -> Tuple[str, int]: - """Convenience function to exchange tokens (Progressive Consent with refresh tokens). - - NOTE: This is for background jobs only. For MCP tool calls, use exchange_token_for_audience(). - - Args: - flow1_token: The MCP session token (aud: "mcp-server") - requested_scopes: Scopes needed for this operation - requested_audience: Target audience (usually "nextcloud") - - Returns: - Tuple of (delegated_token, expires_in) - """ - service = await get_token_exchange_service() - return await service.exchange_token_for_delegation( - flow1_token=flow1_token, - requested_scopes=requested_scopes, - requested_audience=requested_audience, - ) - - -async def exchange_token_for_audience( - subject_token: str, - requested_audience: str = "nextcloud", - requested_scopes: list[str] | None = None, -) -> Tuple[str, int]: - """Convenience function for pure RFC 8693 token exchange (no refresh tokens). - - Use this for ALL MCP tool calls (request-time operations). - - Args: - subject_token: Token being exchanged (from MCP client) - requested_audience: Target audience (usually "nextcloud") - requested_scopes: Optional scopes (may not be supported by all IdPs) - - Returns: - Tuple of (access_token, expires_in) - """ - service = await get_token_exchange_service() - return await service.exchange_token_for_audience( - subject_token=subject_token, - requested_audience=requested_audience, - requested_scopes=requested_scopes, - ) diff --git a/nextcloud_mcp_server/auth/unified_verifier.py b/nextcloud_mcp_server/auth/unified_verifier.py index 41147f7a..cf4b814e 100644 --- a/nextcloud_mcp_server/auth/unified_verifier.py +++ b/nextcloud_mcp_server/auth/unified_verifier.py @@ -60,7 +60,7 @@ class UnifiedTokenVerifier(TokenVerifier): settings: Application settings containing OAuth configuration """ self.settings = settings - self.mode = "exchange" if settings.enable_token_exchange else "multi-audience" + self.mode = "multi-audience" # Common components for all modes self.http_client = nextcloud_httpx_client(timeout=10.0) diff --git a/nextcloud_mcp_server/config.py b/nextcloud_mcp_server/config.py index 6405c938..cc4aa50c 100644 --- a/nextcloud_mcp_server/config.py +++ b/nextcloud_mcp_server/config.py @@ -138,8 +138,7 @@ class Settings: # Deployment mode (ADR-021: explicit mode selection) # Optional: If not set, mode is auto-detected from other settings - # Valid values: single_user_basic, multi_user_basic, oauth_single_audience, - # oauth_token_exchange + # Valid values: single_user_basic, multi_user_basic, oauth_single_audience deployment_mode: str | None = None # OAuth/OIDC settings @@ -168,7 +167,6 @@ class Settings: userinfo_uri: str | None = None # Progressive Consent settings (always enabled - no flag needed) - enable_token_exchange: bool = False enable_offline_access: bool = False # Multi-user BasicAuth pass-through mode (ADR-019 interim solution) @@ -179,9 +177,6 @@ class Settings: # Login Flow v2 settings (ADR-022) enable_login_flow: bool = False - # Token exchange cache settings - token_exchange_cache_ttl: int = 300 # seconds (5 minutes default) - # Token and webhook storage settings # TOKEN_ENCRYPTION_KEY: Optional - Only required for OAuth token storage operations. # Webhook tracking works without encryption key. @@ -507,9 +502,6 @@ def get_settings() -> Settings: introspection_uri=os.getenv("INTROSPECTION_URI"), userinfo_uri=os.getenv("USERINFO_URI"), # Progressive Consent settings (always enabled) - enable_token_exchange=( - os.getenv("ENABLE_TOKEN_EXCHANGE", "false").lower() == "true" - ), enable_offline_access=enable_background_operations, # Smart dependency resolution # Multi-user BasicAuth pass-through mode enable_multi_user_basic_auth=( @@ -517,8 +509,6 @@ def get_settings() -> Settings: ), # Login Flow v2 settings (ADR-022) enable_login_flow=(os.getenv("ENABLE_LOGIN_FLOW", "false").lower() == "true"), - # Token exchange cache settings - token_exchange_cache_ttl=int(os.getenv("TOKEN_EXCHANGE_CACHE_TTL", "300")), # Token and webhook storage settings (encryption key optional for webhook-only usage) token_encryption_key=os.getenv("TOKEN_ENCRYPTION_KEY"), token_storage_db=os.getenv("TOKEN_STORAGE_DB", "/tmp/tokens.db"), diff --git a/nextcloud_mcp_server/config_validators.py b/nextcloud_mcp_server/config_validators.py index d425367f..d2e4ff8a 100644 --- a/nextcloud_mcp_server/config_validators.py +++ b/nextcloud_mcp_server/config_validators.py @@ -26,7 +26,6 @@ class AuthMode(Enum): SINGLE_USER_BASIC = "single_user_basic" MULTI_USER_BASIC = "multi_user_basic" OAUTH_SINGLE_AUDIENCE = "oauth_single" - OAUTH_TOKEN_EXCHANGE = "oauth_exchange" @dataclass @@ -66,7 +65,6 @@ MODE_REQUIREMENTS: dict[AuthMode, ModeRequirements] = { ], forbidden=[ "enable_multi_user_basic_auth", - "enable_token_exchange", "oidc_client_id", "oidc_client_secret", ], @@ -100,7 +98,6 @@ MODE_REQUIREMENTS: dict[AuthMode, ModeRequirements] = { forbidden=[ "nextcloud_username", "nextcloud_password", - "enable_token_exchange", ], conditional={ "enable_offline_access": [ @@ -141,7 +138,6 @@ MODE_REQUIREMENTS: dict[AuthMode, ModeRequirements] = { forbidden=[ "nextcloud_username", "nextcloud_password", - "enable_token_exchange", "enable_multi_user_basic_auth", ], conditional={ @@ -157,46 +153,6 @@ MODE_REQUIREMENTS: dict[AuthMode, ModeRequirements] = { "Tokens work for both MCP server and Nextcloud APIs (pass-through). " "Uses Dynamic Client Registration if credentials not provided.", ), - AuthMode.OAUTH_TOKEN_EXCHANGE: ModeRequirements( - required=["nextcloud_host", "enable_token_exchange"], - optional=[ - # OAuth credentials - "oidc_client_id", - "oidc_client_secret", - "oidc_discovery_url", - # Token exchange settings - "token_exchange_cache_ttl", - # Offline access - "enable_offline_access", - "token_encryption_key", - "token_storage_db", - # Vector sync - "vector_sync_enabled", - "qdrant_url", - "qdrant_location", - "ollama_base_url", - "ollama_embedding_model", - "openai_api_key", - "openai_embedding_model", - ], - forbidden=[ - "nextcloud_username", - "nextcloud_password", - "enable_multi_user_basic_auth", - ], - conditional={ - "enable_offline_access": [ - "token_encryption_key", - "token_storage_db", - ], - # Note: vector_sync_enabled (now ENABLE_SEMANTIC_SEARCH) automatically - # enables background operations in multi-user modes. No explicit - # enable_offline_access setting required. - }, - description="OAuth multi-user deployment with token exchange (RFC 8693). " - "MCP tokens are separate from Nextcloud tokens. " - "Server exchanges MCP token for Nextcloud token on each request.", - ), } @@ -205,10 +161,9 @@ def detect_auth_mode(settings: Settings) -> AuthMode: Mode detection priority (ADR-021): 0. Explicit MCP_DEPLOYMENT_MODE (if set) - NEW in ADR-021 - 1. Token exchange (most specific OAuth mode) - 2. Multi-user BasicAuth - 3. Single-user BasicAuth - 4. OAuth single-audience (default OAuth mode) + 1. Multi-user BasicAuth + 2. Single-user BasicAuth + 3. OAuth single-audience (default OAuth mode) Args: settings: Application settings @@ -231,7 +186,6 @@ def detect_auth_mode(settings: Settings) -> AuthMode: "single_user_basic": AuthMode.SINGLE_USER_BASIC, "multi_user_basic": AuthMode.MULTI_USER_BASIC, "oauth_single_audience": AuthMode.OAUTH_SINGLE_AUDIENCE, - "oauth_token_exchange": AuthMode.OAUTH_TOKEN_EXCHANGE, } if mode_str not in mode_map: @@ -246,10 +200,6 @@ def detect_auth_mode(settings: Settings) -> AuthMode: return explicit_mode # Auto-detection (existing behavior) - # Check for token exchange (most specific OAuth mode) - if settings.enable_token_exchange: - return AuthMode.OAUTH_TOKEN_EXCHANGE - # Check for multi-user BasicAuth if settings.enable_multi_user_basic_auth: return AuthMode.MULTI_USER_BASIC @@ -351,10 +301,7 @@ def validate_configuration(settings: Settings) -> tuple[AuthMode, list[str]]: f"{settings.nextcloud_host}" ) - if mode in [ - AuthMode.OAUTH_SINGLE_AUDIENCE, - AuthMode.OAUTH_TOKEN_EXCHANGE, - ]: + if mode == AuthMode.OAUTH_SINGLE_AUDIENCE: # If OAuth credentials not provided, DCR must be available # (This is a runtime check, not a config check, so we just warn) if not settings.oidc_client_id or not settings.oidc_client_secret: diff --git a/nextcloud_mcp_server/context.py b/nextcloud_mcp_server/context.py index 6a380ea7..c94b15be 100644 --- a/nextcloud_mcp_server/context.py +++ b/nextcloud_mcp_server/context.py @@ -5,10 +5,7 @@ import logging from httpx import BasicAuth from mcp.server.fastmcp import Context -from nextcloud_mcp_server.auth.context_helper import ( - get_client_from_context, - get_session_client_from_context, -) +from nextcloud_mcp_server.auth.context_helper import get_client_from_context from nextcloud_mcp_server.auth.scope_authorization import ProvisioningRequiredError from nextcloud_mcp_server.auth.storage import get_shared_storage from nextcloud_mcp_server.client import NextcloudClient @@ -24,18 +21,9 @@ async def get_client(ctx: Context) -> NextcloudClient: Supports the following deployment modes: 1. BasicAuth mode: Returns shared client from lifespan context - - 2. OAuth mode: - a. Multi-audience mode (ENABLE_TOKEN_EXCHANGE=false, default): - Token already contains both MCP and Nextcloud audiences - use directly - b. Token exchange mode (ENABLE_TOKEN_EXCHANGE=true): - Exchange MCP token for Nextcloud token via RFC 8693 - - SECURITY: Token passthrough has been REMOVED. All OAuth modes validate - proper token audiences per MCP Security Best Practices specification. - - Note: Nextcloud doesn't support OAuth scopes natively. Scopes are enforced - by the MCP server via @require_scopes decorator, not by the IdP. + 2. Login Flow v2: OAuth for MCP session, app password for Nextcloud API + 3. Multi-user BasicAuth: Credentials passed through from request headers + 4. OAuth multi-audience: Token contains both MCP and Nextcloud audiences This function automatically detects the authentication mode by checking the type of the lifespan context. @@ -74,20 +62,11 @@ async def get_client(ctx: Context) -> NextcloudClient: if hasattr(lifespan_ctx, "client"): return lifespan_ctx.client - # OAuth mode (has 'nextcloud_host' attribute) + # OAuth multi-audience mode (has 'nextcloud_host' attribute) if hasattr(lifespan_ctx, "nextcloud_host"): - if settings.enable_token_exchange: - # Mode 2: Exchange MCP token for Nextcloud token - # Token was validated to have MCP audience in UnifiedTokenVerifier - # Now exchange it for Nextcloud audience - return await get_session_client_from_context( - ctx, lifespan_ctx.nextcloud_host - ) - else: - # Mode 1: Multi-audience token - use directly - # Token was validated to have MCP audience in UnifiedTokenVerifier - # Nextcloud will independently validate its own audience when receiving API calls - return get_client_from_context(ctx, lifespan_ctx.nextcloud_host) + # Token was validated to have MCP audience in UnifiedTokenVerifier + # Nextcloud will independently validate its own audience when receiving API calls + return get_client_from_context(ctx, lifespan_ctx.nextcloud_host) # Unknown context type raise AttributeError( diff --git a/nextcloud_mcp_server/observability/metrics.py b/nextcloud_mcp_server/observability/metrics.py index fe19c372..587336cb 100644 --- a/nextcloud_mcp_server/observability/metrics.py +++ b/nextcloud_mcp_server/observability/metrics.py @@ -125,12 +125,6 @@ oauth_token_validations_total = Counter( ["method", "result"], # method: introspect | jwt; result: valid | invalid | error ) -oauth_token_exchange_total = Counter( - "mcp_oauth_token_exchange_total", - "Total OAuth token exchange operations (RFC 8693)", - ["status"], # status: success | error -) - oauth_token_cache_hits_total = Counter( "mcp_oauth_token_cache_hits_total", "Total OAuth token cache lookups", diff --git a/nextcloud_mcp_server/observability/tracing.py b/nextcloud_mcp_server/observability/tracing.py index 2910cdd1..4904ee69 100644 --- a/nextcloud_mcp_server/observability/tracing.py +++ b/nextcloud_mcp_server/observability/tracing.py @@ -232,7 +232,7 @@ def trace_oauth_operation(operation: str, details: dict[str, Any] | None = None) pass Args: - operation: OAuth operation name (e.g., "token.validate", "token.exchange") + operation: OAuth operation name (e.g., "token.validate", "token.refresh") details: Optional operation details (sensitive data will be sanitized) Returns: diff --git a/tests/conftest.py b/tests/conftest.py index e08a9e54..74ac72ba 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,5 +1,4 @@ import base64 -import hashlib import json import logging import os @@ -2868,433 +2867,8 @@ async def test_user_in_group(nc_client: NextcloudClient, test_user, test_group): # =========================================================================================== -# Keycloak External IdP OAuth Fixtures -# =========================================================================================== - - -@pytest.fixture(scope="session") -async def keycloak_oauth_client_credentials(anyio_backend, oauth_callback_server): - """ - Fixture to obtain Keycloak OAuth client credentials for external IdP testing. - - Uses pre-configured client from keycloak/realm-export.json (no DCR needed). - The client (nextcloud-mcp-server) is already configured with: - - serviceAccountsEnabled=true - - token.exchange.grant.enabled=true - - client.token.exchange.standard.enabled=true - - Returns: - Tuple of (client_id, client_secret, callback_url, token_endpoint, authorization_endpoint) - """ - # Get Keycloak configuration from environment - keycloak_discovery_url = os.getenv( - "OIDC_DISCOVERY_URL", - "http://localhost:8888/realms/nextcloud-mcp/.well-known/openid-configuration", - ) - client_id = os.getenv("OIDC_CLIENT_ID", "nextcloud-mcp-server") - client_secret = os.getenv("OIDC_CLIENT_SECRET", "mcp-secret-change-in-production") - - if not all([keycloak_discovery_url, client_id, client_secret]): - pytest.skip( - "Keycloak OAuth requires OIDC_DISCOVERY_URL, OIDC_CLIENT_ID, OIDC_CLIENT_SECRET" - ) - - # Get callback URL from the real callback server - auth_states, callback_url = oauth_callback_server - - logger.info("Setting up Keycloak external IdP OAuth client credentials...") - logger.info(f"Using Keycloak discovery URL: {keycloak_discovery_url}") - logger.info(f"Using static client credentials: {client_id}") - logger.info(f"Using real callback server at: {callback_url}") - - async with httpx.AsyncClient(timeout=30.0) as http_client: - # OIDC Discovery - discovery_response = await http_client.get(keycloak_discovery_url) - discovery_response.raise_for_status() - oidc_config = discovery_response.json() - - token_endpoint = oidc_config.get("token_endpoint") - authorization_endpoint = oidc_config.get("authorization_endpoint") - - if not token_endpoint or not authorization_endpoint: - raise ValueError( - "Keycloak OIDC discovery missing required endpoints (token_endpoint or authorization_endpoint)" - ) - - logger.info(f"✓ Discovered token endpoint: {token_endpoint}") - logger.info(f"✓ Discovered authorization endpoint: {authorization_endpoint}") - - yield ( - client_id, - client_secret, - callback_url, - token_endpoint, - authorization_endpoint, - ) - - # No cleanup needed - client is pre-configured in realm export - - -async def _get_keycloak_oauth_token( - browser, - keycloak_oauth_client_credentials, - oauth_callback_server, - scopes: str, - username: str = "admin", - password: str = "admin", -) -> str: - """ - Helper function to obtain OAuth token from Keycloak using Playwright. - - Args: - browser: Playwright browser instance - keycloak_oauth_client_credentials: Tuple of Keycloak OAuth client credentials - oauth_callback_server: OAuth callback server fixture - scopes: Space-separated list of scopes - username: Keycloak username (default: admin) - password: Keycloak password (default: admin) - - Returns: - OAuth access token string from Keycloak - """ - - # Get auth_states dict from callback server - auth_states, _ = oauth_callback_server - - # Unpack Keycloak client credentials - client_id, client_secret, callback_url, token_endpoint, authorization_endpoint = ( - keycloak_oauth_client_credentials - ) - - logger.info(f"Starting Playwright-based Keycloak OAuth flow with scopes: {scopes}") - logger.info(f"Using Keycloak client: {client_id}") - logger.info(f"Using real callback server at: {callback_url}") - logger.info(f"Authenticating as Keycloak user: {username}") - - # Generate unique state parameter for this OAuth flow - state = secrets.token_urlsafe(32) - logger.debug(f"Generated state: {state[:16]}...") - - # Generate PKCE parameters (required by Keycloak client configuration) - code_verifier = secrets.token_urlsafe(64) # 86 chars base64url - code_challenge = ( - base64.urlsafe_b64encode(hashlib.sha256(code_verifier.encode()).digest()) - .decode() - .rstrip("=") - ) - logger.debug(f"Generated PKCE code_challenge: {code_challenge[:20]}...") - - # URL-encode scopes - scopes_encoded = quote(scopes, safe="") - - # Construct authorization URL with state, scopes, and PKCE parameters - auth_url = ( - f"{authorization_endpoint}?" - f"response_type=code&" - f"client_id={client_id}&" - f"redirect_uri={quote(callback_url, safe='')}&" - f"state={state}&" - f"scope={scopes_encoded}&" - f"code_challenge={code_challenge}&" - f"code_challenge_method=S256" - ) - - logger.info(f"Authorization URL: {auth_url[:100]}...") - - # Create browser context and page - context = await browser.new_context() - page = await context.new_page() - - try: - # Navigate to Keycloak authorization endpoint - logger.info("Navigating to Keycloak authorization endpoint...") - await page.goto(auth_url, wait_until="networkidle", timeout=30000) - - # Handle Keycloak login page - # Keycloak uses input#username and input#password (different from Nextcloud) - logger.info(f"Filling Keycloak login credentials for {username}...") - await page.wait_for_selector("input#username", timeout=10000) - await page.fill("input#username", username) - await page.fill("input#password", password) - - logger.info("Submitting Keycloak login form...") - # Submit the form and wait for navigation - # Use JavaScript to submit the form directly (more reliable than clicking button) - async with page.expect_navigation(timeout=30000): - await page.evaluate("document.querySelector('form').submit()") - - logger.info(f"Keycloak login submitted for {username}, redirected to callback") - - # Check if we need to handle consent screen - # Keycloak consent screen has "Yes" button - consent_button = page.locator('input[name="accept"][value="Yes"]') - if await consent_button.count() > 0: - logger.info("Keycloak consent screen detected, clicking Yes...") - await consent_button.click() - await page.wait_for_load_state("networkidle", timeout=30000) - logger.info("Keycloak consent granted") - - # Wait for callback server to receive auth code with timeout - logger.info(f"Waiting for auth code with state: {state[:16]}...") - timeout = 30 # seconds - start_time = time.time() - auth_code = None - - while time.time() - start_time < timeout: - if state in auth_states: - auth_code = auth_states[state] - logger.info("Auth code received from callback server") - break - await anyio.sleep(0.1) - else: - raise TimeoutError( - f"Auth code not received within {timeout}s. State: {state[:16]}..." - ) - - finally: - await context.close() - - # Exchange authorization code for access token (with PKCE code_verifier) - logger.info("Exchanging authorization code for access token with PKCE...") - async with httpx.AsyncClient(timeout=30.0) as token_client: - token_response = await token_client.post( - token_endpoint, - data={ - "grant_type": "authorization_code", - "code": auth_code, - "redirect_uri": callback_url, - "client_id": client_id, - "client_secret": client_secret, - "code_verifier": code_verifier, # PKCE verifier - }, - ) - - token_response.raise_for_status() - token_data = token_response.json() - access_token = token_data.get("access_token") - - if not access_token: - raise ValueError(f"No access_token in response: {token_data}") - - logger.info( - f"Successfully obtained Keycloak OAuth access token with scopes: {scopes}" - ) - return access_token - - -@pytest.fixture(scope="session") -async def keycloak_oauth_token( - anyio_backend, browser, keycloak_oauth_client_credentials, oauth_callback_server -) -> str: - """ - Fixture to obtain an OAuth access token from Keycloak using Playwright automation. - - This fixture tests the external IdP flow where: - 1. User authenticates with Keycloak (external IdP) - 2. Keycloak issues an access token with Nextcloud custom scopes - 3. Token is used to access Nextcloud APIs via user_oidc app validation - - The Nextcloud custom scopes (notes:read, calendar:write, etc.) are now defined - in Keycloak's realm configuration and can be requested in the OAuth flow. - - Returns: - OAuth access token from Keycloak for the admin user with full scopes - """ - # Standard OIDC scopes + Nextcloud custom scopes (now defined in Keycloak realm) - default_scopes = "openid profile email offline_access notes:read notes:write calendar:read calendar:write contacts:read contacts:write cookbook:read cookbook:write deck:read deck:write tables:read tables:write files:read files:write sharing:read sharing:write todo:read todo:write" - - return await _get_keycloak_oauth_token( - browser, - keycloak_oauth_client_credentials, - oauth_callback_server, - scopes=default_scopes, - username="admin", - password="admin", - ) - - -@pytest.fixture(scope="session") -async def keycloak_oauth_token_read_only( - anyio_backend, browser, keycloak_oauth_client_credentials, oauth_callback_server -) -> str: - """ - Fixture to obtain a Keycloak OAuth token with only read scopes. - - This token will only be able to perform read operations and should - have write tools filtered out from the tool list. - - Returns: - OAuth access token from Keycloak for test_read_only user with read-only scopes - """ - return await _get_keycloak_oauth_token( - browser, - keycloak_oauth_client_credentials, - oauth_callback_server, - scopes=DEFAULT_READ_SCOPES, - username="test_read_only", - password="test123", - ) - - -@pytest.fixture(scope="session") -async def keycloak_oauth_token_write_only( - anyio_backend, browser, keycloak_oauth_client_credentials, oauth_callback_server -) -> str: - """ - Fixture to obtain a Keycloak OAuth token with only write scopes. - - This token will only be able to perform write operations and should - have read tools filtered out from the tool list. - - Returns: - OAuth access token from Keycloak for test_write_only user with write-only scopes - """ - return await _get_keycloak_oauth_token( - browser, - keycloak_oauth_client_credentials, - oauth_callback_server, - scopes=DEFAULT_WRITE_SCOPES, - username="test_write_only", - password="test123", - ) - - -@pytest.fixture(scope="session") -async def keycloak_oauth_token_no_custom_scopes( - anyio_backend, browser, keycloak_oauth_client_credentials, oauth_callback_server -) -> str: - """ - Fixture to obtain a Keycloak OAuth token with NO custom scopes. - - Tests the security behavior when a user grants only default OIDC scopes - (openid, profile, email) but declines application-specific scopes. - - Expected behavior: Should see 0 tools (all tools require custom scopes). - - Returns: - OAuth access token from Keycloak for test_no_scopes user with no custom scopes - """ - return await _get_keycloak_oauth_token( - browser, - keycloak_oauth_client_credentials, - oauth_callback_server, - scopes="openid profile email", # No custom scopes - username="test_no_scopes", - password="test123", - ) - - -@pytest.fixture(scope="session") -async def nc_mcp_keycloak_client( - anyio_backend, keycloak_oauth_token -) -> AsyncGenerator[ClientSession, Any]: - """ - Session-scoped fixture providing an MCP client session authenticated with Keycloak tokens. - - This MCP client connects to the mcp-keycloak service (port 8002) which is configured - to use Keycloak as an external identity provider. The token flow is: - - 1. Keycloak issues OAuth token (via keycloak_oauth_token fixture) - 2. MCP client uses token to authenticate with MCP server - 3. MCP server validates token via Nextcloud user_oidc app - 4. MCP server uses validated token to access Nextcloud APIs - - This tests ADR-002 external IdP integration. - - Yields: - MCP client session for testing tools/resources with Keycloak auth - """ - mcp_url = "http://localhost:8002/mcp" - logger.info(f"Creating MCP client session for Keycloak external IdP at {mcp_url}") - logger.info("Using Keycloak OAuth token for authentication") - - async for session in create_mcp_client_session( - url=mcp_url, token=keycloak_oauth_token, client_name="Keycloak External IdP MCP" - ): - logger.info("✓ MCP client session established with Keycloak authentication") - yield session - logger.info("✓ MCP client session closed") - - -@pytest.fixture(scope="session") -async def nc_mcp_keycloak_client_read_only( - anyio_backend, keycloak_oauth_token_read_only -) -> AsyncGenerator[ClientSession, Any]: - """ - MCP client session authenticated with Keycloak read-only token. - - This client should only see read tools and should get filtered - write tools based on token scopes. - - Uses JWT tokens because they embed scope information in claims, - enabling proper scope-based tool filtering. - """ - mcp_url = "http://localhost:8002/mcp" - logger.info(f"Creating read-only MCP client session for Keycloak at {mcp_url}") - - async for session in create_mcp_client_session( - url=mcp_url, - token=keycloak_oauth_token_read_only, - client_name="Keycloak Read-Only MCP", - ): - yield session - - -@pytest.fixture(scope="session") -async def nc_mcp_keycloak_client_write_only( - anyio_backend, keycloak_oauth_token_write_only -) -> AsyncGenerator[ClientSession, Any]: - """ - MCP client session authenticated with Keycloak write-only token. - - This client should only see write tools and should get filtered - read tools based on token scopes. - - Uses JWT tokens because they embed scope information in claims, - enabling proper scope-based tool filtering. - """ - mcp_url = "http://localhost:8002/mcp" - logger.info(f"Creating write-only MCP client session for Keycloak at {mcp_url}") - - async for session in create_mcp_client_session( - url=mcp_url, - token=keycloak_oauth_token_write_only, - client_name="Keycloak Write-Only MCP", - ): - yield session - - -@pytest.fixture(scope="session") -async def nc_mcp_keycloak_client_no_custom_scopes( - anyio_backend, keycloak_oauth_token_no_custom_scopes -) -> AsyncGenerator[ClientSession, Any]: - """ - MCP client session authenticated with Keycloak token without custom scopes. - - This client has only OIDC default scopes (openid, profile, email) without - application-specific scopes (notes:read, notes:write, etc.). - - Expected behavior: Should see 0 tools (all tools require custom scopes). - - Uses JWT tokens because they embed scope information in claims, - enabling proper scope-based tool filtering. - """ - mcp_url = "http://localhost:8002/mcp" - logger.info( - f"Creating no-custom-scopes MCP client session for Keycloak at {mcp_url}" - ) - - async for session in create_mcp_client_session( - url=mcp_url, - token=keycloak_oauth_token_no_custom_scopes, - client_name="Keycloak No Custom Scopes MCP", - ): - yield session - - -# ======================================================================== # Astrolabe Dynamic Configuration Fixtures -# ======================================================================== +# =========================================================================================== @pytest.fixture(scope="session") diff --git a/tests/unit/test_config_validators.py b/tests/unit/test_config_validators.py index a31bda70..ad242c14 100644 --- a/tests/unit/test_config_validators.py +++ b/tests/unit/test_config_validators.py @@ -22,16 +22,6 @@ from nextcloud_mcp_server.config_validators import ( class TestModeDetection: """Test auth mode detection from configuration.""" - def test_token_exchange_mode_detection(self): - """Test token exchange mode is detected.""" - settings = Settings( - nextcloud_host="http://localhost", - enable_token_exchange=True, - ) - - mode = detect_auth_mode(settings) - assert mode == AuthMode.OAUTH_TOKEN_EXCHANGE - def test_multi_user_basic_mode_detection(self): """Test multi-user BasicAuth mode is detected.""" settings = Settings( @@ -62,18 +52,6 @@ class TestModeDetection: mode = detect_auth_mode(settings) assert mode == AuthMode.OAUTH_SINGLE_AUDIENCE - def test_mode_priority_token_exchange_over_basic(self): - """Test token exchange has priority over BasicAuth.""" - settings = Settings( - nextcloud_host="http://localhost", - nextcloud_username="admin", - nextcloud_password="password", - enable_token_exchange=True, - ) - - mode = detect_auth_mode(settings) - assert mode == AuthMode.OAUTH_TOKEN_EXCHANGE - class TestSingleUserBasicValidation: """Test validation for single-user BasicAuth mode.""" @@ -165,21 +143,6 @@ class TestSingleUserBasicValidation: # It will fail multi-user validation because username/password are forbidden assert len(errors) > 0 - def test_forbidden_token_exchange(self): - """Test error when ENABLE_TOKEN_EXCHANGE is set.""" - settings = Settings( - nextcloud_host="http://localhost", - nextcloud_username="admin", - nextcloud_password="password", - enable_token_exchange=True, - ) - - # Note: This will detect as OAUTH_TOKEN_EXCHANGE due to priority - mode, errors = validate_configuration(settings) - - assert mode == AuthMode.OAUTH_TOKEN_EXCHANGE - # It will fail OAuth validation - def test_vector_sync_without_embedding_provider_uses_fallback(self): """Test that vector sync works with Simple provider fallback (no config needed).""" settings = Settings( @@ -419,51 +382,6 @@ class TestOAuthSingleAudienceValidation: assert settings.enable_offline_access is True -class TestOAuthTokenExchangeValidation: - """Test validation for OAuth token exchange mode.""" - - def test_valid_minimal_config(self): - """Test valid minimal OAuth token exchange config.""" - settings = Settings( - nextcloud_host="http://localhost", - enable_token_exchange=True, - ) - - mode, errors = validate_configuration(settings) - - assert mode == AuthMode.OAUTH_TOKEN_EXCHANGE - assert len(errors) == 0 - - def test_valid_with_credentials(self): - """Test valid config with OAuth credentials.""" - settings = Settings( - nextcloud_host="http://localhost", - enable_token_exchange=True, - oidc_client_id="test-client", - oidc_client_secret="test-secret", - ) - - mode, errors = validate_configuration(settings) - - assert mode == AuthMode.OAUTH_TOKEN_EXCHANGE - assert len(errors) == 0 - - def test_forbidden_username_password(self): - """Test error when username/password are set.""" - settings = Settings( - nextcloud_host="http://localhost", - enable_token_exchange=True, - nextcloud_username="admin", - nextcloud_password="password", - ) - - mode, errors = validate_configuration(settings) - - assert mode == AuthMode.OAUTH_TOKEN_EXCHANGE - assert any("nextcloud_username" in err.lower() for err in errors) - assert any("nextcloud_password" in err.lower() for err in errors) - - class TestModeSummary: """Test mode summary generation.""" @@ -477,14 +395,6 @@ class TestModeSummary: assert "NEXTCLOUD_PASSWORD" in summary assert "VECTOR_SYNC_ENABLED" in summary - def test_oauth_token_exchange_summary(self): - """Test summary for OAuth token exchange mode.""" - summary = get_mode_summary(AuthMode.OAUTH_TOKEN_EXCHANGE) - - assert "oauth_exchange" in summary - assert "ENABLE_TOKEN_EXCHANGE" in summary - assert "RFC 8693" in summary - class TestEdgeCases: """Test edge cases and boundary conditions.""" @@ -800,23 +710,6 @@ class TestExplicitModeSelection: assert mode == AuthMode.OAUTH_SINGLE_AUDIENCE - def test_explicit_oauth_token_exchange_mode(self): - """Test explicit oauth_token_exchange mode selection.""" - with patch.dict( - os.environ, - { - "NEXTCLOUD_HOST": "http://localhost:8080", - "MCP_DEPLOYMENT_MODE": "oauth_token_exchange", - }, - clear=True, - ): - from nextcloud_mcp_server.config import get_settings - - settings = get_settings() - mode = detect_auth_mode(settings) - - assert mode == AuthMode.OAUTH_TOKEN_EXCHANGE - def test_invalid_deployment_mode_raises_error(self): """Test invalid MCP_DEPLOYMENT_MODE raises ValueError.""" with patch.dict( diff --git a/tests/unit/test_management_status_endpoint.py b/tests/unit/test_management_status_endpoint.py index ac6ad1a6..f6ea5f40 100644 --- a/tests/unit/test_management_status_endpoint.py +++ b/tests/unit/test_management_status_endpoint.py @@ -37,7 +37,6 @@ def create_mock_settings( oidc_issuer: str | None = None, vector_sync_enabled: bool = False, nextcloud_url: str = "http://localhost", - enable_token_exchange: bool = False, mcp_client_id: str | None = None, mcp_client_secret: str | None = None, ): @@ -49,7 +48,6 @@ def create_mock_settings( settings.oidc_issuer = oidc_issuer settings.vector_sync_enabled = vector_sync_enabled settings.nextcloud_url = nextcloud_url - settings.enable_token_exchange = enable_token_exchange settings.mcp_client_id = mcp_client_id settings.mcp_client_secret = mcp_client_secret return settings diff --git a/tests/unit/test_unified_verifier.py b/tests/unit/test_unified_verifier.py index 491ee0f2..7747bbde 100644 --- a/tests/unit/test_unified_verifier.py +++ b/tests/unit/test_unified_verifier.py @@ -29,18 +29,9 @@ def base_settings(): nextcloud_resource_uri="http://localhost:8080", jwks_uri="https://idp.example.com/jwks", introspection_uri="https://idp.example.com/introspect", - enable_token_exchange=False, # Multi-audience mode - token_exchange_cache_ttl=300, ) -@pytest.fixture -def exchange_settings(base_settings): - """Create settings for token exchange mode.""" - base_settings.enable_token_exchange = True - return base_settings - - class TestUnifiedTokenVerifierInit: """Test UnifiedTokenVerifier initialization.""" @@ -50,11 +41,11 @@ class TestUnifiedTokenVerifierInit: assert verifier.mode == "multi-audience" assert verifier.settings == base_settings - def test_init_exchange_mode(self, exchange_settings): - """Test verifier initialization in token exchange mode.""" - verifier = UnifiedTokenVerifier(exchange_settings) - assert verifier.mode == "exchange" - assert verifier.settings == exchange_settings + def test_init_always_multi_audience(self, base_settings): + """Test verifier always initializes in multi-audience mode.""" + verifier = UnifiedTokenVerifier(base_settings) + assert verifier.mode == "multi-audience" + assert verifier.settings == base_settings class TestAudienceValidation: @@ -117,9 +108,9 @@ class TestAudienceValidation: # Should pass - we only validate MCP audience per RFC 7519 assert verifier._has_mcp_audience(payload) is True - def test_has_mcp_audience_with_client_id(self, exchange_settings): + def test_has_mcp_audience_with_client_id(self, base_settings): """Test MCP audience validation with client ID.""" - verifier = UnifiedTokenVerifier(exchange_settings) + verifier = UnifiedTokenVerifier(base_settings) payload = { "aud": ["test-client-id"], "sub": "testuser", @@ -128,9 +119,9 @@ class TestAudienceValidation: assert verifier._has_mcp_audience(payload) is True - def test_has_mcp_audience_with_server_url(self, exchange_settings): + def test_has_mcp_audience_with_server_url(self, base_settings): """Test MCP audience validation with server URL.""" - verifier = UnifiedTokenVerifier(exchange_settings) + verifier = UnifiedTokenVerifier(base_settings) payload = { "aud": ["http://localhost:8000"], "sub": "testuser", @@ -139,9 +130,9 @@ class TestAudienceValidation: assert verifier._has_mcp_audience(payload) is True - def test_has_mcp_audience_missing(self, exchange_settings): + def test_has_mcp_audience_missing(self, base_settings): """Test MCP audience validation fails without MCP audience.""" - verifier = UnifiedTokenVerifier(exchange_settings) + verifier = UnifiedTokenVerifier(base_settings) payload = { "aud": ["http://localhost:8080"], # Wrong audience "sub": "testuser", @@ -292,12 +283,12 @@ class TestMultiAudienceVerification: assert result.resource == "testuser" -class TestExchangeModeVerification: - """Test token exchange mode verification.""" +class TestMcpAudienceVerification: + """Test MCP audience verification.""" - async def test_verify_mcp_audience_only_success(self, exchange_settings): + async def test_verify_mcp_audience_only_success(self, base_settings): """Test MCP-only audience verification succeeds with MCP audience.""" - verifier = UnifiedTokenVerifier(exchange_settings) + verifier = UnifiedTokenVerifier(base_settings) # Mock introspection response with MCP audience only introspection_response = { @@ -318,9 +309,9 @@ class TestExchangeModeVerification: assert result is not None assert result.resource == "testuser" - async def test_verify_mcp_audience_only_fails_without_mcp(self, exchange_settings): + async def test_verify_mcp_audience_only_fails_without_mcp(self, base_settings): """Test MCP audience verification fails without MCP audience.""" - verifier = UnifiedTokenVerifier(exchange_settings) + verifier = UnifiedTokenVerifier(base_settings) # Mock introspection response without MCP audience introspection_response = { @@ -503,9 +494,9 @@ class TestVerifyTokenFlow: assert result is not None assert result.resource == "testuser" - async def test_verify_token_exchange_mode(self, exchange_settings): - """Test verify_token in exchange mode.""" - verifier = UnifiedTokenVerifier(exchange_settings) + async def test_verify_token_mcp_audience_only(self, base_settings): + """Test verify_token with MCP audience only.""" + verifier = UnifiedTokenVerifier(base_settings) introspection_response = { "active": True,