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) <noreply@anthropic.com>

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"

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,16 +771,6 @@ async def setup_oauth_config():
     token_verifier = UnifiedTokenVerifier(settings)
 
     # Log the mode
-    enable_token_exchange = (
-        os.getenv("ENABLE_TOKEN_EXCHANGE", "false").lower() == "true"
-    )
-    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"
     )
@@ -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 (
+    oauth_enabled = mode == AuthMode.OAUTH_SINGLE_AUDIENCE
-        AuthMode.OAUTH_SINGLE_AUDIENCE,
-        AuthMode.OAUTH_TOKEN_EXCHANGE,
-    )
     # 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 (
+        if mode == AuthMode.OAUTH_SINGLE_AUDIENCE:
-            mode == AuthMode.OAUTH_SINGLE_AUDIENCE
-            or mode == AuthMode.OAUTH_TOKEN_EXCHANGE
-        ):
             checks["auth_mode"] = "oauth"
             checks["auth_configured"] = "ok"
         elif mode == AuthMode.MULTI_USER_BASIC:

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")

nextcloud_mcp_server/auth/keycloak_oauth.py

@@ -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 <realm> \
-                --uusername service-account-<client-id> \
-                --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"]

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

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]:

nextcloud_mcp_server/auth/token_exchange.py

@@ -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: <client-id> (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,
-    )

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)

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,
+    # Valid values: single_user_basic, multi_user_basic, oauth_single_audience
-    #               oauth_token_exchange
     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"),

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)
+    1. Multi-user BasicAuth
-    2. Multi-user BasicAuth
+    2. Single-user BasicAuth
-    3. Single-user BasicAuth
+    3. OAuth single-audience (default OAuth mode)
-    4. 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 [
+    if mode == AuthMode.OAUTH_SINGLE_AUDIENCE:
-        AuthMode.OAUTH_SINGLE_AUDIENCE,
-        AuthMode.OAUTH_TOKEN_EXCHANGE,
-    ]:
         # 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:

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 (
+from nextcloud_mcp_server.auth.context_helper import get_client_from_context
-    get_client_from_context,
-    get_session_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. Login Flow v2: OAuth for MCP session, app password for Nextcloud API
-    2. OAuth mode:
+    3. Multi-user BasicAuth: Credentials passed through from request headers
-       a. Multi-audience mode (ENABLE_TOKEN_EXCHANGE=false, default):
+    4. OAuth multi-audience: Token contains both MCP and Nextcloud audiences
-          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.
 
     This function automatically detects the authentication mode by checking
     the type of the lifespan context.
@@ -74,17 +62,8 @@ 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)

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",

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:

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")

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(

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

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):
+    def test_init_always_multi_audience(self, base_settings):
-        """Test verifier initialization in token exchange mode."""
+        """Test verifier always initializes in multi-audience mode."""
-        verifier = UnifiedTokenVerifier(exchange_settings)
+        verifier = UnifiedTokenVerifier(base_settings)
-        assert verifier.mode == "exchange"
+        assert verifier.mode == "multi-audience"
-        assert verifier.settings == exchange_settings
+        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:
+class TestMcpAudienceVerification:
-    """Test token exchange mode verification."""
+    """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):
+    async def test_verify_token_mcp_audience_only(self, base_settings):
-        """Test verify_token in exchange mode."""
+        """Test verify_token with MCP audience only."""
-        verifier = UnifiedTokenVerifier(exchange_settings)
+        verifier = UnifiedTokenVerifier(base_settings)
 
         introspection_response = {
             "active": True,