diff --git a/app-hooks/post-installation/10-install-oidc-app.sh b/app-hooks/post-installation/10-install-oidc-app.sh
index 5a0094dd..b4367bc7 100755
--- a/app-hooks/post-installation/10-install-oidc-app.sh
+++ b/app-hooks/post-installation/10-install-oidc-app.sh
@@ -35,5 +35,6 @@ php /var/www/html/occ config:app:set oidc dynamic_client_registration --value='t
php /var/www/html/occ config:app:set oidc proof_key_for_code_exchange --value=true --type=boolean
php /var/www/html/occ config:app:set oidc allow_user_settings --value='enabled'
php /var/www/html/occ config:app:set oidc default_token_type --value='jwt'
+php /var/www/html/occ config:app:set oidc default_resource_identifier --value='http://localhost:8080'
echo "OIDC app installed and configured successfully"
diff --git a/app-hooks/post-installation/30-disable-welcome-wizard.sh b/app-hooks/post-installation/30-disable-welcome-wizard.sh
new file mode 100755
index 00000000..ce352f23
--- /dev/null
+++ b/app-hooks/post-installation/30-disable-welcome-wizard.sh
@@ -0,0 +1,3 @@
+#!/bin/bash
+
+php /var/www/html/occ config:app:set --value false firstrunwizard wizard_enabled
diff --git a/docker-compose.yml b/docker-compose.yml
index 68932523..4109b2c8 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -96,6 +96,7 @@ services:
# OIDC_CLIENT_ID not set - uses Dynamic Client Registration (DCR)
- NEXTCLOUD_HOST=http://app:80
- NEXTCLOUD_MCP_SERVER_URL=http://localhost:8001
+ - NEXTCLOUD_RESOURCE_URI=http://localhost:8080 # ADR-005: Nextcloud resource identifier for audience validation
- NEXTCLOUD_PUBLIC_ISSUER_URL=http://localhost:8080
- NEXTCLOUD_OIDC_SCOPES=openid profile email 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
@@ -104,8 +105,9 @@ services:
- TOKEN_ENCRYPTION_KEY=ESF1BvEQdGYsCluwMx9Cxvw3uh5pFowPH7Rg_nIliyo=
- TOKEN_STORAGE_DB=/app/data/tokens.db
- # ADR-004: Use Hybrid Flow (server intercepts OAuth callback)
- # Set to false to enable Hybrid Flow tests - server stores refresh token and issues MCP codes
+ # ADR-005: Multi-audience mode (default - ENABLE_TOKEN_EXCHANGE=false)
+ # Tokens must contain BOTH MCP and Nextcloud audiences
+ # No token exchange needed - tokens work for both MCP auth and Nextcloud APIs
# NO admin credentials - using OAuth with Dynamic Client Registration (DCR)
# Client credentials registered via RFC 7591 and stored in volume
@@ -159,6 +161,7 @@ services:
# Nextcloud API endpoint (for accessing APIs with validated token)
- NEXTCLOUD_HOST=http://app:80
- NEXTCLOUD_MCP_SERVER_URL=http://localhost:8002
+ - NEXTCLOUD_RESOURCE_URI=nextcloud # ADR-005: Keycloak uses client IDs as audiences, not URLs
- NEXTCLOUD_PUBLIC_ISSUER_URL=http://localhost:8888/realms/nextcloud-mcp
# Refresh token storage (ADR-002 Tier 1 & 2)
@@ -166,8 +169,11 @@ services:
- TOKEN_ENCRYPTION_KEY=ESF1BvEQdGYsCluwMx9Cxvw3uh5pFowPH7Rg_nIliyo=
- TOKEN_STORAGE_DB=/app/data/tokens.db
- # Token exchange (RFC 8693) - convert aud:nextcloud-mcp-server → aud:nextcloud
+ # ADR-005: Token exchange mode (RFC 8693)
+ # Exchange MCP tokens (aud: nextcloud-mcp-server) for Nextcloud tokens (aud: http://localhost:8080)
+ # Provides strict audience separation between MCP session and Nextcloud API access
- ENABLE_TOKEN_EXCHANGE=true
+ - TOKEN_EXCHANGE_CACHE_TTL=300 # Cache exchanged tokens for 5 minutes (default)
# OAuth scopes (optional - uses defaults if not specified)
- NEXTCLOUD_OIDC_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
diff --git a/docs/ADR-005-token-audience-validation.md b/docs/ADR-005-token-audience-validation.md
new file mode 100644
index 00000000..3e6c4d6c
--- /dev/null
+++ b/docs/ADR-005-token-audience-validation.md
@@ -0,0 +1,1010 @@
+# ADR-005: Token Audience Validation and Security Compliance
+
+**Status**: Implemented
+**Date**: 2025-01-05
+**Updated**: 2025-11-05
+**Related**: Issue #261, ADR-004, upstream-oauth.md, RFC 7519, RFC 8707, RFC 9728
+**Supersedes**: Token passthrough mode in ADR-004
+
+## Implementation Note
+
+This ADR has been fully implemented with key simplifications based on RFC 7519 Section 4.1.3:
+- MCP server validates only its own audience (not Nextcloud's)
+- OAuth requests include `resource` parameter (RFC 8707)
+- Clients discover resource via PRM endpoint (RFC 9728)
+- Nextcloud OIDC app uses client-specific resource URLs
+
+## Executive Summary
+
+This ADR addresses a critical security vulnerability where the MCP server was passing tokens intended for itself directly to Nextcloud APIs (token passthrough). We will:
+
+1. **Replace two non-compliant token verifiers** with a single `UnifiedTokenVerifier`
+2. **Implement proper audience validation** requiring tokens to explicitly include appropriate audiences
+3. **Support two compliant modes**:
+ - **Multi-audience mode (default)**: Tokens contain both MCP and Nextcloud audiences
+ - **Token exchange mode (opt-in)**: MCP tokens are exchanged for Nextcloud tokens via RFC 8693
+4. **Remove all token passthrough paths** to comply with MCP security specification
+
+The solution works within python-sdk constraints by implementing a two-layer architecture where token validation happens in the verifier and token exchange happens when creating API clients.
+
+## Context
+
+The MCP Security Best Practices specification explicitly forbids "token passthrough" - an anti-pattern where an MCP server accepts tokens from clients without validating they were properly issued to the MCP server, then passes them through to downstream APIs.
+
+### Current Vulnerability
+
+The Nextcloud MCP server currently supports two OAuth modes via the `ENABLE_TOKEN_EXCHANGE` flag:
+
+1. **Pass-through mode** (`ENABLE_TOKEN_EXCHANGE=false`, **default**):
+ - Accepts Flow 1 tokens with audience matching MCP server URL or client ID
+ - Passes these tokens **directly** to Nextcloud APIs without audience transformation
+ - **Violates MCP specification** - token intended for MCP server is used against Nextcloud
+
+2. **Token exchange mode** (`ENABLE_TOKEN_EXCHANGE=true`, opt-in):
+ - Accepts Flow 1 tokens with audience matching MCP server URL
+ - Uses RFC 8693 to exchange for tokens with Nextcloud resource URI audience
+ - **Compliant** with MCP specification but adds latency
+
+**Location of vulnerability**: `nextcloud_mcp_server/context.py:62-66`
+
+### Security Risks (per MCP specification)
+
+1. **Security Control Circumvention**: Downstream APIs cannot distinguish between clients when all use the same token
+2. **Accountability Issues**: Broken audit trails - logs show wrong identity/source
+3. **Trust Boundary Violations**: Token meant for one service used for another
+4. **Future Compatibility**: Cannot add security controls later without breaking changes
+
+### OAuth Feature Status
+
+The OAuth integration is currently **experimental** and requires an upstream fix in Nextcloud server to properly handle bearer tokens (see `docs/upstream-oauth.md` for details). Until the upstream fix is merged, **all breaking changes are acceptable** to ensure security compliance.
+
+## Decision
+
+We will **remove the token passthrough anti-pattern entirely** and enforce proper token audience validation in all OAuth deployments.
+
+### Architectural Approach
+
+Based on analysis of the existing code and python-sdk constraints, we will:
+
+1. **Consolidate two non-compliant verifiers** (`NextcloudTokenVerifier` and `ProgressiveConsentTokenVerifier`) into a single `UnifiedTokenVerifier`
+2. **Implement a two-layer architecture**:
+ - **Verification Layer**: `UnifiedTokenVerifier` validates audiences only (complies with SDK protocol)
+ - **Exchange Layer**: `context_helper.py` performs token exchange when needed
+3. **Support two compliant modes** determined by the `ENABLE_TOKEN_EXCHANGE` setting:
+
+### Mode 1: Multi-Audience Token Validation (Default)
+
+Use multi-audience tokens directly. Per RFC 7519 Section 4.1.3, resource servers validate only their own presence in the audience claim. The MCP server validates its own audience; Nextcloud independently validates its own audience when receiving API calls. This is the default mode when `ENABLE_TOKEN_EXCHANGE` is false or not set.
+
+**Requirements**:
+- Token must have `aud` claim containing MCP server audience:
+ - Client ID OR
+ - MCP server URL (e.g., `http://localhost:8001`) OR
+ - MCP server URL with /mcp suffix (e.g., `http://localhost:8001/mcp`)
+- For Nextcloud API access to work, token should also include Nextcloud audience (validated by Nextcloud, not MCP)
+- Single token works for both MCP authentication and Nextcloud API access
+- IdP must support multi-audience tokens for full functionality
+
+**Resource URI Configuration**:
+- Nextcloud OIDC app: Set via `default_resource_identifier` (default: `http://localhost:8080`)
+- Keycloak: Configure resource servers with proper URIs
+- MCP Server: Defaults to `NEXTCLOUD_MCP_SERVER_URL` environment variable
+
+**Use case**: Standard deployments where IdP can issue tokens with multiple audiences
+
+**Configuration**:
+```bash
+# Multi-audience mode (default when not set or false)
+ENABLE_TOKEN_EXCHANGE=false # or omit entirely
+
+# Resource URIs for audience validation
+NEXTCLOUD_MCP_SERVER_URL=http://localhost:8000 # MCP server URL (used as audience)
+NEXTCLOUD_RESOURCE_URI=http://localhost:8080 # Nextcloud resource identifier
+
+# Client ID (alternative audience for MCP)
+OIDC_CLIENT_ID=nextcloud-mcp-server
+```
+
+**Token validation logic (RFC 7519 compliant) - Actual Implementation**:
+```python
+def _has_mcp_audience(self, payload: dict[str, Any]) -> bool:
+ """
+ Check if token has MCP audience.
+
+ Per RFC 7519 Section 4.1.3, resource servers should only validate their own
+ presence in the audience claim. We don't validate Nextcloud's audience - that's
+ Nextcloud's responsibility when it receives the token.
+ """
+ audiences = payload.get("aud", [])
+ if isinstance(audiences, str):
+ audiences = [audiences]
+
+ audiences_set = set(audiences)
+
+ # MCP must have at least one: client_id OR server_url OR server_url/mcp
+ return bool(
+ self.settings.oidc_client_id in audiences_set
+ or (
+ self.settings.nextcloud_mcp_server_url
+ and (
+ self.settings.nextcloud_mcp_server_url in audiences_set
+ or f"{self.settings.nextcloud_mcp_server_url}/mcp" in audiences_set
+ )
+ )
+ )
+```
+
+### Mode 2: RFC 8693 Token Exchange (Opt-in)
+
+Exchange MCP session tokens for Nextcloud-specific tokens via RFC 8693. This mode is activated when `ENABLE_TOKEN_EXCHANGE=true`.
+
+**Requirements**:
+- Client provides token with MCP audience (client ID or server URL)
+- Server exchanges it for ephemeral token with Nextcloud resource URI
+- IdP must support RFC 8693 token exchange endpoint
+- Exchanged tokens cached for 5 minutes to reduce latency
+
+**Performance Consideration**: In the context of an agentic LLM application, the additional network call for token exchange (typically 50-100ms) is negligible compared to LLM inference time (seconds). The security benefit far outweighs the minimal latency cost.
+
+**Use case**:
+- Deployments requiring strict audience separation
+- IdPs with full RFC 8693 support (e.g., Keycloak with token exchange enabled)
+
+**Configuration**:
+```bash
+# Token exchange mode (opt-in for strict separation)
+ENABLE_TOKEN_EXCHANGE=true
+
+# Resource URIs
+NEXTCLOUD_MCP_SERVER_URL=http://localhost:8000 # MCP server URL
+NEXTCLOUD_RESOURCE_URI=http://localhost:8080 # Nextcloud resource identifier
+
+# Optional: Cache TTL
+TOKEN_EXCHANGE_CACHE_TTL=300 # seconds (default: 300)
+
+# OIDC discovery URL (token endpoint is auto-discovered from this)
+OIDC_DISCOVERY_URL=http://keycloak:8080/realms/nextcloud-mcp/.well-known/openid-configuration
+```
+
+**Token exchange with caching**:
+```python
+class TokenExchangeCache:
+ """Cache exchanged tokens to reduce exchange frequency."""
+
+ def __init__(self, ttl_seconds: int = 300): # 5-minute default
+ self._cache: dict[str, tuple[str, float]] = {}
+ self._ttl = ttl_seconds
+
+ async def get_or_exchange(
+ self,
+ subject_token: str,
+ token_hash: str,
+ exchange_func: Callable
+ ) -> str:
+ """Get cached token or perform exchange."""
+ now = time.time()
+
+ # Check cache
+ if token_hash in self._cache:
+ cached_token, expiry = self._cache[token_hash]
+ if expiry > now:
+ logger.debug(f"Using cached exchanged token (expires in {expiry - now:.1f}s)")
+ return cached_token
+
+ # Perform exchange
+ logger.debug("Exchanging token for Nextcloud audience")
+ nextcloud_token = await exchange_func(
+ subject_token=subject_token,
+ requested_audience=self.nextcloud_resource_uri
+ )
+
+ # Cache with TTL
+ self._cache[token_hash] = (nextcloud_token, now + self._ttl)
+
+ # Clean expired entries
+ self._cache = {
+ k: v for k, v in self._cache.items()
+ if v[1] > now
+ }
+
+ return nextcloud_token
+```
+
+### Removed: Pass-through Mode (Non-compliant)
+
+The pass-through mode is **removed immediately** as it violates MCP security requirements. No migration period is provided since the OAuth feature is experimental.
+
+## Implementation
+
+### 1. Environment Variables
+
+**Required variables**:
+```bash
+# Resource URIs (required for audience validation)
+NEXTCLOUD_MCP_SERVER_URL=http://localhost:8000 # MCP server URL (used as audience)
+NEXTCLOUD_RESOURCE_URI=http://localhost:8080 # Nextcloud resource identifier
+
+# Client identification
+OIDC_CLIENT_ID=nextcloud-mcp-server # Can also be valid audience for MCP
+```
+
+**Mode selection**:
+```bash
+# Multi-audience mode (default)
+ENABLE_TOKEN_EXCHANGE=false # or omit entirely
+
+# Token exchange mode (opt-in)
+ENABLE_TOKEN_EXCHANGE=true # Activates RFC 8693 exchange
+```
+
+**Optional variables (exchange mode)**:
+```bash
+TOKEN_EXCHANGE_CACHE_TTL=300 # Cache TTL in seconds (default: 300)
+```
+
+### 2. Consolidate Token Verifiers
+
+**Current Issue**: Two TokenVerifier implementations exist (`NextcloudTokenVerifier` and `ProgressiveConsentTokenVerifier`), leading to code duplication, inconsistent validation logic, and pass-through vulnerabilities.
+
+**Solution**: Consolidate into a single `UnifiedTokenVerifier` class that handles both compliant validation modes:
+
+```python
+class UnifiedTokenVerifier(TokenVerifier):
+ """
+ Unified token verifier supporting both multi-audience and token exchange modes.
+ Compliant with MCP security specification - no token pass-through.
+ """
+
+ def __init__(self, settings: Settings):
+ self.settings = settings
+ self.mode = "exchange" if settings.enable_token_exchange else "multi-audience"
+
+ # Common components
+ self.http_client = httpx.AsyncClient(timeout=10.0)
+ self.jwks_client = PyJWKClient(settings.jwks_uri) if settings.jwks_uri else None
+
+ # Mode-specific initialization
+ if self.mode == "exchange":
+ # Exchange mode components (cache is in context helper, not here)
+ self.introspection_uri = settings.introspection_uri
+ self.client_secret = settings.oidc_client_secret
+
+ logger.info(f"Token verifier initialized in {self.mode} mode")
+
+ async def verify_token(self, token: str) -> AccessToken | None:
+ """
+ Verify token according to MCP TokenVerifier protocol.
+
+ Per RFC 7519, we validate only MCP audience. The mode determines what
+ happens AFTER verification in context_helper.py:
+ - Multi-audience mode: Use token directly (Nextcloud validates its own audience)
+ - Exchange mode: Exchange for Nextcloud-audience token via RFC 8693
+
+ Args:
+ token: Bearer token to verify
+
+ Returns:
+ AccessToken if valid with MCP audience, None otherwise
+ """
+ # Check cache first
+ cached = self._get_cached_token(token)
+ if cached:
+ logger.debug("Token found in cache")
+ return cached
+
+ # Both modes do the same validation (MCP audience only)
+ return await self._verify_mcp_audience(token)
+
+ async def _verify_mcp_audience(self, token: str) -> AccessToken | None:
+ """
+ Validate token has MCP audience.
+
+ Per RFC 7519 Section 4.1.3, resource servers validate only their own
+ presence in the audience claim. We don't validate Nextcloud's audience -
+ that's Nextcloud's responsibility when it receives the token.
+
+ Args:
+ token: Bearer token to verify
+
+ Returns:
+ AccessToken if valid with MCP audience, None otherwise
+ """
+ try:
+ # Attempt JWT verification first
+ if self._is_jwt_format(token) and self.jwks_client:
+ payload = await self._verify_jwt_signature(token)
+ else:
+ # Fall back to introspection for opaque tokens
+ payload = await self._introspect_token(token)
+ if not payload:
+ return None
+
+ # Validate MCP audience is present
+ if not self._has_mcp_audience(payload):
+ audiences = payload.get("aud", [])
+ logger.error(
+ f"Token rejected: Missing MCP audience. "
+ f"Got {audiences}, need MCP ({self.settings.oidc_client_id} or "
+ f"{self.settings.nextcloud_mcp_server_url})"
+ )
+ return None
+
+ # Log based on mode for clarity
+ if self.mode == "multi-audience":
+ logger.info(
+ "MCP audience validated - token can be used directly "
+ "(Nextcloud will validate its own audience)"
+ )
+ else:
+ logger.info(
+ "MCP audience validated - token will be exchanged for Nextcloud access"
+ )
+
+ return self._create_access_token(token, payload)
+```
+
+**Key Design Decisions**:
+
+1. **Separation of Concerns**: The verifier ONLY validates tokens. Token exchange happens in `context_helper.py` when creating the NextcloudClient, not in the verifier itself.
+
+2. **Python SDK Compatibility**: The MCP python-sdk's `TokenVerifier` protocol requires returning an `AccessToken` object. We comply with this interface while deferring exchange to the context layer.
+
+3. **Mode Selection**: Single class with mode-based behavior selected at startup via `ENABLE_TOKEN_EXCHANGE` environment variable.
+
+**Benefits**:
+- Single source of truth for token validation logic
+- Clear separation between validation and exchange
+- Compliant with MCP TokenVerifier protocol
+- Eliminates token pass-through vulnerability
+- Consistent error handling across all modes
+
+### 3. Error Handling and Propagation
+
+Token validation errors will be handled consistently:
+
+```python
+class TokenValidationError(Exception):
+ """Raised when token validation fails."""
+
+ def __init__(self, message: str, details: dict = None):
+ super().__init__(message)
+ self.details = details or {}
+ self.http_status = 401 # Unauthorized
+
+async def _verify_jwt_token(self, token: str) -> AccessToken:
+ """Verify JWT token with proper audience validation."""
+ try:
+ payload = jwt.decode(token, options={"verify_signature": False})
+ except jwt.InvalidTokenError as e:
+ raise TokenValidationError(
+ "Invalid JWT token format",
+ details={"error": str(e)}
+ )
+
+ # Validate audiences
+ if not await self.validate_token_audiences(payload, self.settings):
+ raise TokenValidationError(
+ "Token audiences do not meet requirements",
+ details={
+ "got": payload.get("aud"),
+ "need_mcp": [self.settings.oidc_client_id, self.settings.mcp_resource_uri],
+ "need_nextcloud": self.settings.nextcloud_resource_uri
+ }
+ )
+
+ # Additional validation (expiry, issuer, etc.)
+ # ...
+
+ return AccessToken(...)
+```
+
+### 4. Configuration Validation
+
+Startup validation ensures consistent configuration:
+
+```python
+def validate_oauth_configuration(settings: Settings):
+ """Validate OAuth configuration at startup."""
+ if not settings.nextcloud_mcp_server_url:
+ raise ValueError("NEXTCLOUD_MCP_SERVER_URL is required for audience validation")
+
+ if not settings.nextcloud_resource_uri:
+ raise ValueError("NEXTCLOUD_RESOURCE_URI is required for audience validation")
+
+ if settings.enable_token_exchange:
+ logger.info("Token exchange mode enabled - using RFC 8693 for strict audience separation")
+ if not settings.oidc_discovery_url:
+ logger.warning(
+ "No OIDC_DISCOVERY_URL configured. "
+ "Token endpoint discovery may fail."
+ )
+ else:
+ logger.info("Multi-audience mode enabled - tokens must contain both MCP and Nextcloud audiences")
+```
+
+### 5. OAuth Resource Parameters and PRM Discovery
+
+To ensure tokens have the correct audience, OAuth authorization requests must include the `resource` parameter (RFC 8707):
+
+**OAuth Authorization Requests**:
+```python
+# Flow 1 (MCP Client Authentication)
+idp_params = {
+ "client_id": idp_client_id,
+ "redirect_uri": callback_uri,
+ "response_type": "code",
+ "scope": scopes,
+ "state": idp_state,
+ "prompt": "consent",
+ "resource": f"{oauth_config['mcp_server_url']}/mcp", # MCP server audience
+}
+
+# Flow 2 (Nextcloud Resource Access)
+idp_params = {
+ ...
+ "resource": oauth_config["nextcloud_resource_uri"], # Nextcloud audience
+}
+```
+
+**Protected Resource Metadata (PRM) Endpoint**:
+
+The MCP server exposes PRM metadata at `/.well-known/oauth-protected-resource` (RFC 9728):
+```json
+{
+ "resource": "http://localhost:8001/mcp",
+ "scopes_supported": ["notes:read", "notes:write", ...],
+ "authorization_servers": ["http://localhost:8080"],
+ "bearer_methods_supported": ["header"],
+ "resource_signing_alg_values_supported": ["RS256"]
+}
+```
+
+**Client Discovery Pattern**:
+```python
+# Clients should discover resource identifier from PRM
+prm_url = f"{mcp_server_url}/.well-known/oauth-protected-resource"
+async with httpx.AsyncClient() as client:
+ prm_response = await client.get(prm_url, timeout=10)
+ prm_data = prm_response.json()
+ resource_identifier = prm_data.get("resource")
+
+# Use discovered resource in OAuth request
+auth_url = f"{authorization_endpoint}?resource={quote(resource_identifier, safe='')}&..."
+```
+
+### 6. Context Helper Updates
+
+Update `context.py` to handle token exchange at the NextcloudClient creation point:
+
+```python
+async def get_client(ctx: Context) -> NextcloudClient:
+ """Get NextcloudClient based on authentication mode."""
+ settings = get_settings()
+ lifespan_ctx = ctx.request_context.lifespan_context
+
+ # BasicAuth mode - unchanged
+ if hasattr(lifespan_ctx, "client"):
+ return lifespan_ctx.client
+
+ # OAuth mode
+ if hasattr(lifespan_ctx, "nextcloud_host"):
+ if settings.enable_token_exchange:
+ # Mode 2: Exchange MCP token for Nextcloud token
+ logger.debug("Token exchange mode - exchanging token")
+ return await get_session_client_from_context(
+ ctx, lifespan_ctx.nextcloud_host
+ )
+ else:
+ # Mode 1: Token already has both audiences, use directly
+ logger.debug("Multi-audience mode - using token directly")
+ return get_client_from_context(ctx, lifespan_ctx.nextcloud_host)
+
+ raise AttributeError("Unknown context type")
+
+
+# In context_helper.py
+async def get_session_client_from_context(
+ ctx: Context, base_url: str
+) -> NextcloudClient:
+ """
+ Create NextcloudClient using RFC 8693 token exchange.
+
+ CRITICAL: This is where token exchange happens, NOT in the verifier.
+ The verifier already validated the MCP audience; now we exchange for Nextcloud.
+ """
+ # Extract validated MCP token from context
+ access_token: AccessToken = ctx.request_context.request.user.access_token
+ mcp_token = access_token.token
+ username = access_token.resource # Username from verifier
+
+ # 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("Using cached exchanged token")
+ return NextcloudClient.from_token(
+ base_url=base_url, token=cached_token, username=username
+ )
+
+ # Perform RFC 8693 token exchange
+ logger.info("Exchanging MCP token for Nextcloud API token")
+ exchanged_token, expires_in = await exchange_token_for_audience(
+ subject_token=mcp_token,
+ requested_audience=settings.nextcloud_resource_uri,
+ requested_scopes=None, # Nextcloud doesn't enforce scopes
+ )
+
+ # Cache the exchanged token
+ _exchange_cache[cache_key] = (
+ exchanged_token,
+ time.time() + min(expires_in, settings.token_exchange_cache_ttl)
+ )
+
+ # Create client with exchanged token
+ return NextcloudClient.from_token(
+ base_url=base_url, token=exchanged_token, username=username
+ )
+
+
+def get_client_from_context(ctx: Context, base_url: str) -> NextcloudClient:
+ """
+ Create NextcloudClient for multi-audience mode (no exchange needed).
+ Token already contains both MCP and Nextcloud audiences.
+ """
+ access_token: AccessToken = ctx.request_context.request.user.access_token
+
+ # Token was already validated to have both audiences
+ # Can use directly without exchange
+ return NextcloudClient.from_token(
+ base_url=base_url,
+ token=access_token.token,
+ username=access_token.resource # Username from verifier
+ )
+```
+
+**Key Implementation Details**:
+
+1. **Token Exchange Location**: Exchange happens in `get_session_client_from_context()`, not in the verifier
+2. **Caching**: Exchange cache is maintained in the context helper to prevent repeated exchanges
+3. **Python SDK Integration**: We work with the SDK's `AccessToken` object and create `NextcloudClient` with the appropriate token
+
+### 6. Performance Benchmarks
+
+Expected performance characteristics:
+
+| Mode | Latency Impact | Use Case |
+|------|---------------|----------|
+| Multi-Audience | 0ms (no extra calls) | Default, best performance |
+| Token Exchange (cached) | ~1ms (cache lookup) | Recently used tokens |
+| Token Exchange (fresh) | 50-100ms (network call) | First use or after cache expiry |
+
+In context of LLM operations:
+- LLM inference: 2-10 seconds typical
+- Token exchange: 0.05-0.1 seconds (1-2% of total request time)
+- **Conclusion**: Performance impact is negligible
+
+### 7. IdP Configuration Examples
+
+#### Nextcloud Built-in OIDC (Multi-Audience)
+```bash
+# Set resource identifier for Nextcloud
+php occ config:app:set oidc default_resource_identifier --value="http://localhost:8080"
+
+# MCP server configuration (multi-audience mode)
+ENABLE_TOKEN_EXCHANGE=false # or omit
+NEXTCLOUD_MCP_SERVER_URL=http://localhost:8000
+NEXTCLOUD_RESOURCE_URI=http://localhost:8080
+```
+
+#### Keycloak with Multi-Audience
+```bash
+# 1. Create resource servers in Keycloak
+# Admin Console > Clients > Create Client
+# - MCP Resource Server: http://localhost:8000
+# - Nextcloud Resource Server: http://localhost:8080
+
+# 2. Configure token mapper for multi-audience
+# Client > Mappers > Create
+# - Mapper Type: Audience
+# - Included Client Audience: Select both resource servers
+
+# 3. MCP server configuration
+ENABLE_TOKEN_EXCHANGE=false # Multi-audience mode
+NEXTCLOUD_MCP_SERVER_URL=http://localhost:8000
+NEXTCLOUD_RESOURCE_URI=http://localhost:8080
+OIDC_DISCOVERY_URL=http://keycloak:8080/realms/nextcloud-mcp/.well-known/openid-configuration
+```
+
+#### Keycloak with Token Exchange
+```bash
+# 1. Enable token exchange in Keycloak
+# Realm Settings > Client Policies > Add permission for token-exchange
+
+# 2. MCP server configuration
+ENABLE_TOKEN_EXCHANGE=true # Exchange mode
+NEXTCLOUD_MCP_SERVER_URL=http://localhost:8000
+NEXTCLOUD_RESOURCE_URI=http://localhost:8080
+OIDC_DISCOVERY_URL=http://keycloak:8080/realms/nextcloud-mcp/.well-known/openid-configuration
+# Note: Token endpoint is auto-discovered from the OIDC discovery URL
+```
+
+## Testing
+
+### Unit Tests
+```python
+@pytest.mark.unit
+async def test_multi_audience_validation():
+ """Test multi-audience token validation logic."""
+ validator = UnifiedTokenVerifier(
+ nextcloud_mcp_server_url="http://localhost:8000",
+ nextcloud_resource_uri="http://localhost:8080",
+ oidc_client_id="test-client"
+ )
+
+ # Valid: Both resource URIs
+ token = {"aud": ["http://localhost:8000", "http://localhost:8080"]}
+ assert await validator.validate_token_audiences(token)
+
+ # Valid: Client ID + Nextcloud URI
+ token = {"aud": ["test-client", "http://localhost:8080"]}
+ assert await validator.validate_token_audiences(token)
+
+ # Invalid: Missing Nextcloud
+ token = {"aud": ["http://localhost:8000"]}
+ assert not await validator.validate_token_audiences(token)
+
+ # Invalid: Missing MCP
+ token = {"aud": ["http://localhost:8080"]}
+ assert not await validator.validate_token_audiences(token)
+
+@pytest.mark.unit
+async def test_token_exchange_caching():
+ """Test token exchange caching behavior."""
+ cache = TokenExchangeCache(ttl_seconds=5)
+ exchange_count = 0
+
+ async def mock_exchange(subject_token: str, requested_audience: str):
+ nonlocal exchange_count
+ exchange_count += 1
+ return f"exchanged-{exchange_count}"
+
+ # First call - should exchange
+ token1 = await cache.get_or_exchange("subject-1", "hash-1", mock_exchange)
+ assert token1 == "exchanged-1"
+ assert exchange_count == 1
+
+ # Second call with same hash - should use cache
+ token2 = await cache.get_or_exchange("subject-1", "hash-1", mock_exchange)
+ assert token2 == "exchanged-1"
+ assert exchange_count == 1 # No new exchange
+
+ # Different hash - should exchange
+ token3 = await cache.get_or_exchange("subject-2", "hash-2", mock_exchange)
+ assert token3 == "exchanged-2"
+ assert exchange_count == 2
+```
+
+### Integration Tests
+```python
+@pytest.mark.integration
+async def test_multi_audience_e2e(nc_mcp_oauth_client):
+ """Test end-to-end multi-audience token flow."""
+ # Token should have both audiences
+ result = await nc_mcp_oauth_client.call_tool("nc_notes_list_notes")
+ assert result.success
+
+ # Verify token was not exchanged (check logs)
+ logs = await get_server_logs()
+ assert "Token exchange" not in logs
+ assert "Multi-audience validation passed" in logs
+
+@pytest.mark.integration
+async def test_token_exchange_e2e(nc_mcp_keycloak_client):
+ """Test end-to-end token exchange flow."""
+ # Start with MCP-only token
+ result = await nc_mcp_keycloak_client.call_tool("nc_notes_list_notes")
+ assert result.success
+
+ # Verify exchange happened
+ logs = await get_server_logs()
+ assert "Exchanging token for Nextcloud audience" in logs
+
+ # Second call should use cache
+ result2 = await nc_mcp_keycloak_client.call_tool("nc_notes_list_notes")
+ assert result2.success
+
+ logs2 = await get_server_logs()
+ assert "Using cached exchanged token" in logs2
+
+@pytest.mark.integration
+async def test_invalid_audience_rejection(nc_mcp_oauth_client):
+ """Test that invalid audiences are rejected with clear errors."""
+ # Manually inject token with wrong audience
+ invalid_token = create_test_token(aud=["wrong-audience"])
+
+ with pytest.raises(TokenValidationError) as exc_info:
+ await nc_mcp_oauth_client.call_tool(
+ "nc_notes_list_notes",
+ token=invalid_token
+ )
+
+ assert exc_info.value.http_status == 401
+ assert "Token audiences do not meet requirements" in str(exc_info.value)
+ assert exc_info.value.details["need_nextcloud"] == "http://localhost:8080"
+```
+
+### Load Tests
+```python
+@pytest.mark.load
+async def test_token_validation_performance():
+ """Benchmark token validation overhead."""
+ # Test both modes under load
+ results = {}
+
+ for enable_exchange in [False, True]:
+ os.environ["ENABLE_TOKEN_EXCHANGE"] = str(enable_exchange).lower()
+ mode = "exchange" if enable_exchange else "multi-audience"
+
+ start = time.time()
+ await run_concurrent_requests(
+ num_workers=50,
+ requests_per_worker=100,
+ operation="nc_notes_list_notes"
+ )
+ duration = time.time() - start
+
+ results[mode] = {
+ "total_time": duration,
+ "requests_per_second": 5000 / duration,
+ "avg_latency_ms": (duration / 5000) * 1000
+ }
+
+ # Multi-audience should be faster (no exchange)
+ assert results["multi-audience"]["avg_latency_ms"] < results["exchange"]["avg_latency_ms"]
+
+ # But both should be acceptable for LLM context
+ assert results["exchange"]["avg_latency_ms"] < 200 # Max 200ms overhead
+```
+
+## Troubleshooting
+
+### Common Issues and Solutions
+
+1. **"Token audiences do not meet requirements"**
+ - Check token with jwt.io to see actual audiences
+ - Verify `NEXTCLOUD_MCP_SERVER_URL` and `NEXTCLOUD_RESOURCE_URI` match IdP configuration
+ - For Nextcloud OIDC: Check `occ config:app:get oidc default_resource_identifier`
+
+2. **"Token exchange failed"**
+ - Verify IdP supports RFC 8693 token exchange
+ - Check that OIDC discovery URL is correctly configured
+ - Verify token endpoint is accessible from the MCP server
+ - Enable debug logging: `LOG_LEVEL=DEBUG`
+
+3. **"Configuration validation failed at startup"**
+ - Ensure `ENABLE_TOKEN_EXCHANGE` is set correctly (true for exchange mode, false/omit for multi-audience)
+ - Both resource URIs must be configured (`NEXTCLOUD_MCP_SERVER_URL` and `NEXTCLOUD_RESOURCE_URI`)
+ - Check that IdP is configured to issue tokens with appropriate audiences
+
+4. **Performance issues with exchange mode**
+ - Check cache hit rate in logs
+ - Increase `TOKEN_EXCHANGE_CACHE_TTL` if tokens are long-lived
+ - Consider switching to multi-audience mode if IdP supports it
+
+### Debug Commands
+
+```bash
+# Check current token audiences (requires jq)
+echo $ACCESS_TOKEN | cut -d. -f2 | base64 -d | jq '.aud'
+
+# Test multi-audience validation
+curl -X POST http://localhost:8000/mcp/v1/tools/nc_notes_list_notes \
+ -H "Authorization: Bearer $ACCESS_TOKEN" \
+ -H "Content-Type: application/json"
+
+# Check server logs for validation details
+docker compose logs mcp-oauth | grep -E "(audience|validation|exchange)"
+
+# Verify IdP resource configuration (Keycloak)
+curl http://keycloak:8080/realms/nextcloud-mcp/.well-known/openid-configuration | jq '.resource_servers'
+```
+
+## Security Considerations
+
+### Threat Model
+
+**Threat**: Malicious client uses stolen MCP token against Nextcloud directly
+- **Mitigation**: Tokens must contain correct resource URI audiences
+- **Multi-Audience**: Requires token with both audiences (harder to obtain)
+- **Exchange**: MCP token cannot be used directly against Nextcloud
+
+**Threat**: Token reuse across services
+- **Mitigation**: Strict audience validation ensures tokens only work for intended services
+- **Validation**: Both MCP and Nextcloud validate their respective audiences
+
+**Threat**: Audit trail confusion
+- **Mitigation**: Clear separation of token contexts
+- **Multi-Audience**: Different audience claims identify service context
+- **Exchange**: Completely different tokens for each service
+
+### Compliance
+
+This implementation ensures **full compliance** with:
+- [MCP Security Best Practices - Token Passthrough](https://modelcontextprotocol.io/specification/2025-06-18/basic/security_best_practices#token-passthrough)
+- OAuth 2.0 Resource Indicators (RFC 8707)
+- OAuth 2.0 Token Exchange (RFC 8693)
+
+## Migration Guide
+
+### For Existing Deployments
+
+**BREAKING CHANGE**: All OAuth deployments must be reconfigured to comply with the new audience validation requirements.
+
+#### Step 1: Update Environment Variables
+
+Add the required resource URI configuration:
+
+```bash
+# Required for all OAuth modes
+NEXTCLOUD_MCP_SERVER_URL=http://your-mcp-server:8000 # Your MCP server URL
+NEXTCLOUD_RESOURCE_URI=http://your-nextcloud:8080 # Your Nextcloud instance URL
+```
+
+#### Step 2: Choose Your Mode
+
+**Option A: Multi-Audience Mode (Recommended for most deployments)**
+```bash
+ENABLE_TOKEN_EXCHANGE=false # or omit entirely
+```
+
+Configure your IdP to issue tokens with both audiences:
+- MCP audience: Your client ID or MCP server URL
+- Nextcloud audience: Your Nextcloud resource URI
+
+**Option B: Token Exchange Mode (For strict separation)**
+```bash
+ENABLE_TOKEN_EXCHANGE=true
+TOKEN_EXCHANGE_CACHE_TTL=300 # Optional, default is 300 seconds
+```
+
+Configure your IdP to:
+- Issue tokens with MCP audience only
+- Support RFC 8693 token exchange
+
+#### Step 3: Update IdP Configuration
+
+**For Nextcloud OIDC**:
+```bash
+# Set the resource identifier
+docker compose exec app php occ config:app:set oidc default_resource_identifier --value="http://your-nextcloud:8080"
+```
+
+**For Keycloak**:
+1. Create resource servers for both MCP and Nextcloud
+2. Configure audience mappers appropriately
+3. Enable token exchange if using exchange mode
+
+#### Step 4: Test Your Configuration
+
+```bash
+# Test multi-audience validation
+curl -X POST http://localhost:8000/mcp/v1/tools/nc_notes_list_notes \
+ -H "Authorization: Bearer $ACCESS_TOKEN" \
+ -H "Content-Type: application/json"
+
+# Check logs for validation details
+docker compose logs mcp-oauth | grep -E "(audience|validation)"
+```
+
+### Code Migration
+
+If you have custom code using the old verifiers:
+
+**Before**:
+```python
+from nextcloud_mcp_server.auth.token_verifier import NextcloudTokenVerifier
+verifier = NextcloudTokenVerifier(...)
+```
+
+**After**:
+```python
+from nextcloud_mcp_server.auth.unified_verifier import UnifiedTokenVerifier
+verifier = UnifiedTokenVerifier(settings)
+```
+
+## Consequences
+
+### Positive
+
+1. **Security Compliance**: Eliminates token passthrough vulnerability
+2. **OAuth Spec Compliance**: Follows RFC 7519 Section 4.1.3 - resource servers validate only their own audience
+3. **Clear Architecture**: Explicit validation modes with resource URI semantics
+4. **Performance**: Negligible impact in LLM context (1-2% of request time)
+5. **Flexibility**: Supports both simple (multi-audience) and strict (exchange) modes
+6. **Audit Trail**: Proper audience separation enables accurate logging
+7. **Simpler Logic**: Each resource server independently validates its own audience, reducing complexity
+
+### Negative
+
+1. **Breaking Change**: Existing deployments must reconfigure
+2. **Configuration Required**: Must specify resource URIs explicitly
+3. **IdP Requirements**: Requires proper resource server configuration
+
+### Neutral
+
+1. **Experimental Status**: Breaking changes acceptable until upstream fix merged
+2. **Performance Trade-off**: Security benefit outweighs minimal latency cost
+
+## References
+
+- [Issue #261: Avoid Token Passthrough in OAuth flow](https://github.com/cbcoutinho/nextcloud-mcp-server/issues/261)
+- [MCP Security Best Practices](https://modelcontextprotocol.io/specification/2025-06-18/basic/security_best_practices)
+- [RFC 8693: OAuth 2.0 Token Exchange](https://datatracker.ietf.org/doc/html/rfc8693)
+- [RFC 8707: Resource Indicators for OAuth 2.0](https://datatracker.ietf.org/doc/html/rfc8707)
+- [ADR-004: Federated Authentication Architecture](./ADR-004-mcp-application-oauth.md)
+- [Upstream OAuth Requirements](./upstream-oauth.md)
+
+## Python SDK Constraints and Architecture
+
+### SDK TokenVerifier Protocol
+
+The MCP python-sdk defines a strict `TokenVerifier` protocol that our implementation must follow:
+
+```python
+class TokenVerifier(Protocol):
+ async def verify_token(self, token: str) -> AccessToken | None:
+ """Verify a bearer token and return access info if valid."""
+```
+
+**Key Constraints**:
+
+1. **Single Method Interface**: The verifier can only validate tokens, not modify or exchange them
+2. **Return Type**: Must return an `AccessToken` object or `None`
+3. **Token Access**: The original bearer token is passed through the SDK to API calls unless we intervene at a different layer
+
+### Architecture Decisions
+
+Given these constraints, we implement a **two-layer architecture**:
+
+1. **Token Verifier Layer** (`UnifiedTokenVerifier`):
+ - Validates token audiences according to configured mode
+ - Returns `AccessToken` objects to satisfy SDK protocol
+ - Does NOT perform token exchange
+
+2. **Context Helper Layer** (`context_helper.py`):
+ - Extracts tokens from MCP context
+ - Performs RFC 8693 token exchange when needed
+ - Creates `NextcloudClient` with appropriate token
+ - Maintains exchange cache to minimize latency
+
+This separation ensures:
+- Compliance with MCP SDK protocol
+- Clean separation of concerns
+- Token exchange happens only when creating API clients
+- Pass-through vulnerability is eliminated
+
+## Implementation Checklist
+
+- [x] Create `UnifiedTokenVerifier` class replacing both existing verifiers
+- [x] Remove pass-through mode from `context.py` entirely
+- [x] Update `context_helper.py` to implement token exchange with caching
+- [x] Implement RFC 7519 compliant validation in unified verifier (MCP audience only)
+- [x] Add token exchange caching mechanism in context helper layer
+- [x] Add OAuth resource parameters to authorization requests (RFC 8707)
+- [x] Implement PRM endpoint for resource discovery (RFC 9728)
+- [x] Update tests to discover resource from PRM endpoint
+- [x] Fix Nextcloud OIDC app to use client-specific resource_url
+- [x] Update docker-compose.yml with resource URI configuration:
+ - `NEXTCLOUD_MCP_SERVER_URL` (required)
+ - `NEXTCLOUD_RESOURCE_URI` (required)
+ - `TOKEN_EXCHANGE_CACHE_TTL` (optional, default: 300)
+- [x] Configure Nextcloud OIDC `default_resource_identifier`
+- [ ] Configure Keycloak resource servers with proper audiences
+- [x] Remove `NextcloudTokenVerifier` class
+- [x] Remove `ProgressiveConsentTokenVerifier` class
+- [x] Write unit tests for unified verifier
+- [x] Write integration tests for OAuth flows
+- [x] Update documentation with IdP configuration guides
+- [ ] Add performance benchmarks to CI pipeline
+- [ ] Update CHANGELOG.md with breaking changes notice
\ No newline at end of file
diff --git a/docs/ADR-006-progressive-consent-elicitation.md b/docs/ADR-006-progressive-consent-elicitation.md
new file mode 100644
index 00000000..d39a4738
--- /dev/null
+++ b/docs/ADR-006-progressive-consent-elicitation.md
@@ -0,0 +1,651 @@
+# ADR-006: Progressive Consent via URL Elicitation (SEP-1036)
+
+**Status**: Proposed
+**Date**: 2025-01-05
+**Related**: [SEP-1036](https://github.com/modelcontextprotocol/specification/pull/887), ADR-004
+**Depends On**: ADR-005 (token validation)
+
+## Context
+
+The current progressive consent implementation (ADR-004) requires users to manually visit OAuth URLs returned by MCP tools. This creates a poor user experience:
+
+1. User calls `provision_nextcloud_access` tool
+2. Tool returns a URL as text in the response
+3. User must manually copy URL and open in browser
+4. No indication when provisioning is complete
+5. User must retry the original operation manually
+
+### SEP-1036: URL Mode Elicitation
+
+The MCP specification now supports **URL mode elicitation** ([SEP-1036](https://github.com/modelcontextprotocol/specification/pull/887)), which enables servers to:
+
+- Request out-of-band user interactions via secure URLs
+- Handle sensitive operations like OAuth flows without exposing credentials to the client
+- Provide progress tracking for async operations
+- Return errors that automatically trigger elicitation flows
+
+**Key benefits for progressive consent**:
+- **Automatic URL Opening**: Client opens URL in browser automatically (with user consent)
+- **Progress Tracking**: Server can notify client when provisioning is complete
+- **Error-Triggered Flows**: Server can return `ElicitationRequired` error to trigger provisioning
+- **Better UX**: User doesn't manually copy/paste URLs
+
+### Current Implementation Limitations
+
+The current progressive consent flow in `nextcloud_mcp_server/server/oauth_tools.py`:
+
+```python
+@mcp.tool(name="provision_nextcloud_access")
+async def tool_provision_access(ctx: Context) -> ProvisioningResult:
+ """Returns OAuth URL as text - user must manually open it."""
+ return ProvisioningResult(
+ success=True,
+ authorization_url=auth_url, # User must copy this
+ message="Please visit the authorization URL..."
+ )
+```
+
+**Problems**:
+1. Manual URL handling (copy/paste)
+2. No progress tracking
+3. No automatic retry after provisioning
+4. Tool call required just to get URL
+5. No client integration (URL just displayed as text)
+
+## Decision
+
+We will **migrate progressive consent from manual tools to URL mode elicitation**, leveraging SEP-1036 for better user experience and OAuth security.
+
+### New Architecture: Elicitation-Driven Consent
+
+Instead of explicit tools, use **automatic elicitation** triggered by authorization errors:
+
+```
+User → Calls Nextcloud Tool → Server Checks Provisioning
+ ↓ Not Provisioned
+ Error: ElicitationRequired
+ ↓
+ Client Shows Consent UI
+ ↓ User Accepts
+ Client Opens OAuth URL
+ ↓
+ User Completes OAuth
+ ↓
+ Server Sends Progress Update
+ ↓
+ Original Tool Call Auto-Retries
+```
+
+### Mode 1: Elicitation-Required Error (Primary)
+
+When a tool requires provisioning, return an **ElicitationRequired error** (-32000):
+
+```python
+# In any Nextcloud tool decorated with @require_provisioning
+@mcp.tool()
+@require_provisioning # New decorator
+async def nc_notes_list_notes(ctx: Context):
+ """List notes - auto-triggers provisioning if needed."""
+ # If not provisioned, decorator returns ElicitationRequired error
+ # If provisioned, continues normally
+ client = await get_client(ctx)
+ return await client.notes.list_notes()
+```
+
+**Error response structure**:
+```json
+{
+ "jsonrpc": "2.0",
+ "id": 1,
+ "error": {
+ "code": -32000,
+ "message": "Nextcloud access provisioning required",
+ "data": {
+ "elicitations": [
+ {
+ "mode": "url",
+ "elicitationId": "550e8400-e29b-41d4-a716-446655440000",
+ "url": "https://mcp.example.com/oauth/provision?id=550e8400...",
+ "message": "Grant the MCP server access to your Nextcloud account to continue."
+ }
+ ]
+ }
+ }
+}
+```
+
+**Client behavior**:
+1. Receives error with elicitation
+2. Shows consent UI: "App wants to access Nextcloud. Open authorization page?"
+3. On user acceptance, opens URL in browser
+4. Optionally tracks progress via `elicitation/track`
+5. Auto-retries original tool call when complete
+
+### Mode 2: Explicit Elicitation Request (Fallback)
+
+For clients that don't support error-triggered elicitation, provide explicit tool:
+
+```python
+@mcp.tool(name="request_nextcloud_access")
+async def request_access(ctx: Context) -> ElicitationResponse:
+ """Explicitly request provisioning via elicitation."""
+ # Send elicitation/create request
+ return await create_elicitation(
+ mode="url",
+ url=generate_oauth_url(),
+ message="Grant access to Nextcloud",
+ elicitation_id=generate_id()
+ )
+```
+
+**Note**: This is a fallback for compatibility. Primary flow uses error-triggered elicitation.
+
+## Implementation
+
+### 1. New Decorator: `@require_provisioning`
+
+Replace explicit provisioning checks with a decorator that returns `ElicitationRequired`:
+
+```python
+# nextcloud_mcp_server/auth/provisioning_decorator.py
+
+def require_provisioning(func):
+ """
+ Decorator that ensures user has provisioned Nextcloud access.
+
+ If not provisioned, returns ElicitationRequired error with OAuth URL.
+ Otherwise, proceeds with normal tool execution.
+ """
+ @functools.wraps(func)
+ async def wrapper(ctx: Context, *args, **kwargs):
+ # Extract user ID from token
+ user_id = get_user_id_from_context(ctx)
+
+ # Check if provisioned
+ storage = RefreshTokenStorage.from_env()
+ await storage.initialize()
+
+ if not await storage.has_refresh_token(user_id):
+ # Not provisioned - return ElicitationRequired error
+ elicitation_id = str(uuid.uuid4())
+ oauth_url = await generate_oauth_url_for_provisioning(
+ user_id=user_id,
+ elicitation_id=elicitation_id,
+ ctx=ctx
+ )
+
+ # Store elicitation for tracking
+ await storage.store_elicitation(
+ elicitation_id=elicitation_id,
+ user_id=user_id,
+ status="pending",
+ created_at=datetime.now(timezone.utc)
+ )
+
+ raise McpError(
+ code=ErrorCode.ELICITATION_REQUIRED, # -32000
+ message="Nextcloud access provisioning required",
+ data={
+ "elicitations": [
+ {
+ "mode": "url",
+ "elicitationId": elicitation_id,
+ "url": oauth_url,
+ "message": (
+ "Grant the MCP server access to your Nextcloud "
+ "account to continue. This is a one-time setup."
+ )
+ }
+ ]
+ }
+ )
+
+ # Already provisioned - proceed normally
+ return await func(ctx, *args, **kwargs)
+
+ return wrapper
+```
+
+### 2. Elicitation Tracking Endpoint
+
+Implement `elicitation/track` to provide progress updates:
+
+```python
+# nextcloud_mcp_server/server/elicitation.py
+
+@mcp.request_handler("elicitation/track")
+async def track_elicitation(
+ elicitation_id: str,
+ _meta: dict = None
+) -> dict:
+ """
+ Track progress of an elicitation request.
+
+ Returns when elicitation is complete or times out.
+ """
+ progress_token = _meta.get("progressToken") if _meta else None
+
+ storage = RefreshTokenStorage.from_env()
+ await storage.initialize()
+
+ # Poll for completion (with timeout)
+ timeout = 300 # 5 minutes
+ start_time = datetime.now(timezone.utc)
+
+ while (datetime.now(timezone.utc) - start_time).seconds < timeout:
+ elicitation = await storage.get_elicitation(elicitation_id)
+
+ if not elicitation:
+ raise McpError(
+ code=-32602, # Invalid params
+ message=f"Unknown elicitation ID: {elicitation_id}"
+ )
+
+ # Send progress notification if token provided
+ if progress_token and elicitation["status"] == "pending":
+ await send_progress_notification(
+ progress_token=progress_token,
+ progress=50,
+ message="Waiting for OAuth authorization..."
+ )
+
+ # Check if complete
+ if elicitation["status"] == "complete":
+ return {"status": "complete"}
+
+ # Check if failed
+ if elicitation["status"] == "failed":
+ return {
+ "status": "failed",
+ "error": elicitation.get("error_message")
+ }
+
+ # Wait before polling again
+ await asyncio.sleep(2)
+
+ # Timeout
+ raise McpError(
+ code=-32000,
+ message="Elicitation timed out - user did not complete authorization"
+ )
+```
+
+### 3. OAuth Callback Updates
+
+Update the OAuth callback to mark elicitations as complete:
+
+```python
+# nextcloud_mcp_server/auth/oauth_routes.py
+
+async def oauth_callback(request: Request) -> Response:
+ """Handle OAuth callback and mark elicitation complete."""
+ code = request.query_params.get("code")
+ state = request.query_params.get("state")
+
+ # Validate and exchange code for tokens
+ tokens = await exchange_authorization_code(code)
+
+ # Store refresh token
+ await storage.store_refresh_token(
+ user_id=user_id,
+ refresh_token=tokens["refresh_token"]
+ )
+
+ # Mark elicitation as complete
+ elicitation_id = request.query_params.get("elicitation_id")
+ if elicitation_id:
+ await storage.update_elicitation(
+ elicitation_id=elicitation_id,
+ status="complete",
+ completed_at=datetime.now(timezone.utc)
+ )
+
+ return Response(
+ content="Authorization Complete!
"
+ "You can close this window and return to the application.
",
+ media_type="text/html"
+ )
+```
+
+### 4. Update All Nextcloud Tools
+
+Add `@require_provisioning` decorator to all Nextcloud tools:
+
+```python
+# nextcloud_mcp_server/server/notes.py
+
+@mcp.tool()
+@require_scopes("notes:read")
+@require_provisioning # NEW: Auto-triggers provisioning
+async def nc_notes_list_notes(
+ ctx: Context,
+ category: Optional[str] = None
+) -> NotesListResponse:
+ """List all notes - automatically handles provisioning."""
+ client = await get_client(ctx)
+ # Tool logic proceeds only if provisioned
+ notes = await client.notes.list_notes(category=category)
+ return NotesListResponse(results=notes)
+```
+
+### 5. Capability Declaration
+
+Declare URL elicitation support during initialization:
+
+```python
+# nextcloud_mcp_server/app.py
+
+capabilities = {
+ "elicitation": {
+ "url": {} # Declare URL mode support
+ # Note: We don't support "form" mode (in-band data collection)
+ },
+ # ... other capabilities
+}
+```
+
+### 6. Environment Variables
+
+**New variables**:
+```bash
+# ELICITATION_CALLBACK_URL: Base URL for OAuth callbacks with elicitation tracking
+# Default: NEXTCLOUD_MCP_SERVER_URL + /oauth/callback
+ELICITATION_CALLBACK_URL=http://localhost:8000/oauth/callback
+
+# ELICITATION_TIMEOUT_SECONDS: How long to wait for user to complete OAuth
+# Default: 300 (5 minutes)
+ELICITATION_TIMEOUT_SECONDS=300
+```
+
+**Removed variables** (no longer needed):
+```bash
+# ENABLE_PROGRESSIVE_CONSENT - removed, now always enabled in OAuth mode
+# MCP_SERVER_CLIENT_ID - merged into OIDC_CLIENT_ID
+```
+
+## User Experience Comparison
+
+### Before (ADR-004 Manual Tools)
+
+```
+User: "List my notes"
+Assistant: *calls nc_notes_list_notes*
+Server: Error - not provisioned
+Assistant: "You need to provision access first. Let me do that."
+Assistant: *calls provision_nextcloud_access*
+Server: {authorization_url: "https://..."}
+Assistant: "Please visit this URL: https://..."
+User: *copies URL, opens browser, completes OAuth*
+User: "OK, I'm done"
+Assistant: *calls nc_notes_list_notes again*
+Server: Success! [notes...]
+```
+
+**Issues**: 4 interactions, manual URL handling, no automation
+
+### After (ADR-006 Elicitation)
+
+```
+User: "List my notes"
+Assistant: *calls nc_notes_list_notes*
+Server: ElicitationRequired error
+Client: Shows dialog: "Grant access to Nextcloud? [Yes] [No]"
+User: *clicks Yes*
+Client: Opens OAuth URL in browser automatically
+User: *completes OAuth*
+Server: Sends progress notification "Complete!"
+Client: Auto-retries nc_notes_list_notes
+Server: Success! [notes...]
+Assistant: "Here are your notes: ..."
+```
+
+**Benefits**: 1 interaction, automatic URL opening, seamless retry
+
+## Migration Path
+
+### Phase 1: Add Elicitation Support (v0.26.0)
+
+- Implement `@require_provisioning` decorator
+- Add `elicitation/track` endpoint
+- Keep existing tools (`provision_nextcloud_access`) for compatibility
+- Update OAuth callback to track elicitations
+- Add capability declaration
+
+**Breaking changes**: None (additive)
+
+### Phase 2: Update Documentation (v0.27.0)
+
+- Document elicitation-based flow as primary
+- Mark manual tools as deprecated
+- Update examples and guides
+
+**Breaking changes**: None (documentation only)
+
+### Phase 3: Remove Manual Tools (v0.28.0)
+
+- Remove `provision_nextcloud_access` tool
+- Remove `check_provisioning_status` tool (status in error message)
+- Remove `revoke_nextcloud_access` (or keep for explicit revocation?)
+
+**Breaking changes**: Yes (removed tools)
+
+### Phase 4: Optimize (v0.29.0+)
+
+- Add elicitation result caching
+- Implement retry strategies
+- Add metrics and monitoring
+
+## Testing
+
+### Test Cases
+
+1. **First-Time User Flow**
+ ```python
+ @pytest.mark.oauth
+ async def test_elicitation_first_time_user(nc_mcp_oauth_client):
+ """Test that first tool call triggers elicitation."""
+ # User has no provisioning
+ with pytest.raises(McpError) as exc:
+ await nc_mcp_oauth_client.call_tool("nc_notes_list_notes")
+
+ # Should get ElicitationRequired error
+ assert exc.value.code == -32000
+ assert "elicitations" in exc.value.data
+ assert exc.value.data["elicitations"][0]["mode"] == "url"
+
+ # Verify URL is valid OAuth URL
+ url = exc.value.data["elicitations"][0]["url"]
+ assert "oauth" in url
+ assert "elicitationId" in url
+ ```
+
+2. **Progress Tracking**
+ ```python
+ @pytest.mark.oauth
+ async def test_elicitation_progress_tracking(nc_mcp_oauth_client):
+ """Test progress tracking during OAuth flow."""
+ # Trigger elicitation
+ elicitation_id = trigger_elicitation()
+
+ # Start tracking
+ track_task = asyncio.create_task(
+ nc_mcp_oauth_client.track_elicitation(
+ elicitation_id=elicitation_id,
+ progress_token="test-token"
+ )
+ )
+
+ # Simulate OAuth completion
+ await asyncio.sleep(1)
+ await complete_oauth_flow(elicitation_id)
+
+ # Track should complete
+ result = await track_task
+ assert result["status"] == "complete"
+ ```
+
+3. **Auto-Retry After Provisioning**
+ ```python
+ @pytest.mark.oauth
+ async def test_auto_retry_after_provisioning(nc_mcp_oauth_client):
+ """Test that client auto-retries after elicitation."""
+ # Mock client that auto-retries on ElicitationRequired
+ client = AutoRetryMcpClient(nc_mcp_oauth_client)
+
+ # First call triggers elicitation, client handles it, retries
+ result = await client.call_tool_with_elicitation("nc_notes_list_notes")
+
+ # Should succeed after provisioning
+ assert result.success
+ assert "notes" in result.data
+ ```
+
+4. **Timeout Handling**
+ ```python
+ @pytest.mark.oauth
+ async def test_elicitation_timeout(nc_mcp_oauth_client):
+ """Test timeout if user doesn't complete OAuth."""
+ elicitation_id = trigger_elicitation()
+
+ # Track with short timeout
+ with pytest.raises(McpError, match="timed out"):
+ await nc_mcp_oauth_client.track_elicitation(
+ elicitation_id=elicitation_id,
+ timeout=5 # 5 seconds
+ )
+ ```
+
+## Security Considerations
+
+### Out-of-Band OAuth Flow
+
+**Benefit**: OAuth credentials never pass through MCP client
+- User enters credentials directly on IdP page
+- MCP server receives only authorization code
+- Client never sees passwords or refresh tokens
+
+**Threat mitigation**:
+- **Credential theft**: Client can't intercept credentials (out-of-band)
+- **Token exposure**: Client never receives Nextcloud refresh tokens
+- **CSRF**: State parameter validates OAuth callback
+- **URL tampering**: Elicitation ID ties OAuth flow to user session
+
+### Elicitation ID as Security Token
+
+The `elicitationId` serves as a capability token:
+- Cryptographically random (UUID v4)
+- Single-use (invalidated after completion)
+- Time-limited (expires after timeout)
+- User-scoped (tied to user session)
+
+**Validation**:
+```python
+async def validate_elicitation_id(elicitation_id: str, user_id: str) -> bool:
+ """Validate that elicitation belongs to user and is still valid."""
+ elicitation = await storage.get_elicitation(elicitation_id)
+
+ if not elicitation:
+ return False
+
+ # Check ownership
+ if elicitation["user_id"] != user_id:
+ logger.warning(f"Elicitation ID mismatch: {elicitation_id}")
+ return False
+
+ # Check expiry
+ if elicitation["expires_at"] < datetime.now(timezone.utc):
+ return False
+
+ # Check not already used
+ if elicitation["status"] != "pending":
+ return False
+
+ return True
+```
+
+### Progress Tracking Security
+
+**Risk**: Progress token reuse across users
+
+**Mitigation**:
+- Progress tokens tied to elicitation ID
+- Elicitation ID tied to user session
+- Server validates ownership before sending updates
+
+## Consequences
+
+### Positive
+
+1. **Better UX**: Automatic URL opening, no manual copy/paste
+2. **Seamless Flow**: Auto-retry after provisioning
+3. **Progress Feedback**: User knows when OAuth is complete
+4. **Spec Compliance**: Implements SEP-1036 correctly
+5. **Secure by Design**: Out-of-band OAuth prevents credential exposure
+6. **Simpler API**: No explicit provisioning tools needed
+
+### Negative
+
+1. **Client Dependency**: Requires client support for URL elicitation
+2. **Complexity**: More moving parts (elicitation tracking, callbacks)
+3. **Polling**: Progress tracking uses polling (not ideal)
+4. **Breaking Change**: Removes manual provisioning tools (in v0.28.0)
+
+### Neutral
+
+1. **Storage Requirements**: Need to store elicitation state
+2. **Timeout Management**: Must handle long-running OAuth flows
+3. **Fallback Support**: Still need compatibility for older clients
+
+## Alternatives Considered
+
+### 1. Keep Manual Tools Only (Rejected)
+
+**Pros**: Simple, no client changes needed
+**Cons**: Poor UX, doesn't leverage SEP-1036
+
+**Rejection reason**: SEP-1036 provides better UX and security
+
+### 2. Form Mode Elicitation (Rejected)
+
+**Pros**: No browser redirect needed
+**Cons**: Would expose OAuth credentials to client (security violation)
+
+**Rejection reason**: Form mode only for non-sensitive data per SEP-1036
+
+### 3. Hybrid: Both Tools and Elicitation (Considered)
+
+**Pros**: Maximum compatibility, gradual migration
+**Cons**: API duplication, maintenance burden, confusing for users
+
+**Decision**: Support during migration (v0.26-0.27), remove in v0.28
+
+### 4. WebSocket for Progress (Rejected)
+
+**Pros**: Real-time updates instead of polling
+**Cons**: MCP spec uses polling pattern, adds complexity
+
+**Rejection reason**: Follow spec pattern (polling via elicitation/track)
+
+## References
+
+- [SEP-1036: URL Mode Elicitation](https://github.com/modelcontextprotocol/specification/pull/887)
+- [MCP Elicitation Specification](https://modelcontextprotocol.io/specification/draft/client/elicitation)
+- [ADR-004: Federated Authentication Architecture](./ADR-004-mcp-application-oauth.md)
+- [ADR-005: Token Audience Validation](./ADR-005-token-audience-validation.md)
+- [RFC 8252: OAuth 2.0 for Native Apps](https://datatracker.ietf.org/doc/html/rfc8252)
+
+## Implementation Checklist
+
+- [ ] Implement `@require_provisioning` decorator with ElicitationRequired error
+- [ ] Add `elicitation/track` request handler
+- [ ] Update OAuth callback to mark elicitations complete
+- [ ] Add elicitation storage (ID, user, status, timestamps)
+- [ ] Update all Nextcloud tools with `@require_provisioning`
+- [ ] Add URL elicitation capability declaration
+- [ ] Write integration tests for elicitation flow
+- [ ] Write tests for progress tracking
+- [ ] Update documentation with elicitation examples
+- [ ] Add migration guide for manual tools → elicitation
+- [ ] Keep manual tools with deprecation warnings (v0.26-0.27)
+- [ ] Remove manual tools (v0.28.0)
+- [ ] Update CHANGELOG.md with migration timeline
diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py
index 227deb38..71fb6185 100644
--- a/nextcloud_mcp_server/app.py
+++ b/nextcloud_mcp_server/app.py
@@ -27,9 +27,7 @@ from nextcloud_mcp_server.auth import (
has_required_scopes,
is_jwt_token,
)
-from nextcloud_mcp_server.auth.progressive_token_verifier import (
- ProgressiveConsentTokenVerifier,
-)
+from nextcloud_mcp_server.auth.unified_verifier import UnifiedTokenVerifier
from nextcloud_mcp_server.client import NextcloudClient
from nextcloud_mcp_server.config import (
LOGGING_CONFIG,
@@ -215,9 +213,7 @@ class OAuthAppContext:
"""Application context for OAuth mode."""
nextcloud_host: str
- token_verifier: (
- object # Can be NextcloudTokenVerifier or ProgressiveConsentTokenVerifier
- )
+ token_verifier: object # UnifiedTokenVerifier (ADR-005 compliant)
refresh_token_storage: Optional["RefreshTokenStorage"] = None
oauth_client: Optional[object] = None # NextcloudOAuthClient or KeycloakOAuthClient
oauth_provider: str = "nextcloud" # "nextcloud" or "keycloak"
@@ -555,46 +551,80 @@ async def setup_oauth_config():
else:
client_issuer = issuer
- # Progressive Consent mode (always enabled) - dual OAuth flows with audience separation
- logger.info("✓ Progressive Consent mode enabled - dual OAuth flows active")
+ # ADR-005: Unified Token Verifier with proper audience validation
+ # Get MCP server URL for audience validation
+ mcp_server_url = os.getenv("NEXTCLOUD_MCP_SERVER_URL", "http://localhost:8000")
+ nextcloud_resource_uri = os.getenv("NEXTCLOUD_RESOURCE_URI", nextcloud_host)
- # Get encryption key for token broker
- encryption_key = os.getenv("TOKEN_ENCRYPTION_KEY")
- if not encryption_key:
+ # Warn if resource URIs are not configured (required for ADR-005 compliance)
+ if not os.getenv("NEXTCLOUD_MCP_SERVER_URL"):
logger.warning(
- "TOKEN_ENCRYPTION_KEY not set - token broker will not be available"
+ f"NEXTCLOUD_MCP_SERVER_URL not set, defaulting to: {mcp_server_url}. "
+ "This should be set explicitly for proper audience validation."
+ )
+ if not os.getenv("NEXTCLOUD_RESOURCE_URI"):
+ logger.warning(
+ f"NEXTCLOUD_RESOURCE_URI not set, defaulting to: {nextcloud_resource_uri}. "
+ "This should be set explicitly for proper audience validation."
)
- # Create token broker service
- from nextcloud_mcp_server.auth.token_broker import TokenBrokerService
+ # Create settings for UnifiedTokenVerifier
+ from nextcloud_mcp_server.config import get_settings
- token_broker = None
- if encryption_key and refresh_token_storage:
- token_broker = TokenBrokerService(
- storage=refresh_token_storage,
- oidc_discovery_url=discovery_url,
- nextcloud_host=nextcloud_host,
- encryption_key=encryption_key,
+ settings = get_settings()
+ # Override with discovered values if not set in environment
+ if not settings.oidc_client_id:
+ settings.oidc_client_id = client_id
+ if not settings.oidc_client_secret:
+ settings.oidc_client_secret = client_secret
+ if not settings.jwks_uri:
+ settings.jwks_uri = jwks_uri
+ if not settings.introspection_uri:
+ settings.introspection_uri = introspection_uri
+ if not settings.userinfo_uri:
+ settings.userinfo_uri = userinfo_uri
+ if not settings.oidc_issuer:
+ # Use client_issuer which handles public URL override
+ settings.oidc_issuer = client_issuer
+ if not settings.nextcloud_mcp_server_url:
+ settings.nextcloud_mcp_server_url = mcp_server_url
+ if not settings.nextcloud_resource_uri:
+ settings.nextcloud_resource_uri = nextcloud_resource_uri
+
+ # Create Unified Token Verifier (ADR-005 compliant)
+ 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("✓ Token Broker service initialized for audience-specific tokens")
+ logger.info(f" MCP audience: {client_id} or {mcp_server_url}")
+ logger.info(f" Nextcloud audience: {nextcloud_resource_uri}")
+ else:
+ logger.info(
+ "✓ Multi-audience mode enabled (ADR-005) - tokens must contain both MCP and Nextcloud audiences"
+ )
+ logger.info(f" Required MCP audience: {client_id} or {mcp_server_url}")
+ logger.info(f" Required Nextcloud audience: {nextcloud_resource_uri}")
- # Create Progressive Consent token verifier
- token_verifier = ProgressiveConsentTokenVerifier(
- token_storage=refresh_token_storage,
- token_broker=token_broker,
- oidc_discovery_url=discovery_url,
- nextcloud_host=nextcloud_host,
- encryption_key=encryption_key,
- mcp_client_id=client_id,
- introspection_uri=introspection_uri,
- client_secret=client_secret,
- )
-
- logger.info(
- "✓ Progressive Consent verifier configured - enforcing audience separation"
- )
if introspection_uri:
logger.info("✓ Opaque token introspection enabled (RFC 7662)")
+ if jwks_uri:
+ logger.info("✓ JWT signature verification enabled (JWKS)")
+
+ # Progressive Consent mode (for offline access / background jobs)
+ encryption_key = os.getenv("TOKEN_ENCRYPTION_KEY")
+ if enable_offline_access and encryption_key and refresh_token_storage:
+ logger.info("✓ Progressive Consent mode enabled - offline access available")
+
+ # Note: Token Broker service would be initialized here for background job support
+ # Currently not used in ADR-005 implementation as it's specific to offline access patterns
+ # 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
@@ -837,6 +867,9 @@ def get_app(transport: str = "sse", enabled_apps: list[str] | None = None):
mcp_server_url = os.getenv(
"NEXTCLOUD_MCP_SERVER_URL", "http://localhost:8000"
)
+ nextcloud_resource_uri = os.getenv(
+ "NEXTCLOUD_RESOURCE_URI", nextcloud_host
+ )
discovery_url = os.getenv(
"OIDC_DISCOVERY_URL",
f"{nextcloud_host}/.well-known/openid-configuration",
@@ -854,6 +887,7 @@ def get_app(transport: str = "sse", enabled_apps: list[str] | None = None):
"client_secret": client_secret, # From setup_oauth_config (DCR or static)
"scopes": scopes,
"nextcloud_host": nextcloud_host,
+ "nextcloud_resource_uri": nextcloud_resource_uri,
"oauth_provider": oauth_provider,
},
}
diff --git a/nextcloud_mcp_server/auth/__init__.py b/nextcloud_mcp_server/auth/__init__.py
index dbb34f54..3795fe5b 100644
--- a/nextcloud_mcp_server/auth/__init__.py
+++ b/nextcloud_mcp_server/auth/__init__.py
@@ -14,11 +14,11 @@ from .scope_authorization import (
is_jwt_token,
require_scopes,
)
-from .token_verifier import NextcloudTokenVerifier
+from .unified_verifier import UnifiedTokenVerifier
__all__ = [
"BearerAuth",
- "NextcloudTokenVerifier",
+ "UnifiedTokenVerifier",
"register_client",
"ensure_oauth_client",
"get_client_from_context",
diff --git a/nextcloud_mcp_server/auth/context_helper.py b/nextcloud_mcp_server/auth/context_helper.py
index a9640537..6c8049e7 100644
--- a/nextcloud_mcp_server/auth/context_helper.py
+++ b/nextcloud_mcp_server/auth/context_helper.py
@@ -1,6 +1,11 @@
-"""Helper functions for extracting OAuth context from MCP requests."""
+"""Helper functions for extracting OAuth context from MCP requests.
+ADR-005 compliant implementation with token exchange caching.
+"""
+
+import hashlib
import logging
+import time
from mcp.server.auth.provider import AccessToken
from mcp.server.fastmcp import Context
@@ -11,35 +16,36 @@ 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:
"""
- Extract authenticated user context from MCP request and create NextcloudClient.
+ Create NextcloudClient for multi-audience mode (no exchange needed).
- This function retrieves the OAuth access token from the MCP context,
- extracts the username from the token's resource field (where we stored it
- during token verification), and creates a NextcloudClient with bearer auth.
+ ADR-005 Mode 1: Use multi-audience tokens directly.
+ The UnifiedTokenVerifier validated MCP audience per RFC 7519.
+ Nextcloud will independently validate its own audience.
Args:
ctx: MCP request context containing session info
base_url: Nextcloud base URL
Returns:
- NextcloudClient configured with bearer token auth
+ NextcloudClient configured with multi-audience token
Raises:
AttributeError: If context doesn't contain expected OAuth session data
ValueError: If username cannot be extracted from token
"""
try:
- # In Starlette with FastMCP OAuth, the authenticated user info is stored in request.user
- # The FastMCP auth middleware sets request.user to an AuthenticatedUser object
- # which contains the access_token
+ # Extract validated access token from MCP 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
- logger.debug("Retrieved access token from request.user for OAuth request")
+ logger.debug("Retrieved multi-audience token from request.user")
else:
logger.error(
"OAuth authentication failed: No access token found in request"
@@ -47,16 +53,20 @@ def get_client_from_context(ctx: Context, base_url: str) -> NextcloudClient:
raise AttributeError("No access token found in OAuth request context")
# Extract username from resource field (RFC 8707)
- # We stored the username here during token verification
+ # UnifiedTokenVerifier stored the username here during validation
username = access_token.resource
if not username:
logger.error("No username found in access token resource field")
raise ValueError("Username not available in OAuth token context")
- logger.debug(f"Creating OAuth NextcloudClient for user: {username}")
+ logger.debug(
+ f"Creating NextcloudClient for user {username} with multi-audience token "
+ f"(no exchange needed)"
+ )
- # Create client with bearer token
+ # Token was validated to have MCP audience
+ # Nextcloud will validate its own audience independently
return NextcloudClient.from_token(
base_url=base_url, token=access_token.token, username=username
)
@@ -71,12 +81,19 @@ async def get_session_client_from_context(
ctx: Context, base_url: str
) -> NextcloudClient:
"""
- Create NextcloudClient using RFC 8693 token exchange for session operations.
+ 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 Flow 1 token from context (aud: "mcp-server")
- 2. Exchange it for ephemeral Nextcloud token via RFC 8693
- 3. Create client with delegated token (NOT stored)
+ 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,
@@ -88,7 +105,7 @@ async def get_session_client_from_context(
base_url: Nextcloud base URL
Returns:
- NextcloudClient configured with ephemeral delegated token
+ NextcloudClient configured with ephemeral exchanged token
Raises:
AttributeError: If context doesn't contain expected OAuth session data
@@ -96,43 +113,60 @@ async def get_session_client_from_context(
"""
settings = get_settings()
- # Check if token exchange is enabled
- if not settings.enable_token_exchange:
- logger.info("Token exchange disabled, falling back to standard OAuth flow")
- return get_client_from_context(ctx, base_url)
-
try:
- # Extract Flow 1 token from context
+ # 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
- flow1_token = access_token.token
- username = access_token.resource # Username stored during verification
- logger.debug(f"Retrieved Flow 1 token for user: {username}")
+ 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 Flow 1 token found in request context")
+ 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")
- logger.info("Exchanging client token for Nextcloud API token (pure RFC 8693)")
+ # 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)"
+ )
+ 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]
- # Perform pure RFC 8693 token exchange (no refresh tokens)
- # Note: We don't pass scopes since Nextcloud doesn't enforce them.
- # The MCP server's @require_scopes decorator handles authorization.
+ # Perform RFC 8693 token exchange
+ logger.info(f"Exchanging MCP token for Nextcloud API token (user: {username})")
+
+ # Exchange for Nextcloud resource URI audience
exchanged_token, expires_in = await exchange_token_for_audience(
- subject_token=flow1_token,
- requested_audience="nextcloud",
+ subject_token=mcp_token,
+ requested_audience=settings.nextcloud_resource_uri or "nextcloud",
requested_scopes=None, # Nextcloud doesn't support scopes
)
- logger.info(f"Pure token exchange successful. Token expires in {expires_in}s")
+ logger.info(f"Token exchange successful. Token expires in {expires_in}s")
+
+ # 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
- # This token is ephemeral (per-request) and NOT stored
return NextcloudClient.from_token(
base_url=base_url, token=exchanged_token, username=username
)
@@ -143,3 +177,21 @@ async def get_session_client_from_context(
except Exception as e:
logger.error(f"Token exchange failed: {e}")
raise RuntimeError(f"Token exchange required but failed: {e}") from e
+
+
+def _cleanup_exchange_cache():
+ """Remove expired entries from the token exchange cache."""
+ global _exchange_cache
+ now = time.time()
+ expired_keys = [k for k, (_, expiry) in _exchange_cache.items() if expiry <= now]
+ for key in expired_keys:
+ del _exchange_cache[key]
+ if expired_keys:
+ logger.debug(f"Cleaned up {len(expired_keys)} expired cache entries")
+
+
+def clear_exchange_cache():
+ """Clear the entire token exchange cache. Useful for testing."""
+ global _exchange_cache
+ _exchange_cache.clear()
+ logger.debug("Token exchange cache cleared")
diff --git a/nextcloud_mcp_server/auth/oauth_routes.py b/nextcloud_mcp_server/auth/oauth_routes.py
index c2afc6fe..5f35f02f 100644
--- a/nextcloud_mcp_server/auth/oauth_routes.py
+++ b/nextcloud_mcp_server/auth/oauth_routes.py
@@ -252,6 +252,7 @@ async def oauth_authorize(request: Request) -> RedirectResponse | JSONResponse:
"scope": scopes,
"state": idp_state,
"prompt": "consent", # Ensure refresh token
+ "resource": f"{oauth_config['mcp_server_url']}/mcp", # MCP server audience
}
auth_url = f"{authorization_endpoint}?{urlencode(idp_params)}"
@@ -359,6 +360,7 @@ async def oauth_authorize_nextcloud(
"state": state,
"prompt": "consent", # Force consent to show resource access
"access_type": "offline", # Request refresh token
+ "resource": oauth_config["nextcloud_resource_uri"], # Nextcloud audience
}
auth_url = f"{authorization_endpoint}?{urlencode(idp_params)}"
diff --git a/nextcloud_mcp_server/auth/progressive_token_verifier.py b/nextcloud_mcp_server/auth/progressive_token_verifier.py
deleted file mode 100644
index a7af61ba..00000000
--- a/nextcloud_mcp_server/auth/progressive_token_verifier.py
+++ /dev/null
@@ -1,366 +0,0 @@
-"""
-Token Verifier for ADR-004 Progressive Consent Architecture.
-
-This module implements token verification with strict audience separation:
-- Flow 1 tokens have aud: for MCP authentication
-- Flow 2 tokens have aud: "nextcloud" for resource access
-- Token Broker manages the exchange between audiences
-"""
-
-import logging
-import os
-from datetime import datetime, timezone
-from typing import Optional
-
-import httpx
-import jwt
-from mcp.server.auth.provider import AccessToken
-
-from nextcloud_mcp_server.auth.refresh_token_storage import RefreshTokenStorage
-from nextcloud_mcp_server.auth.token_broker import TokenBrokerService
-
-logger = logging.getLogger(__name__)
-
-
-class ProgressiveConsentTokenVerifier:
- """
- Token verifier for Progressive Consent dual OAuth flows.
-
- This verifier:
- 1. Validates Flow 1 tokens (aud: ) for MCP authentication
- 2. Checks if user has provisioned Nextcloud access (Flow 2)
- 3. Uses Token Broker to obtain aud: "nextcloud" tokens when needed
- """
-
- def __init__(
- self,
- token_storage: RefreshTokenStorage | None,
- token_broker: Optional[TokenBrokerService] = None,
- oidc_discovery_url: Optional[str] = None,
- nextcloud_host: Optional[str] = None,
- encryption_key: Optional[str] = None,
- mcp_client_id: Optional[str] = None,
- introspection_uri: Optional[str] = None,
- client_secret: Optional[str] = None,
- ):
- """
- Initialize the Progressive Consent token verifier.
-
- Args:
- token_storage: Storage for refresh tokens
- token_broker: Token broker service (created if not provided)
- oidc_discovery_url: OIDC provider discovery URL
- nextcloud_host: Nextcloud server URL
- encryption_key: Fernet key for token encryption
- mcp_client_id: MCP server OAuth client ID for audience validation
- introspection_uri: OAuth introspection endpoint URL (for opaque tokens)
- client_secret: OAuth client secret (required for introspection)
- """
- self.storage = token_storage
- self.oidc_discovery_url = oidc_discovery_url or os.getenv(
- "OIDC_DISCOVERY_URL",
- f"{os.getenv('NEXTCLOUD_HOST')}/.well-known/openid-configuration",
- )
- self.nextcloud_host = nextcloud_host or os.getenv("NEXTCLOUD_HOST")
- self.encryption_key = encryption_key or os.getenv("TOKEN_ENCRYPTION_KEY")
- self.mcp_client_id = mcp_client_id or os.getenv("OIDC_CLIENT_ID")
- self.introspection_uri = introspection_uri
- self.client_secret = client_secret or os.getenv("OIDC_CLIENT_SECRET")
-
- # HTTP client for introspection requests
- self._http_client: Optional[httpx.AsyncClient] = None
- if self.introspection_uri and self.mcp_client_id and self.client_secret:
- self._http_client = httpx.AsyncClient(timeout=10.0)
- logger.info(f"Introspection support enabled: {introspection_uri}")
- elif self.introspection_uri:
- logger.warning(
- "Introspection URI provided but missing client credentials - introspection disabled"
- )
-
- # Create token broker if not provided
- if token_broker:
- self.token_broker = token_broker
- elif self.encryption_key and token_storage and self.nextcloud_host:
- self.token_broker = TokenBrokerService(
- storage=token_storage,
- oidc_discovery_url=self.oidc_discovery_url,
- nextcloud_host=self.nextcloud_host,
- encryption_key=self.encryption_key,
- )
- else:
- self.token_broker = None
- if not self.encryption_key:
- logger.warning("Token broker not available - encryption key missing")
- elif not token_storage:
- logger.warning("Token broker not available - token storage missing")
- elif not self.nextcloud_host:
- logger.warning("Token broker not available - nextcloud host missing")
-
- async def verify_token(self, token: str) -> Optional[AccessToken]:
- """
- Verify a Flow 1 token (aud: ).
-
- This validates that:
- 1. Token has correct audience for MCP server (matches client ID)
- 2. Token is not expired
- 3. Token has valid signature (if verification enabled)
-
- Supports both JWT and opaque tokens:
- - JWT tokens: Decoded directly from payload
- - Opaque tokens: Validated via introspection endpoint (RFC 7662)
-
- Args:
- token: Access token from Flow 1 (JWT or opaque)
-
- Returns:
- AccessToken if valid, None otherwise
- """
- logger.info("🔐 verify_token called - attempting to validate token")
- logger.info(f"Token (first 50 chars): {token[:50]}...")
- logger.info(f"Expected MCP client ID: {self.mcp_client_id}")
-
- # Check if token is JWT format (has 3 parts separated by dots)
- is_jwt = "." in token and token.count(".") == 2
- logger.info(f"Token format: {'JWT' if is_jwt else 'opaque'}")
-
- if is_jwt:
- # Try JWT verification
- return await self._verify_jwt_token(token)
- else:
- # Fall back to introspection for opaque tokens
- return await self._verify_opaque_token(token)
-
- async def _verify_jwt_token(self, token: str) -> Optional[AccessToken]:
- """Verify JWT token by decoding payload."""
- try:
- # Decode without signature verification (IdP handles that)
- # In production, would verify signature with IdP public key
- payload = jwt.decode(token, options={"verify_signature": False})
- logger.info(f"Token payload decoded: {payload}")
-
- # CRITICAL: Verify audience is for MCP server (Flow 1)
- audiences = payload.get("aud", [])
- if isinstance(audiences, str):
- audiences = [audiences]
-
- # Audience validation:
- # - Accept tokens with no audience (will validate via introspection if needed)
- # - Accept tokens with MCP client ID in audience (Keycloak multi-audience)
- # - Accept tokens with resource URL in audience (Nextcloud JWT redirect URI)
- # - Reject tokens with "nextcloud" audience only (wrong flow)
- if audiences:
- # Check if MCP client ID is in the audience (Keycloak multi-audience)
- if self.mcp_client_id in audiences:
- logger.debug(
- f"Token has audience {audiences} - MCP client ID present"
- )
- # Check if this is a Nextcloud-only token (wrong flow)
- elif audiences == ["nextcloud"]:
- logger.warning(
- f"Token rejected: Nextcloud-only audience {audiences}"
- )
- logger.error(
- "Received Nextcloud token in MCP context - "
- "client may be using wrong token"
- )
- return None
- # Otherwise accept (likely resource URL audience from Nextcloud JWT)
- else:
- logger.info(
- f"Token has audience {audiences} (resource URL or non-standard) - accepting"
- )
- else:
- logger.info(
- "Token has no audience claim - accepting for MCP server validation"
- )
-
- # Check expiry
- exp = payload.get("exp", 0)
- if exp < datetime.now(timezone.utc).timestamp():
- logger.warning(
- f"❌ Token expired: exp={exp}, now={datetime.now(timezone.utc).timestamp()}"
- )
- return None
-
- # Extract user info
- user_id = payload.get("sub", "unknown")
- client_id = payload.get("client_id", "unknown")
- scopes = payload.get("scope", "").split()
- exp = payload.get("exp", None)
-
- logger.info(
- f"✅ Token validation successful! user={user_id}, scopes={scopes}"
- )
-
- # Create AccessToken for MCP framework
- return AccessToken(
- token=token,
- client_id=client_id,
- scopes=scopes,
- expires_at=exp,
- resource=user_id, # Store user_id in resource field (RFC 8707)
- )
-
- except jwt.InvalidTokenError as e:
- logger.warning(f"❌ Invalid token (JWT decode failed): {e}")
- return None
- except Exception as e:
- logger.error(f"❌ Token verification failed with exception: {e}")
- return None
-
- async def _verify_opaque_token(self, token: str) -> Optional[AccessToken]:
- """
- Verify opaque token via introspection endpoint (RFC 7662).
-
- Args:
- token: Opaque access token
-
- Returns:
- AccessToken if active and valid, None otherwise
- """
- if not self._http_client or not self.introspection_uri:
- logger.error(
- "❌ Cannot verify opaque token - introspection not configured. "
- "Set introspection_uri and client credentials."
- )
- return None
-
- try:
- logger.info(f"Introspecting token at {self.introspection_uri}")
-
- # Call introspection endpoint (requires client authentication)
- response = await self._http_client.post(
- self.introspection_uri,
- data={"token": token},
- auth=(self.mcp_client_id, self.client_secret),
- )
-
- if response.status_code != 200:
- logger.warning(
- f"❌ Introspection failed: HTTP {response.status_code} - {response.text[:200]}"
- )
- return None
-
- introspection_data = response.json()
- logger.info(f"Introspection response: {introspection_data}")
-
- # Check if token is active
- if not introspection_data.get("active", False):
- logger.warning("❌ Token introspection returned active=false")
- return None
-
- # Extract user info
- user_id = introspection_data.get("sub") or introspection_data.get(
- "username"
- )
- if not user_id:
- logger.error("❌ No username found in introspection response")
- return None
-
- # Extract scopes (space-separated string)
- scope_string = introspection_data.get("scope", "")
- scopes = scope_string.split() if scope_string else []
-
- # Extract client ID and expiration
- client_id = introspection_data.get("client_id", "unknown")
- exp = introspection_data.get("exp")
-
- logger.info(f"✅ Opaque token validated! user={user_id}, scopes={scopes}")
-
- return AccessToken(
- token=token,
- client_id=client_id,
- scopes=scopes,
- expires_at=int(exp) if exp else None,
- resource=user_id,
- )
-
- except httpx.TimeoutException:
- logger.error("❌ Timeout while introspecting token")
- return None
- except httpx.RequestError as e:
- logger.error(f"❌ Network error during introspection: {e}")
- return None
- except Exception as e:
- logger.error(f"❌ Introspection failed with exception: {e}")
- return None
-
- async def check_provisioning(self, user_id: str) -> bool:
- """
- Check if user has provisioned Nextcloud access (Flow 2).
-
- Args:
- user_id: User identifier from Flow 1 token
-
- Returns:
- True if user has completed Flow 2, False otherwise
- """
- if not self.storage:
- return False
-
- refresh_data = await self.storage.get_refresh_token(user_id)
- return refresh_data is not None
-
- async def get_nextcloud_token(self, user_id: str) -> Optional[str]:
- """
- Get a Nextcloud access token (aud: "nextcloud") for the user.
-
- This uses the Token Broker to:
- 1. Check for cached Nextcloud token
- 2. If expired, refresh using stored master refresh token
- 3. Return token with aud: "nextcloud" for API access
-
- Args:
- user_id: User identifier from Flow 1 token
-
- Returns:
- Nextcloud access token if provisioned, None otherwise
- """
- if not self.token_broker:
- logger.error("Token broker not available")
- return None
-
- # Check if user has provisioned access
- if not await self.check_provisioning(user_id):
- logger.info(f"User {user_id} has not provisioned Nextcloud access")
- return None
-
- # Get or refresh Nextcloud token
- try:
- nextcloud_token = await self.token_broker.get_nextcloud_token(user_id)
- if nextcloud_token:
- logger.debug(f"Obtained Nextcloud token for user {user_id}")
- return nextcloud_token
- except Exception as e:
- logger.error(f"Failed to get Nextcloud token: {e}")
- return None
-
- async def validate_scopes(
- self, token: AccessToken, required_scopes: list[str]
- ) -> bool:
- """
- Validate that token has required scopes.
-
- Args:
- token: The access token
- required_scopes: List of required scopes
-
- Returns:
- True if all required scopes present, False otherwise
- """
- token_scopes = set(token.scopes) if token.scopes else set()
- required = set(required_scopes)
-
- missing = required - token_scopes
- if missing:
- logger.debug(f"Token missing required scopes: {missing}")
- return False
-
- return True
-
- async def close(self):
- """Clean up resources."""
- if self.token_broker:
- await self.token_broker.close()
- if self._http_client:
- await self._http_client.aclose()
diff --git a/nextcloud_mcp_server/auth/token_verifier.py b/nextcloud_mcp_server/auth/token_verifier.py
deleted file mode 100644
index cc94eeba..00000000
--- a/nextcloud_mcp_server/auth/token_verifier.py
+++ /dev/null
@@ -1,492 +0,0 @@
-"""Token verification using Nextcloud OIDC userinfo endpoint."""
-
-import logging
-import time
-from typing import Any
-
-import httpx
-import jwt
-from jwt import PyJWKClient
-from mcp.server.auth.provider import AccessToken, TokenVerifier
-
-logger = logging.getLogger(__name__)
-
-
-class NextcloudTokenVerifier(TokenVerifier):
- """
- Validates access tokens using JWT verification with JWKS or userinfo endpoint fallback.
-
- This verifier supports both JWT and opaque tokens:
- 1. For JWT tokens: Verifies signature with JWKS and extracts scopes from payload
- 2. For opaque tokens: Falls back to userinfo endpoint validation
- 3. Caches successful responses to avoid repeated API calls/verifications
-
- JWT validation provides:
- - Faster validation (no HTTP call needed)
- - Direct scope extraction from token payload
- - Signature verification using JWKS
-
- Userinfo fallback provides:
- - Support for opaque tokens
- - Backward compatibility
- - Additional validation layer
- """
-
- def __init__(
- self,
- nextcloud_host: str,
- userinfo_uri: str,
- jwks_uri: str | None = None,
- issuer: str | None = None,
- introspection_uri: str | None = None,
- client_id: str | None = None,
- client_secret: str | None = None,
- cache_ttl: int = 3600,
- ):
- """
- Initialize the token verifier.
-
- Args:
- nextcloud_host: Base URL of the Nextcloud instance (e.g., https://cloud.example.com)
- userinfo_uri: Full URL to the userinfo endpoint
- jwks_uri: Full URL to the JWKS endpoint (for JWT verification)
- issuer: Expected issuer claim value (for JWT verification)
- introspection_uri: Full URL to the introspection endpoint (for opaque tokens)
- client_id: OAuth client ID (required for introspection)
- client_secret: OAuth client secret (required for introspection)
- cache_ttl: Time-to-live for cached tokens in seconds (default: 3600)
- """
- self.nextcloud_host = nextcloud_host.rstrip("/")
- self.userinfo_uri = userinfo_uri
- self.jwks_uri = jwks_uri
- self.issuer = issuer
- self.introspection_uri = introspection_uri
- self.client_id = client_id
- self.client_secret = client_secret
- self.cache_ttl = cache_ttl
-
- # Cache: token -> (userinfo, expiry_timestamp)
- self._token_cache: dict[str, tuple[dict[str, Any], float]] = {}
-
- # HTTP client for userinfo/introspection requests
- self._client = httpx.AsyncClient(timeout=10.0)
-
- # PyJWKClient for JWT verification (lazy initialization)
- self._jwks_client: PyJWKClient | None = None
- if jwks_uri:
- logger.info(f"JWT verification enabled with JWKS URI: {jwks_uri}")
- self._jwks_client = PyJWKClient(jwks_uri, cache_keys=True)
-
- # Introspection support
- if introspection_uri and client_id and client_secret:
- logger.info(f"Token introspection enabled: {introspection_uri}")
- elif introspection_uri:
- logger.warning(
- "Introspection URI provided but missing client credentials - introspection disabled"
- )
-
- async def verify_token(self, token: str) -> AccessToken | None:
- """
- Verify a bearer token using JWT verification, introspection, or userinfo endpoint.
-
- This method:
- 1. Checks the cache first for recent validations
- 2. Attempts JWT verification if JWKS is configured and token looks like JWT
- 3. Falls back to introspection for opaque tokens (if configured)
- 4. Falls back to userinfo endpoint as last resort
- 5. Returns AccessToken with username and scopes
-
- Args:
- token: The bearer token to verify
-
- Returns:
- AccessToken if valid, None if invalid or expired
- """
- # Check cache first
- cached = self._get_cached_token(token)
- if cached:
- logger.debug("Token found in cache")
- return cached
-
- # Try JWT verification first if enabled and token looks like JWT
- is_jwt_format = self._is_jwt_format(token)
- logger.debug(
- f"Token format check: is_jwt_format={is_jwt_format}, _jwks_client={self._jwks_client is not None}"
- )
- if self._jwks_client and is_jwt_format:
- logger.debug("Attempting JWT verification...")
- jwt_result = self._verify_jwt(token)
- if jwt_result:
- logger.info("Token validated via JWT verification")
- return jwt_result
- else:
- logger.warning("JWT verification failed, will try other methods")
-
- # For opaque tokens, try introspection if available
- if self.introspection_uri and self.client_id and self.client_secret:
- logger.debug("Attempting token introspection...")
- try:
- introspection_result = await self._verify_via_introspection(token)
- if introspection_result:
- logger.info("Token validated via introspection")
- return introspection_result
- except Exception as e:
- logger.warning(f"Introspection failed: {e}")
-
- # Fall back to userinfo endpoint validation (last resort)
- logger.debug("Attempting userinfo endpoint validation...")
- try:
- return await self._verify_via_userinfo(token)
- except Exception as e:
- logger.warning(f"Token verification failed: {e}")
- return None
-
- def _is_jwt_format(self, token: str) -> bool:
- """
- Check if token looks like a JWT (has 3 parts separated by dots).
-
- Args:
- token: The token to check
-
- Returns:
- True if token appears to be JWT format
- """
- return "." in token and token.count(".") == 2
-
- def _verify_jwt(self, token: str) -> AccessToken | None:
- """
- Verify JWT token with signature validation using JWKS.
-
- Args:
- token: The JWT token to verify
-
- Returns:
- AccessToken if valid, None if invalid
- """
- try:
- # Get signing key from JWKS
- assert self._jwks_client is not None # Caller should check before calling
- signing_key = self._jwks_client.get_signing_key_from_jwt(token)
-
- # Verify and decode JWT
- # Accept tokens with audience: "mcp-server" or ["mcp-server", "nextcloud"]
- # This allows:
- # 1. Tokens from MCP clients (aud: "mcp-server")
- # 2. Tokens for Nextcloud APIs (aud: "nextcloud")
- # 3. Tokens for both (aud: ["mcp-server", "nextcloud"])
- payload = jwt.decode(
- token,
- signing_key.key,
- algorithms=["RS256"],
- issuer=self.issuer,
- audience=["mcp-server", "nextcloud"], # Accept either audience
- options={
- "verify_signature": True,
- "verify_exp": True,
- "verify_iat": True,
- "verify_iss": True if self.issuer else False,
- "verify_aud": True, # Enable audience validation
- },
- )
-
- logger.debug(f"JWT verified successfully for user: {payload.get('sub')}")
- logger.debug(f"Full JWT payload: {payload}")
-
- # Extract username (sub claim, with fallback to preferred_username)
- # Some OIDC providers (like Keycloak) may not include sub in access tokens
- username = payload.get("sub") or payload.get("preferred_username")
- if not username:
- logger.error(
- "No 'sub' or 'preferred_username' claim found in JWT payload"
- )
- return None
-
- # Extract scopes from scope claim (space-separated string)
- scope_string = payload.get("scope", "")
- scopes = scope_string.split() if scope_string else []
- logger.debug(
- f"Extracted scopes from JWT - scope claim: '{scope_string}' -> scopes list: {scopes}"
- )
-
- # Extract expiration
- exp = payload.get("exp")
- if not exp:
- logger.warning("No 'exp' claim in JWT, using default TTL")
- exp = int(time.time() + self.cache_ttl)
-
- # Cache the result
- userinfo = {
- "sub": username,
- "scope": scope_string,
- **{k: v for k, v in payload.items() if k not in ["sub", "scope"]},
- }
- self._token_cache[token] = (userinfo, exp)
-
- return AccessToken(
- token=token,
- client_id=payload.get("client_id", ""),
- scopes=scopes,
- expires_at=exp,
- resource=username, # Store username in resource field (RFC 8707)
- )
-
- except jwt.ExpiredSignatureError:
- logger.info("JWT token has expired")
- return None
- except jwt.InvalidIssuerError as e:
- logger.warning(f"JWT issuer validation failed: {e}")
- return None
- except jwt.InvalidTokenError as e:
- logger.warning(f"JWT validation failed: {e}")
- return None
- except Exception as e:
- logger.error(f"Unexpected error during JWT verification: {e}")
- return None
-
- async def _verify_via_introspection(self, token: str) -> AccessToken | None:
- """
- Validate token by calling the introspection endpoint (RFC 7662).
-
- This method validates opaque tokens and retrieves their scopes.
-
- Args:
- token: The bearer token to introspect
-
- Returns:
- AccessToken if active, None if inactive or invalid
- """
- try:
- # Introspection requires client authentication
- response = await self._client.post(
- self.introspection_uri, # type: ignore
- data={"token": token},
- auth=(self.client_id, self.client_secret),
- )
-
- if response.status_code == 200:
- introspection_data = response.json()
-
- # Check if token is active
- if not introspection_data.get("active", False):
- logger.info("Token introspection returned inactive=false")
- return None
-
- logger.debug(
- f"Token introspected successfully for user: {introspection_data.get('sub')}"
- )
-
- # Extract username
- username = introspection_data.get("sub") or introspection_data.get(
- "username"
- )
- if not username:
- logger.error("No username found in introspection response")
- return None
-
- # Extract scopes (space-separated string)
- scope_string = introspection_data.get("scope", "")
- scopes = scope_string.split() if scope_string else []
- logger.debug(f"Extracted scopes from introspection: {scopes}")
-
- # Extract expiration
- exp = introspection_data.get("exp")
- if exp:
- expiry = float(exp)
- else:
- logger.warning(
- "No 'exp' in introspection response, using default TTL"
- )
- expiry = time.time() + self.cache_ttl
-
- # Cache the result
- cache_data = {
- "sub": username,
- "scope": scope_string,
- **{
- k: v
- for k, v in introspection_data.items()
- if k not in ["sub", "scope", "active"]
- },
- }
- self._token_cache[token] = (cache_data, expiry)
-
- return AccessToken(
- token=token,
- client_id=introspection_data.get("client_id", ""),
- scopes=scopes,
- expires_at=int(expiry),
- resource=username,
- )
-
- elif response.status_code in (400, 401, 403):
- logger.warning(
- f"Token introspection failed: HTTP {response.status_code}. "
- f"This may indicate: (1) Client credentials mismatch - trying to introspect "
- f"token issued to different OAuth client, (2) Expired client credentials, "
- f"(3) Invalid token. Will fall back to userinfo endpoint. "
- f"Response: {response.text[:200] if response.text else 'empty'}"
- )
- return None
- else:
- logger.warning(
- f"Unexpected response from introspection: {response.status_code}. "
- f"Response: {response.text[:200] if response.text else 'empty'}"
- )
- return None
-
- except httpx.TimeoutException:
- logger.error("Timeout while introspecting token")
- return None
- except httpx.RequestError as e:
- logger.error(f"Network error while introspecting token: {e}")
- return None
- except Exception as e:
- logger.error(f"Unexpected error during token introspection: {e}")
- return None
-
- async def _verify_via_userinfo(self, token: str) -> AccessToken | None:
- """
- Validate token by calling the userinfo endpoint.
-
- Args:
- token: The bearer token to verify
-
- Returns:
- AccessToken if valid, None otherwise
- """
- try:
- response = await self._client.get(
- self.userinfo_uri, headers={"Authorization": f"Bearer {token}"}
- )
-
- if response.status_code == 200:
- userinfo = response.json()
- logger.debug(
- f"Token validated successfully for user: {userinfo.get('sub')}"
- )
-
- # Cache the result
- expiry = time.time() + self.cache_ttl
- self._token_cache[token] = (userinfo, expiry)
-
- # Create AccessToken with username in resource field (workaround for MCP SDK)
- username = userinfo.get("sub") or userinfo.get("preferred_username")
- if not username:
- logger.error("No username found in userinfo response")
- return None
-
- return AccessToken(
- token=token,
- client_id="", # Not available from userinfo
- scopes=self._extract_scopes(userinfo),
- expires_at=int(expiry),
- resource=username, # Store username in resource field (RFC 8707)
- )
-
- elif response.status_code in (400, 401, 403):
- logger.info(f"Token validation failed: HTTP {response.status_code}")
- return None
- else:
- logger.warning(
- f"Unexpected response from userinfo: {response.status_code}"
- )
- return None
-
- except httpx.TimeoutException:
- logger.error("Timeout while validating token via userinfo endpoint")
- return None
- except httpx.RequestError as e:
- logger.error(f"Network error while validating token: {e}")
- return None
- except Exception as e:
- logger.error(f"Unexpected error during token validation: {e}")
- return None
-
- def _get_cached_token(self, token: str) -> AccessToken | None:
- """
- Retrieve a token from cache if not expired.
-
- Args:
- token: The bearer token to look up
-
- Returns:
- AccessToken if cached and valid, None otherwise
- """
- if token not in self._token_cache:
- return None
-
- userinfo, expiry = self._token_cache[token]
-
- # Check if expired
- if time.time() >= expiry:
- logger.debug("Cached token expired, removing from cache")
- del self._token_cache[token]
- return None
-
- # Return cached AccessToken
- username = userinfo.get("sub") or userinfo.get("preferred_username")
- return AccessToken(
- token=token,
- client_id="",
- scopes=self._extract_scopes(userinfo),
- expires_at=int(expiry),
- resource=username,
- )
-
- def _extract_scopes(self, userinfo: dict[str, Any]) -> list[str]:
- """
- Extract scopes from userinfo response.
-
- First attempts to read actual scopes from the 'scope' field (RFC 8693).
- If not present, infers scopes from the claims present in the response.
-
- Args:
- userinfo: The userinfo response dictionary
-
- Returns:
- List of scopes (actual or inferred)
- """
- # Try to get actual scopes from userinfo response (if OIDC provider includes it)
- scope_string = userinfo.get("scope")
- if scope_string:
- scopes = scope_string.split() if isinstance(scope_string, str) else []
- if scopes:
- logger.debug(
- f"Using actual scopes from userinfo: {scopes} (scope field present)"
- )
- return scopes
-
- # Fallback: Infer scopes from claims present in response
- # This maintains backward compatibility with OIDC providers that don't
- # include the scope field in userinfo responses
- logger.debug(
- "No scope field in userinfo response, inferring scopes from claims"
- )
- scopes = ["openid"] # Always present
-
- if "email" in userinfo:
- scopes.append("email")
-
- if any(
- key in userinfo for key in ["name", "given_name", "family_name", "picture"]
- ):
- scopes.append("profile")
-
- if "roles" in userinfo:
- scopes.append("roles")
-
- if "groups" in userinfo:
- scopes.append("groups")
-
- logger.debug(f"Inferred scopes from userinfo claims: {scopes}")
- return scopes
-
- def clear_cache(self):
- """Clear the token cache."""
- self._token_cache.clear()
- logger.debug("Token cache cleared")
-
- async def close(self):
- """Cleanup resources."""
- await self._client.aclose()
- logger.debug("Token verifier closed")
diff --git a/nextcloud_mcp_server/auth/unified_verifier.py b/nextcloud_mcp_server/auth/unified_verifier.py
new file mode 100644
index 00000000..bd7e99b5
--- /dev/null
+++ b/nextcloud_mcp_server/auth/unified_verifier.py
@@ -0,0 +1,417 @@
+"""
+Unified Token Verifier for ADR-005 Token Audience Validation.
+
+This module replaces both NextcloudTokenVerifier and ProgressiveConsentTokenVerifier
+with a single implementation that supports two compliant OAuth modes:
+
+1. Multi-audience mode (default): Validates MCP audience per RFC 7519 (resource servers
+ validate only their own audience). Nextcloud independently validates its own audience.
+2. Token exchange mode (opt-in): Tokens have MCP audience only, exchanged for Nextcloud tokens
+
+Key Design Principles:
+- Token verification happens HERE (validates MCP audience per OAuth spec)
+- Token exchange happens in context_helper.py (when creating NextcloudClient)
+- No token passthrough allowed (complies with MCP Security Specification)
+- Token reuse IS allowed for multi-audience tokens (RFC 8707)
+"""
+
+import hashlib
+import logging
+import time
+from typing import Any
+
+import httpx
+import jwt
+from jwt import PyJWKClient
+from mcp.server.auth.provider import AccessToken, TokenVerifier
+
+from nextcloud_mcp_server.config import Settings
+
+logger = logging.getLogger(__name__)
+
+
+class UnifiedTokenVerifier(TokenVerifier):
+ """
+ Unified token verifier supporting both multi-audience and token exchange modes.
+ Compliant with MCP security specification - no token pass-through.
+
+ This verifier:
+ 1. Validates tokens using JWT verification with JWKS or introspection fallback
+ 2. Enforces proper audience validation based on configured mode
+ 3. Caches successful validations to avoid repeated API calls
+
+ Mode Selection (via ENABLE_TOKEN_EXCHANGE setting):
+ - False/omit (default): Multi-audience mode - validates MCP audience only (per RFC 7519).
+ Nextcloud independently validates its own audience when receiving API calls.
+ - True: Exchange mode - requires MCP audience only, then exchanges for Nextcloud token
+ """
+
+ def __init__(self, settings: Settings):
+ """
+ Initialize the unified token verifier.
+
+ Args:
+ settings: Application settings containing OAuth configuration
+ """
+ self.settings = settings
+ self.mode = "exchange" if settings.enable_token_exchange else "multi-audience"
+
+ # Common components for all modes
+ self.http_client = httpx.AsyncClient(timeout=10.0)
+
+ # JWT verification support
+ self.jwks_client: PyJWKClient | None = None
+ if hasattr(settings, "jwks_uri") and settings.jwks_uri:
+ logger.info(f"JWT verification enabled with JWKS URI: {settings.jwks_uri}")
+ self.jwks_client = PyJWKClient(settings.jwks_uri, cache_keys=True)
+
+ # Introspection support (for opaque tokens)
+ self.introspection_uri: str | None = None
+ if (
+ hasattr(settings, "introspection_uri")
+ and settings.introspection_uri
+ and settings.oidc_client_id
+ and settings.oidc_client_secret
+ ):
+ self.introspection_uri = settings.introspection_uri
+ logger.info(f"Token introspection enabled: {self.introspection_uri}")
+
+ # Token cache: token_hash -> (userinfo, expiry_timestamp)
+ self._token_cache: dict[str, tuple[dict[str, Any], float]] = {}
+ self.cache_ttl = 3600 # 1 hour default
+
+ logger.info(
+ f"UnifiedTokenVerifier initialized in {self.mode} mode. "
+ f"MCP audience: {settings.oidc_client_id} or {settings.nextcloud_mcp_server_url}, "
+ f"Nextcloud resource URI: {settings.nextcloud_resource_uri}"
+ )
+
+ async def verify_token(self, token: str) -> AccessToken | None:
+ """
+ Verify token according to MCP TokenVerifier protocol.
+
+ Per RFC 7519, we validate only MCP audience. The mode determines what
+ happens AFTER verification in context_helper.py:
+ - Multi-audience mode: Use token directly (Nextcloud validates its own audience)
+ - Exchange mode: Exchange for Nextcloud-audience token via RFC 8693
+
+ Args:
+ token: Bearer token to verify
+
+ Returns:
+ AccessToken if valid with MCP audience, None otherwise
+ """
+ # Check cache first
+ cached = self._get_cached_token(token)
+ if cached:
+ logger.debug("Token found in cache")
+ return cached
+
+ # Both modes do the same validation (MCP audience only)
+ return await self._verify_mcp_audience(token)
+
+ async def _verify_mcp_audience(self, token: str) -> AccessToken | None:
+ """
+ Validate token has MCP audience.
+
+ Per RFC 7519 Section 4.1.3, resource servers validate only their own
+ presence in the audience claim. We don't validate Nextcloud's audience -
+ that's Nextcloud's responsibility when it receives the token.
+
+ Args:
+ token: Bearer token to verify
+
+ Returns:
+ AccessToken if valid with MCP audience, None otherwise
+ """
+ try:
+ # Attempt JWT verification first
+ if self._is_jwt_format(token) and self.jwks_client:
+ payload = await self._verify_jwt_signature(token)
+ else:
+ # Fall back to introspection for opaque tokens
+ payload = await self._introspect_token(token)
+ if not payload:
+ return None
+
+ # Check payload is valid
+ if not payload:
+ return None
+
+ # Validate MCP audience is present
+ if not self._has_mcp_audience(payload):
+ audiences = payload.get("aud", [])
+ logger.error(
+ f"Token rejected: Missing MCP audience. "
+ f"Got {audiences}, need MCP ({self.settings.oidc_client_id} or "
+ f"{self.settings.nextcloud_mcp_server_url})"
+ )
+ return None
+
+ # Log based on mode for clarity
+ if self.mode == "multi-audience":
+ logger.info(
+ "MCP audience validated - token can be used directly "
+ "(Nextcloud will validate its own audience)"
+ )
+ else:
+ logger.info(
+ "MCP audience validated - token will be exchanged for Nextcloud access"
+ )
+
+ return self._create_access_token(token, payload)
+
+ except Exception as e:
+ logger.error(f"Token verification failed: {e}")
+ return None
+
+ def _has_mcp_audience(self, payload: dict[str, Any]) -> bool:
+ """
+ Check if token has MCP audience.
+
+ Per RFC 7519 Section 4.1.3, resource servers should only validate their own
+ presence in the audience claim. We don't validate Nextcloud's audience - that's
+ Nextcloud's responsibility when it receives the token.
+
+ Args:
+ payload: Decoded token payload
+
+ Returns:
+ True if MCP audience present, False otherwise
+ """
+ audiences = payload.get("aud", [])
+ if isinstance(audiences, str):
+ audiences = [audiences]
+
+ audiences_set = set(audiences)
+
+ # MCP must have at least one: client_id OR server_url OR server_url/mcp
+ return bool(
+ self.settings.oidc_client_id in audiences_set
+ or (
+ self.settings.nextcloud_mcp_server_url
+ and (
+ self.settings.nextcloud_mcp_server_url in audiences_set
+ or f"{self.settings.nextcloud_mcp_server_url}/mcp" in audiences_set
+ )
+ )
+ )
+
+ def _is_jwt_format(self, token: str) -> bool:
+ """
+ Check if token looks like a JWT (has 3 parts separated by dots).
+
+ Args:
+ token: The token to check
+
+ Returns:
+ True if token appears to be JWT format
+ """
+ return "." in token and token.count(".") == 2
+
+ async def _verify_jwt_signature(self, token: str) -> dict[str, Any] | None:
+ """
+ Verify JWT token with signature validation using JWKS.
+
+ Args:
+ token: JWT token to verify
+
+ Returns:
+ Decoded payload if valid, None if invalid
+ """
+ try:
+ assert self.jwks_client is not None # Caller should check before calling
+
+ # Get signing key from JWKS
+ signing_key = self.jwks_client.get_signing_key_from_jwt(token)
+
+ # Verify and decode JWT
+ # Note: We don't validate audience here - that's done separately based on mode
+ payload = jwt.decode(
+ token,
+ signing_key.key,
+ algorithms=["RS256"],
+ issuer=self.settings.oidc_issuer
+ if hasattr(self.settings, "oidc_issuer")
+ else None,
+ options={
+ "verify_signature": True,
+ "verify_exp": True,
+ "verify_iat": True,
+ "verify_iss": True
+ if hasattr(self.settings, "oidc_issuer")
+ and self.settings.oidc_issuer
+ else False,
+ "verify_aud": False, # We handle audience validation separately
+ },
+ )
+
+ logger.debug(f"JWT signature verified for user: {payload.get('sub')}")
+ return payload
+
+ except jwt.ExpiredSignatureError:
+ logger.info("JWT token has expired")
+ return None
+ except jwt.InvalidIssuerError as e:
+ logger.warning(f"JWT issuer validation failed: {e}")
+ return None
+ except jwt.InvalidTokenError as e:
+ logger.warning(f"JWT validation failed: {e}")
+ return None
+ except Exception as e:
+ logger.error(f"Unexpected error during JWT verification: {e}")
+ return None
+
+ async def _introspect_token(self, token: str) -> dict[str, Any] | None:
+ """
+ Validate token by calling the introspection endpoint (RFC 7662).
+
+ Args:
+ token: Bearer token to introspect
+
+ Returns:
+ Token payload if active, None if inactive or invalid
+ """
+ if not self.introspection_uri:
+ logger.debug("No introspection endpoint configured")
+ return None
+
+ try:
+ # Introspection requires client authentication
+ response = await self.http_client.post(
+ self.introspection_uri,
+ data={"token": token},
+ auth=(self.settings.oidc_client_id, self.settings.oidc_client_secret),
+ )
+
+ if response.status_code == 200:
+ introspection_data = response.json()
+
+ # Check if token is active
+ if not introspection_data.get("active", False):
+ logger.info("Token introspection returned inactive=false")
+ return None
+
+ logger.debug(
+ f"Token introspected successfully for user: {introspection_data.get('sub')}"
+ )
+ return introspection_data
+
+ elif response.status_code in (400, 401, 403):
+ logger.warning(
+ f"Token introspection failed: HTTP {response.status_code}. "
+ f"Response: {response.text[:200] if response.text else 'empty'}"
+ )
+ return None
+ else:
+ logger.warning(
+ f"Unexpected response from introspection: {response.status_code}. "
+ f"Response: {response.text[:200] if response.text else 'empty'}"
+ )
+ return None
+
+ except httpx.TimeoutException:
+ logger.error("Timeout while introspecting token")
+ return None
+ except httpx.RequestError as e:
+ logger.error(f"Network error while introspecting token: {e}")
+ return None
+ except Exception as e:
+ logger.error(f"Unexpected error during token introspection: {e}")
+ return None
+
+ def _create_access_token(
+ self, token: str, payload: dict[str, Any]
+ ) -> AccessToken | None:
+ """
+ Create AccessToken object from validated token payload.
+
+ Args:
+ token: The bearer token
+ payload: Validated token payload
+
+ Returns:
+ AccessToken object or None if required fields missing
+ """
+ # Extract username (sub claim, with fallback to preferred_username)
+ username = payload.get("sub") or payload.get("preferred_username")
+ if not username:
+ logger.error(
+ "No 'sub' or 'preferred_username' claim found in token payload"
+ )
+ return None
+
+ # Extract scopes from scope claim (space-separated string)
+ scope_string = payload.get("scope", "")
+ scopes = scope_string.split() if scope_string else []
+ logger.debug(
+ f"Extracted scopes from token - scope claim: '{scope_string}' -> scopes list: {scopes}"
+ )
+
+ # Extract expiration
+ exp = payload.get("exp")
+ if not exp:
+ logger.warning("No 'exp' claim in token, using default TTL")
+ exp = int(time.time() + self.cache_ttl)
+
+ # Cache the result
+ token_hash = hashlib.sha256(token.encode()).hexdigest()
+ userinfo = {
+ "sub": username,
+ "scope": scope_string,
+ **{k: v for k, v in payload.items() if k not in ["sub", "scope"]},
+ }
+ self._token_cache[token_hash] = (userinfo, exp)
+
+ return AccessToken(
+ token=token,
+ client_id=payload.get("client_id", ""),
+ scopes=scopes,
+ expires_at=exp,
+ resource=username, # Store username in resource field (RFC 8707)
+ )
+
+ def _get_cached_token(self, token: str) -> AccessToken | None:
+ """
+ Retrieve a token from cache if not expired.
+
+ Args:
+ token: The bearer token to look up
+
+ Returns:
+ AccessToken if cached and valid, None otherwise
+ """
+ token_hash = hashlib.sha256(token.encode()).hexdigest()
+ if token_hash not in self._token_cache:
+ return None
+
+ userinfo, expiry = self._token_cache[token_hash]
+
+ # Check if expired
+ if time.time() >= expiry:
+ logger.debug("Cached token expired, removing from cache")
+ del self._token_cache[token_hash]
+ return None
+
+ # Return cached AccessToken
+ username = userinfo.get("sub") or userinfo.get("preferred_username")
+ scope_string = userinfo.get("scope", "")
+ scopes = scope_string.split() if scope_string else []
+
+ return AccessToken(
+ token=token,
+ client_id=userinfo.get("client_id", ""),
+ scopes=scopes,
+ expires_at=int(expiry),
+ resource=username,
+ )
+
+ def clear_cache(self):
+ """Clear the token cache."""
+ self._token_cache.clear()
+ logger.debug("Token cache cleared")
+
+ async def close(self):
+ """Cleanup resources."""
+ await self.http_client.aclose()
+ logger.debug("Unified token verifier closed")
diff --git a/nextcloud_mcp_server/config.py b/nextcloud_mcp_server/config.py
index e7a36ffc..73d86e4a 100644
--- a/nextcloud_mcp_server/config.py
+++ b/nextcloud_mcp_server/config.py
@@ -129,16 +129,29 @@ class Settings:
oidc_discovery_url: Optional[str] = None
oidc_client_id: Optional[str] = None
oidc_client_secret: Optional[str] = None
+ oidc_issuer: Optional[str] = None
# Nextcloud settings
nextcloud_host: Optional[str] = None
nextcloud_username: Optional[str] = None
nextcloud_password: Optional[str] = None
+ # ADR-005: Token Audience Validation (required for OAuth mode)
+ nextcloud_mcp_server_url: Optional[str] = None # MCP server URL (used as audience)
+ nextcloud_resource_uri: Optional[str] = None # Nextcloud resource identifier
+
+ # Token verification endpoints
+ jwks_uri: Optional[str] = None
+ introspection_uri: Optional[str] = None
+ userinfo_uri: Optional[str] = None
+
# Progressive Consent settings (always enabled - no flag needed)
enable_token_exchange: bool = False
enable_offline_access: bool = False
+ # Token exchange cache settings
+ token_exchange_cache_ttl: int = 300 # seconds (5 minutes default)
+
# Token settings
token_encryption_key: Optional[str] = None
token_storage_db: Optional[str] = None
@@ -155,10 +168,18 @@ def get_settings() -> Settings:
oidc_discovery_url=os.getenv("OIDC_DISCOVERY_URL"),
oidc_client_id=os.getenv("OIDC_CLIENT_ID"),
oidc_client_secret=os.getenv("OIDC_CLIENT_SECRET"),
+ oidc_issuer=os.getenv("OIDC_ISSUER"),
# Nextcloud settings
nextcloud_host=os.getenv("NEXTCLOUD_HOST"),
nextcloud_username=os.getenv("NEXTCLOUD_USERNAME"),
nextcloud_password=os.getenv("NEXTCLOUD_PASSWORD"),
+ # ADR-005: Token Audience Validation
+ nextcloud_mcp_server_url=os.getenv("NEXTCLOUD_MCP_SERVER_URL"),
+ nextcloud_resource_uri=os.getenv("NEXTCLOUD_RESOURCE_URI"),
+ # Token verification endpoints
+ jwks_uri=os.getenv("JWKS_URI"),
+ 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"
@@ -166,6 +187,8 @@ def get_settings() -> Settings:
enable_offline_access=(
os.getenv("ENABLE_OFFLINE_ACCESS", "false").lower() == "true"
),
+ # Token exchange cache settings
+ token_exchange_cache_ttl=int(os.getenv("TOKEN_EXCHANGE_CACHE_TTL", "300")),
# Token settings
token_encryption_key=os.getenv("TOKEN_ENCRYPTION_KEY"),
token_storage_db=os.getenv("TOKEN_STORAGE_DB", "/tmp/tokens.db"),
diff --git a/nextcloud_mcp_server/context.py b/nextcloud_mcp_server/context.py
index c568e29b..505274b1 100644
--- a/nextcloud_mcp_server/context.py
+++ b/nextcloud_mcp_server/context.py
@@ -10,12 +10,15 @@ async def get_client(ctx: Context) -> NextcloudClient:
"""
Get the appropriate Nextcloud client based on authentication mode.
- This function handles three modes:
+ ADR-005 compliant implementation supporting two modes:
1. BasicAuth mode: Returns shared client from lifespan context
- 2. OAuth pass-through mode (ENABLE_TOKEN_EXCHANGE=false, default):
- Verifies Flow 1 token and passes it to Nextcloud
- 3. OAuth token exchange mode (ENABLE_TOKEN_EXCHANGE=true):
- Exchanges Flow 1 token for ephemeral Nextcloud token via RFC 8693
+ 2. Multi-audience mode (ENABLE_TOKEN_EXCHANGE=false, default):
+ Token already contains both MCP and Nextcloud audiences - use directly
+ 3. 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.
@@ -49,20 +52,22 @@ async def get_client(ctx: Context) -> NextcloudClient:
# OAuth mode (has 'nextcloud_host' attribute)
if hasattr(lifespan_ctx, "nextcloud_host"):
- # Check if token exchange is enabled
- if settings.enable_token_exchange:
- from nextcloud_mcp_server.auth.context_helper import (
- get_session_client_from_context,
- )
+ from nextcloud_mcp_server.auth.context_helper import (
+ get_client_from_context,
+ get_session_client_from_context,
+ )
- # Token exchange mode: Exchange Flow 1 token for ephemeral Nextcloud token
+ 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:
- # Pass-through mode (default): Verify and pass Flow 1 token to Nextcloud
- from nextcloud_mcp_server.auth import get_client_from_context
-
+ # 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)
# Unknown context type
diff --git a/tests/conftest.py b/tests/conftest.py
index 47f0f21f..b98e5bba 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -2193,17 +2193,35 @@ async def _get_oauth_token_for_user(
logger.info(f"Getting OAuth token for user: {username}...")
logger.info(f"Using shared OAuth client: {client_id[:16]}...")
+ # Fetch resource identifier from PRM endpoint (RFC 9728)
+ mcp_server_url = os.getenv("NEXTCLOUD_MCP_SERVER_URL", "http://localhost:8001")
+ prm_url = f"{mcp_server_url}/.well-known/oauth-protected-resource"
+
+ logger.debug(f"Fetching PRM metadata from: {prm_url}")
+ async with httpx.AsyncClient() as client:
+ prm_response = await client.get(prm_url, timeout=10)
+ if prm_response.status_code != 200:
+ logger.warning(f"Failed to fetch PRM metadata: {prm_response.status_code}")
+ # Fallback to default if PRM fetch fails
+ mcp_server_resource = f"{mcp_server_url}/mcp"
+ else:
+ prm_data = prm_response.json()
+ mcp_server_resource = prm_data.get("resource", f"{mcp_server_url}/mcp")
+ logger.info(f"Using resource from PRM: {mcp_server_resource}")
+
# Generate unique state parameter for this OAuth flow
state = secrets.token_urlsafe(32)
logger.debug(f"Generated state for {username}: {state[:16]}...")
# Construct authorization URL with state parameter
+ # Include resource parameter discovered from PRM endpoint
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"resource={quote(mcp_server_resource, safe='')}&" # Resource URI from PRM
f"scope=openid%20profile%20email%20notes:read%20notes:write%20calendar:read%20calendar:write%20contacts:read%20contacts:write%20cookbook:read%20cookbook:write%20deck:read%20deck:write%20tables:read%20tables:write%20files:read%20files:write%20sharing:read%20sharing:write"
)
diff --git a/tests/unit/test_unified_verifier.py b/tests/unit/test_unified_verifier.py
new file mode 100644
index 00000000..491ee0f2
--- /dev/null
+++ b/tests/unit/test_unified_verifier.py
@@ -0,0 +1,523 @@
+"""
+Unit tests for UnifiedTokenVerifier (ADR-005).
+
+Tests token audience validation for both multi-audience and token exchange modes
+without requiring real network calls or IdP connections.
+"""
+
+import time
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import jwt
+import pytest
+
+from nextcloud_mcp_server.auth.unified_verifier import UnifiedTokenVerifier
+from nextcloud_mcp_server.config import Settings
+
+pytestmark = pytest.mark.unit
+
+
+@pytest.fixture
+def base_settings():
+ """Create base settings for testing."""
+ return Settings(
+ oidc_client_id="test-client-id",
+ oidc_client_secret="test-client-secret",
+ oidc_issuer="https://idp.example.com",
+ nextcloud_host="https://nextcloud.example.com",
+ nextcloud_mcp_server_url="http://localhost:8000",
+ 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."""
+
+ def test_init_multi_audience_mode(self, base_settings):
+ """Test verifier initialization in multi-audience mode."""
+ verifier = UnifiedTokenVerifier(base_settings)
+ assert verifier.mode == "multi-audience"
+ assert verifier.settings == base_settings
+
+ def test_init_exchange_mode(self, exchange_settings):
+ """Test verifier initialization in token exchange mode."""
+ verifier = UnifiedTokenVerifier(exchange_settings)
+ assert verifier.mode == "exchange"
+ assert verifier.settings == exchange_settings
+
+
+class TestAudienceValidation:
+ """Test audience validation logic."""
+
+ def test_validate_multi_audience_both_present(self, base_settings):
+ """Test MCP audience validation with both audiences present."""
+ verifier = UnifiedTokenVerifier(base_settings)
+ payload = {
+ "aud": ["test-client-id", "http://localhost:8080"],
+ "sub": "testuser",
+ "exp": int(time.time() + 3600),
+ }
+
+ assert verifier._has_mcp_audience(payload) is True
+
+ def test_validate_multi_audience_server_url_and_resource(self, base_settings):
+ """Test MCP audience validation with server URL instead of client ID."""
+ verifier = UnifiedTokenVerifier(base_settings)
+ payload = {
+ "aud": ["http://localhost:8000", "http://localhost:8080"],
+ "sub": "testuser",
+ "exp": int(time.time() + 3600),
+ }
+
+ assert verifier._has_mcp_audience(payload) is True
+
+ def test_validate_multi_audience_missing_mcp(self, base_settings):
+ """Test MCP audience validation fails without MCP audience."""
+ verifier = UnifiedTokenVerifier(base_settings)
+ payload = {
+ "aud": ["http://localhost:8080"], # Only Nextcloud
+ "sub": "testuser",
+ "exp": int(time.time() + 3600),
+ }
+
+ assert verifier._has_mcp_audience(payload) is False
+
+ def test_validate_multi_audience_missing_nextcloud(self, base_settings):
+ """Test MCP audience validation succeeds with only MCP audience (RFC 7519 compliant)."""
+ verifier = UnifiedTokenVerifier(base_settings)
+ payload = {
+ "aud": ["test-client-id"], # Only MCP
+ "sub": "testuser",
+ "exp": int(time.time() + 3600),
+ }
+
+ # Per RFC 7519, we only validate MCP audience. Nextcloud validates its own.
+ assert verifier._has_mcp_audience(payload) is True
+
+ def test_validate_multi_audience_string_audience(self, base_settings):
+ """Test MCP audience validation with string audience works (RFC 7519 compliant)."""
+ verifier = UnifiedTokenVerifier(base_settings)
+ payload = {
+ "aud": "test-client-id", # Single audience as string
+ "sub": "testuser",
+ "exp": int(time.time() + 3600),
+ }
+
+ # 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):
+ """Test MCP audience validation with client ID."""
+ verifier = UnifiedTokenVerifier(exchange_settings)
+ payload = {
+ "aud": ["test-client-id"],
+ "sub": "testuser",
+ "exp": int(time.time() + 3600),
+ }
+
+ assert verifier._has_mcp_audience(payload) is True
+
+ def test_has_mcp_audience_with_server_url(self, exchange_settings):
+ """Test MCP audience validation with server URL."""
+ verifier = UnifiedTokenVerifier(exchange_settings)
+ payload = {
+ "aud": ["http://localhost:8000"],
+ "sub": "testuser",
+ "exp": int(time.time() + 3600),
+ }
+
+ assert verifier._has_mcp_audience(payload) is True
+
+ def test_has_mcp_audience_missing(self, exchange_settings):
+ """Test MCP audience validation fails without MCP audience."""
+ verifier = UnifiedTokenVerifier(exchange_settings)
+ payload = {
+ "aud": ["http://localhost:8080"], # Wrong audience
+ "sub": "testuser",
+ "exp": int(time.time() + 3600),
+ }
+
+ assert verifier._has_mcp_audience(payload) is False
+
+
+class TestTokenFormatDetection:
+ """Test JWT format detection."""
+
+ def test_is_jwt_format_valid(self, base_settings):
+ """Test JWT format detection with valid JWT."""
+ verifier = UnifiedTokenVerifier(base_settings)
+ jwt_token = "eyJhbGc.eyJzdWI.signature"
+ assert verifier._is_jwt_format(jwt_token) is True
+
+ def test_is_jwt_format_opaque(self, base_settings):
+ """Test JWT format detection with opaque token."""
+ verifier = UnifiedTokenVerifier(base_settings)
+ opaque_token = "opaque-token-12345"
+ assert verifier._is_jwt_format(opaque_token) is False
+
+
+class TestTokenCaching:
+ """Test token caching functionality."""
+
+ async def test_cache_stores_and_retrieves(self, base_settings):
+ """Test token caching stores and retrieves tokens."""
+ verifier = UnifiedTokenVerifier(base_settings)
+
+ # Create a valid access token
+ payload = {
+ "aud": ["test-client-id", "http://localhost:8080"],
+ "sub": "testuser",
+ "scope": "openid profile",
+ "exp": int(time.time() + 3600),
+ "client_id": "test-client-id",
+ }
+ test_token = jwt.encode(payload, "secret", algorithm="HS256")
+
+ # Create AccessToken and cache it
+ access_token = verifier._create_access_token(test_token, payload)
+ assert access_token is not None
+
+ # Should retrieve from cache
+ cached = verifier._get_cached_token(test_token)
+ assert cached is not None
+ assert cached.resource == "testuser"
+ assert cached.scopes == ["openid", "profile"]
+
+ async def test_cache_respects_expiry(self, base_settings):
+ """Test that expired tokens are not returned from cache."""
+ verifier = UnifiedTokenVerifier(base_settings)
+
+ # Create expired token payload
+ payload = {
+ "aud": ["test-client-id", "http://localhost:8080"],
+ "sub": "testuser",
+ "scope": "openid profile",
+ "exp": int(time.time() - 100), # Expired 100 seconds ago
+ "client_id": "test-client-id",
+ }
+ test_token = jwt.encode(payload, "secret", algorithm="HS256")
+
+ # Create and cache
+ access_token = verifier._create_access_token(test_token, payload)
+ assert access_token is not None
+
+ # Should not retrieve expired token
+ cached = verifier._get_cached_token(test_token)
+ assert cached is None
+
+ async def test_cache_clear(self, base_settings):
+ """Test cache clearing."""
+ verifier = UnifiedTokenVerifier(base_settings)
+
+ # Create and cache token
+ payload = {
+ "aud": ["test-client-id", "http://localhost:8080"],
+ "sub": "testuser",
+ "exp": int(time.time() + 3600),
+ }
+ test_token = jwt.encode(payload, "secret", algorithm="HS256")
+ verifier._create_access_token(test_token, payload)
+
+ # Clear cache
+ verifier.clear_cache()
+
+ # Should not retrieve after clear
+ cached = verifier._get_cached_token(test_token)
+ assert cached is None
+
+
+class TestMultiAudienceVerification:
+ """Test multi-audience token verification."""
+
+ async def test_verify_multi_audience_with_introspection(self, base_settings):
+ """Test multi-audience verification using introspection."""
+ verifier = UnifiedTokenVerifier(base_settings)
+
+ # Mock introspection response
+ introspection_response = {
+ "active": True,
+ "sub": "testuser",
+ "aud": ["test-client-id", "http://localhost:8080"],
+ "scope": "openid profile",
+ "exp": int(time.time() + 3600),
+ "client_id": "test-client-id",
+ }
+
+ with patch.object(
+ verifier, "_introspect_token", return_value=introspection_response
+ ):
+ opaque_token = "opaque-token-12345"
+ result = await verifier._verify_mcp_audience(opaque_token)
+
+ assert result is not None
+ assert result.resource == "testuser"
+ assert result.scopes == ["openid", "profile"]
+
+ async def test_verify_multi_audience_fails_without_both_audiences(
+ self, base_settings
+ ):
+ """Test MCP audience verification succeeds with only MCP audience (RFC 7519 compliant)."""
+ verifier = UnifiedTokenVerifier(base_settings)
+
+ # Mock introspection response with only MCP audience
+ introspection_response = {
+ "active": True,
+ "sub": "testuser",
+ "aud": [
+ "test-client-id"
+ ], # Only MCP audience (Nextcloud validates its own)
+ "scope": "openid profile",
+ "exp": int(time.time() + 3600),
+ }
+
+ with patch.object(
+ verifier, "_introspect_token", return_value=introspection_response
+ ):
+ opaque_token = "opaque-token-12345"
+ result = await verifier._verify_mcp_audience(opaque_token)
+
+ # Should succeed with only MCP audience per RFC 7519
+ assert result is not None
+ assert result.resource == "testuser"
+
+
+class TestExchangeModeVerification:
+ """Test token exchange mode verification."""
+
+ async def test_verify_mcp_audience_only_success(self, exchange_settings):
+ """Test MCP-only audience verification succeeds with MCP audience."""
+ verifier = UnifiedTokenVerifier(exchange_settings)
+
+ # Mock introspection response with MCP audience only
+ introspection_response = {
+ "active": True,
+ "sub": "testuser",
+ "aud": ["test-client-id"],
+ "scope": "openid profile",
+ "exp": int(time.time() + 3600),
+ "client_id": "test-client-id",
+ }
+
+ with patch.object(
+ verifier, "_introspect_token", return_value=introspection_response
+ ):
+ opaque_token = "opaque-token-12345"
+ result = await verifier._verify_mcp_audience(opaque_token)
+
+ assert result is not None
+ assert result.resource == "testuser"
+
+ async def test_verify_mcp_audience_only_fails_without_mcp(self, exchange_settings):
+ """Test MCP audience verification fails without MCP audience."""
+ verifier = UnifiedTokenVerifier(exchange_settings)
+
+ # Mock introspection response without MCP audience
+ introspection_response = {
+ "active": True,
+ "sub": "testuser",
+ "aud": ["http://localhost:8080"], # Wrong audience
+ "scope": "openid profile",
+ "exp": int(time.time() + 3600),
+ }
+
+ with patch.object(
+ verifier, "_introspect_token", return_value=introspection_response
+ ):
+ opaque_token = "opaque-token-12345"
+ result = await verifier._verify_mcp_audience(opaque_token)
+
+ assert result is None
+
+
+class TestIntrospection:
+ """Test token introspection."""
+
+ async def test_introspect_active_token(self, base_settings):
+ """Test introspection of active token."""
+ verifier = UnifiedTokenVerifier(base_settings)
+
+ # Mock HTTP response
+ mock_response = MagicMock()
+ mock_response.status_code = 200
+ mock_response.json.return_value = {
+ "active": True,
+ "sub": "testuser",
+ "aud": ["test-client-id", "http://localhost:8080"],
+ "scope": "openid profile",
+ "exp": int(time.time() + 3600),
+ "client_id": "test-client-id",
+ }
+
+ verifier.http_client.post = AsyncMock(return_value=mock_response)
+
+ result = await verifier._introspect_token("test-token")
+ assert result is not None
+ assert result["active"] is True
+ assert result["sub"] == "testuser"
+
+ async def test_introspect_inactive_token(self, base_settings):
+ """Test introspection of inactive token."""
+ verifier = UnifiedTokenVerifier(base_settings)
+
+ # Mock HTTP response
+ mock_response = MagicMock()
+ mock_response.status_code = 200
+ mock_response.json.return_value = {"active": False}
+
+ verifier.http_client.post = AsyncMock(return_value=mock_response)
+
+ result = await verifier._introspect_token("test-token")
+ assert result is None
+
+ async def test_introspect_without_endpoint(self, base_settings):
+ """Test introspection when endpoint not configured."""
+ base_settings.introspection_uri = None
+ verifier = UnifiedTokenVerifier(base_settings)
+
+ result = await verifier._introspect_token("test-token")
+ assert result is None
+
+
+class TestAccessTokenCreation:
+ """Test AccessToken object creation."""
+
+ def test_create_access_token_success(self, base_settings):
+ """Test successful AccessToken creation."""
+ verifier = UnifiedTokenVerifier(base_settings)
+
+ payload = {
+ "sub": "testuser",
+ "scope": "openid profile email",
+ "exp": int(time.time() + 3600),
+ "client_id": "test-client-id",
+ }
+ token = "test-token-123"
+
+ result = verifier._create_access_token(token, payload)
+ assert result is not None
+ assert result.token == token
+ assert result.resource == "testuser"
+ assert result.scopes == ["openid", "profile", "email"]
+ assert result.client_id == "test-client-id"
+
+ def test_create_access_token_with_preferred_username(self, base_settings):
+ """Test AccessToken creation with preferred_username fallback."""
+ verifier = UnifiedTokenVerifier(base_settings)
+
+ payload = {
+ "preferred_username": "testuser", # No 'sub' claim
+ "scope": "openid profile",
+ "exp": int(time.time() + 3600),
+ }
+ token = "test-token-123"
+
+ result = verifier._create_access_token(token, payload)
+ assert result is not None
+ assert result.resource == "testuser"
+
+ def test_create_access_token_no_username(self, base_settings):
+ """Test AccessToken creation fails without username."""
+ verifier = UnifiedTokenVerifier(base_settings)
+
+ payload = {
+ # No sub or preferred_username
+ "scope": "openid profile",
+ "exp": int(time.time() + 3600),
+ }
+ token = "test-token-123"
+
+ result = verifier._create_access_token(token, payload)
+ assert result is None
+
+ def test_create_access_token_no_expiry(self, base_settings):
+ """Test AccessToken creation uses default TTL without expiry."""
+ verifier = UnifiedTokenVerifier(base_settings)
+
+ payload = {
+ "sub": "testuser",
+ "scope": "openid profile",
+ # No exp claim
+ }
+ token = "test-token-123"
+
+ result = verifier._create_access_token(token, payload)
+ assert result is not None
+ # Should have set a default expiry
+ assert result.expires_at > int(time.time())
+
+
+class TestVerifyTokenFlow:
+ """Test complete verify_token flow."""
+
+ async def test_verify_token_from_cache(self, base_settings):
+ """Test verify_token returns cached token."""
+ verifier = UnifiedTokenVerifier(base_settings)
+
+ payload = {
+ "aud": ["test-client-id", "http://localhost:8080"],
+ "sub": "testuser",
+ "scope": "openid profile",
+ "exp": int(time.time() + 3600),
+ }
+ token = jwt.encode(payload, "secret", algorithm="HS256")
+
+ # First call - should cache
+ result1 = verifier._create_access_token(token, payload)
+ assert result1 is not None
+
+ # Mock _verify_mcp_audience to ensure it's not called
+ with patch.object(verifier, "_verify_mcp_audience") as mock_verify:
+ result2 = await verifier.verify_token(token)
+ assert result2 is not None
+ assert result2.resource == "testuser"
+ # Should not call verification since it's cached
+ mock_verify.assert_not_called()
+
+ async def test_verify_token_multi_audience_mode(self, base_settings):
+ """Test verify_token in multi-audience mode."""
+ verifier = UnifiedTokenVerifier(base_settings)
+
+ introspection_response = {
+ "active": True,
+ "sub": "testuser",
+ "aud": ["test-client-id", "http://localhost:8080"],
+ "scope": "openid profile",
+ "exp": int(time.time() + 3600),
+ }
+
+ with patch.object(
+ verifier, "_introspect_token", return_value=introspection_response
+ ):
+ result = await verifier.verify_token("opaque-token")
+ assert result is not None
+ assert result.resource == "testuser"
+
+ async def test_verify_token_exchange_mode(self, exchange_settings):
+ """Test verify_token in exchange mode."""
+ verifier = UnifiedTokenVerifier(exchange_settings)
+
+ introspection_response = {
+ "active": True,
+ "sub": "testuser",
+ "aud": ["test-client-id"], # MCP audience only
+ "scope": "openid profile",
+ "exp": int(time.time() + 3600),
+ }
+
+ with patch.object(
+ verifier, "_introspect_token", return_value=introspection_response
+ ):
+ result = await verifier.verify_token("opaque-token")
+ assert result is not None
+ assert result.resource == "testuser"
diff --git a/third_party/oidc b/third_party/oidc
index 19cad6e5..e83dabba 160000
--- a/third_party/oidc
+++ b/third_party/oidc
@@ -1 +1 @@
-Subproject commit 19cad6e5b69127e62e4a7d03b139ada1dedc92a1
+Subproject commit e83dabbac17142f7ece6219e5b33ca145fc99ae6