admin/mcp-nextcloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: remove RFC 8693 token exchange and Keycloak OAuth implementation

Browse Source 
Nextcloud doesn't support OAuth bearer tokens without upstream patches,
making the RFC 8693 token exchange path untestable and dead code.

Removed:
- nextcloud_mcp_server/auth/token_exchange.py (597 lines)
- nextcloud_mcp_server/auth/keycloak_oauth.py (586 lines)
- OAUTH_TOKEN_EXCHANGE deployment mode from AuthMode enum
- get_session_client_from_context() from context_helper.py
- get_session_token() from token_broker.py
- enable_token_exchange / token_exchange_cache_ttl config fields
- oauth_token_exchange_total Prometheus metric
- Keycloak fixture block from tests/conftest.py (~408 lines)
- Token exchange unit tests from test_config_validators.py,
  test_unified_verifier.py, test_management_status_endpoint.py

Preserved:
- Multi-audience OAuth mode (OAUTH_SINGLE_AUDIENCE)
- Login Flow v2 provisioning with elicitation support
- Token broker background token management
- All existing test coverage for non-exchange paths

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2026-04-03 00:22:20 +02:00
co-authored by Claude Opus 4.6
parent c6316dbb91
commit 5730313574
17 changed files with 48 additions and 2124 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/conftest.py
+1 -427
View File
@@ -1,5 +1,4 @@
import base64
import hashlib
import json
import logging
import os
@@ -2868,433 +2867,8 @@ async def test_user_in_group(nc_client: NextcloudClient, test_user, test_group):
# ===========================================================================================
# Keycloak External IdP OAuth Fixtures
# ===========================================================================================
@pytest.fixture(scope="session")
async def keycloak_oauth_client_credentials(anyio_backend, oauth_callback_server):
"""
Fixture to obtain Keycloak OAuth client credentials for external IdP testing.
Uses pre-configured client from keycloak/realm-export.json (no DCR needed).
The client (nextcloud-mcp-server) is already configured with:
- serviceAccountsEnabled=true
- token.exchange.grant.enabled=true
- client.token.exchange.standard.enabled=true
Returns:
Tuple of (client_id, client_secret, callback_url, token_endpoint, authorization_endpoint)
"""
# Get Keycloak configuration from environment
keycloak_discovery_url = os.getenv(
"OIDC_DISCOVERY_URL",
"http://localhost:8888/realms/nextcloud-mcp/.well-known/openid-configuration",
)
client_id = os.getenv("OIDC_CLIENT_ID", "nextcloud-mcp-server")
client_secret = os.getenv("OIDC_CLIENT_SECRET", "mcp-secret-change-in-production")
if not all([keycloak_discovery_url, client_id, client_secret]):
pytest.skip(
"Keycloak OAuth requires OIDC_DISCOVERY_URL, OIDC_CLIENT_ID, OIDC_CLIENT_SECRET"
)
# Get callback URL from the real callback server
auth_states, callback_url = oauth_callback_server
logger.info("Setting up Keycloak external IdP OAuth client credentials...")
logger.info(f"Using Keycloak discovery URL: {keycloak_discovery_url}")
logger.info(f"Using static client credentials: {client_id}")
logger.info(f"Using real callback server at: {callback_url}")
async with httpx.AsyncClient(timeout=30.0) as http_client:
# OIDC Discovery
discovery_response = await http_client.get(keycloak_discovery_url)
discovery_response.raise_for_status()
oidc_config = discovery_response.json()
token_endpoint = oidc_config.get("token_endpoint")
authorization_endpoint = oidc_config.get("authorization_endpoint")
if not token_endpoint or not authorization_endpoint:
raise ValueError(
"Keycloak OIDC discovery missing required endpoints (token_endpoint or authorization_endpoint)"
)
logger.info(f"✓ Discovered token endpoint: {token_endpoint}")
logger.info(f"✓ Discovered authorization endpoint: {authorization_endpoint}")
yield (
client_id,
client_secret,
callback_url,
token_endpoint,
authorization_endpoint,
)
# No cleanup needed - client is pre-configured in realm export
async def _get_keycloak_oauth_token(
browser,
keycloak_oauth_client_credentials,
oauth_callback_server,
scopes: str,
username: str = "admin",
password: str = "admin",
) -> str:
"""
Helper function to obtain OAuth token from Keycloak using Playwright.
Args:
browser: Playwright browser instance
keycloak_oauth_client_credentials: Tuple of Keycloak OAuth client credentials
oauth_callback_server: OAuth callback server fixture
scopes: Space-separated list of scopes
username: Keycloak username (default: admin)
password: Keycloak password (default: admin)
Returns:
OAuth access token string from Keycloak
"""
# Get auth_states dict from callback server
auth_states, _ = oauth_callback_server
# Unpack Keycloak client credentials
client_id, client_secret, callback_url, token_endpoint, authorization_endpoint = (
keycloak_oauth_client_credentials
)
logger.info(f"Starting Playwright-based Keycloak OAuth flow with scopes: {scopes}")
logger.info(f"Using Keycloak client: {client_id}")
logger.info(f"Using real callback server at: {callback_url}")
logger.info(f"Authenticating as Keycloak user: {username}")
# Generate unique state parameter for this OAuth flow
state = secrets.token_urlsafe(32)
logger.debug(f"Generated state: {state[:16]}...")
# Generate PKCE parameters (required by Keycloak client configuration)
code_verifier = secrets.token_urlsafe(64) # 86 chars base64url
code_challenge = (
base64.urlsafe_b64encode(hashlib.sha256(code_verifier.encode()).digest())
.decode()
.rstrip("=")
)
logger.debug(f"Generated PKCE code_challenge: {code_challenge[:20]}...")
# URL-encode scopes
scopes_encoded = quote(scopes, safe="")
# Construct authorization URL with state, scopes, and PKCE parameters
auth_url = (
f"{authorization_endpoint}?"
f"response_type=code&"
f"client_id={client_id}&"
f"redirect_uri={quote(callback_url, safe='')}&"
f"state={state}&"
f"scope={scopes_encoded}&"
f"code_challenge={code_challenge}&"
f"code_challenge_method=S256"
)
logger.info(f"Authorization URL: {auth_url[:100]}...")
# Create browser context and page
context = await browser.new_context()
page = await context.new_page()
try:
# Navigate to Keycloak authorization endpoint
logger.info("Navigating to Keycloak authorization endpoint...")
await page.goto(auth_url, wait_until="networkidle", timeout=30000)
# Handle Keycloak login page
# Keycloak uses input#username and input#password (different from Nextcloud)
logger.info(f"Filling Keycloak login credentials for {username}...")
await page.wait_for_selector("input#username", timeout=10000)
await page.fill("input#username", username)
await page.fill("input#password", password)
logger.info("Submitting Keycloak login form...")
# Submit the form and wait for navigation
# Use JavaScript to submit the form directly (more reliable than clicking button)
async with page.expect_navigation(timeout=30000):
await page.evaluate("document.querySelector('form').submit()")
logger.info(f"Keycloak login submitted for {username}, redirected to callback")
# Check if we need to handle consent screen
# Keycloak consent screen has "Yes" button
consent_button = page.locator('input[name="accept"][value="Yes"]')
if await consent_button.count() > 0:
logger.info("Keycloak consent screen detected, clicking Yes...")
await consent_button.click()
await page.wait_for_load_state("networkidle", timeout=30000)
logger.info("Keycloak consent granted")
# Wait for callback server to receive auth code with timeout
logger.info(f"Waiting for auth code with state: {state[:16]}...")
timeout = 30 # seconds
start_time = time.time()
auth_code = None
while time.time() - start_time < timeout:
if state in auth_states:
auth_code = auth_states[state]
logger.info("Auth code received from callback server")
break
await anyio.sleep(0.1)
else:
raise TimeoutError(
f"Auth code not received within {timeout}s. State: {state[:16]}..."
)
finally:
await context.close()
# Exchange authorization code for access token (with PKCE code_verifier)
logger.info("Exchanging authorization code for access token with PKCE...")
async with httpx.AsyncClient(timeout=30.0) as token_client:
token_response = await token_client.post(
token_endpoint,
data={
"grant_type": "authorization_code",
"code": auth_code,
"redirect_uri": callback_url,
"client_id": client_id,
"client_secret": client_secret,
"code_verifier": code_verifier, # PKCE verifier
},
)
token_response.raise_for_status()
token_data = token_response.json()
access_token = token_data.get("access_token")
if not access_token:
raise ValueError(f"No access_token in response: {token_data}")
logger.info(
f"Successfully obtained Keycloak OAuth access token with scopes: {scopes}"
)
return access_token
@pytest.fixture(scope="session")
async def keycloak_oauth_token(
anyio_backend, browser, keycloak_oauth_client_credentials, oauth_callback_server
) -> str:
"""
Fixture to obtain an OAuth access token from Keycloak using Playwright automation.
This fixture tests the external IdP flow where:
1. User authenticates with Keycloak (external IdP)
2. Keycloak issues an access token with Nextcloud custom scopes
3. Token is used to access Nextcloud APIs via user_oidc app validation
The Nextcloud custom scopes (notes:read, calendar:write, etc.) are now defined
in Keycloak's realm configuration and can be requested in the OAuth flow.
Returns:
OAuth access token from Keycloak for the admin user with full scopes
"""
# Standard OIDC scopes + Nextcloud custom scopes (now defined in Keycloak realm)
default_scopes = "openid profile email offline_access notes:read notes:write calendar:read calendar:write contacts:read contacts:write cookbook:read cookbook:write deck:read deck:write tables:read tables:write files:read files:write sharing:read sharing:write todo:read todo:write"
return await _get_keycloak_oauth_token(
browser,
keycloak_oauth_client_credentials,
oauth_callback_server,
scopes=default_scopes,
username="admin",
password="admin",
)
@pytest.fixture(scope="session")
async def keycloak_oauth_token_read_only(
anyio_backend, browser, keycloak_oauth_client_credentials, oauth_callback_server
) -> str:
"""
Fixture to obtain a Keycloak OAuth token with only read scopes.
This token will only be able to perform read operations and should
have write tools filtered out from the tool list.
Returns:
OAuth access token from Keycloak for test_read_only user with read-only scopes
"""
return await _get_keycloak_oauth_token(
browser,
keycloak_oauth_client_credentials,
oauth_callback_server,
scopes=DEFAULT_READ_SCOPES,
username="test_read_only",
password="test123",
)
@pytest.fixture(scope="session")
async def keycloak_oauth_token_write_only(
anyio_backend, browser, keycloak_oauth_client_credentials, oauth_callback_server
) -> str:
"""
Fixture to obtain a Keycloak OAuth token with only write scopes.
This token will only be able to perform write operations and should
have read tools filtered out from the tool list.
Returns:
OAuth access token from Keycloak for test_write_only user with write-only scopes
"""
return await _get_keycloak_oauth_token(
browser,
keycloak_oauth_client_credentials,
oauth_callback_server,
scopes=DEFAULT_WRITE_SCOPES,
username="test_write_only",
password="test123",
)
@pytest.fixture(scope="session")
async def keycloak_oauth_token_no_custom_scopes(
anyio_backend, browser, keycloak_oauth_client_credentials, oauth_callback_server
) -> str:
"""
Fixture to obtain a Keycloak OAuth token with NO custom scopes.
Tests the security behavior when a user grants only default OIDC scopes
(openid, profile, email) but declines application-specific scopes.
Expected behavior: Should see 0 tools (all tools require custom scopes).
Returns:
OAuth access token from Keycloak for test_no_scopes user with no custom scopes
"""
return await _get_keycloak_oauth_token(
browser,
keycloak_oauth_client_credentials,
oauth_callback_server,
scopes="openid profile email", # No custom scopes
username="test_no_scopes",
password="test123",
)
@pytest.fixture(scope="session")
async def nc_mcp_keycloak_client(
anyio_backend, keycloak_oauth_token
) -> AsyncGenerator[ClientSession, Any]:
"""
Session-scoped fixture providing an MCP client session authenticated with Keycloak tokens.
This MCP client connects to the mcp-keycloak service (port 8002) which is configured
to use Keycloak as an external identity provider. The token flow is:
1. Keycloak issues OAuth token (via keycloak_oauth_token fixture)
2. MCP client uses token to authenticate with MCP server
3. MCP server validates token via Nextcloud user_oidc app
4. MCP server uses validated token to access Nextcloud APIs
This tests ADR-002 external IdP integration.
Yields:
MCP client session for testing tools/resources with Keycloak auth
"""
mcp_url = "http://localhost:8002/mcp"
logger.info(f"Creating MCP client session for Keycloak external IdP at {mcp_url}")
logger.info("Using Keycloak OAuth token for authentication")
async for session in create_mcp_client_session(
url=mcp_url, token=keycloak_oauth_token, client_name="Keycloak External IdP MCP"
):
logger.info("✓ MCP client session established with Keycloak authentication")
yield session
logger.info("✓ MCP client session closed")
@pytest.fixture(scope="session")
async def nc_mcp_keycloak_client_read_only(
anyio_backend, keycloak_oauth_token_read_only
) -> AsyncGenerator[ClientSession, Any]:
"""
MCP client session authenticated with Keycloak read-only token.
This client should only see read tools and should get filtered
write tools based on token scopes.
Uses JWT tokens because they embed scope information in claims,
enabling proper scope-based tool filtering.
"""
mcp_url = "http://localhost:8002/mcp"
logger.info(f"Creating read-only MCP client session for Keycloak at {mcp_url}")
async for session in create_mcp_client_session(
url=mcp_url,
token=keycloak_oauth_token_read_only,
client_name="Keycloak Read-Only MCP",
):
yield session
@pytest.fixture(scope="session")
async def nc_mcp_keycloak_client_write_only(
anyio_backend, keycloak_oauth_token_write_only
) -> AsyncGenerator[ClientSession, Any]:
"""
MCP client session authenticated with Keycloak write-only token.
This client should only see write tools and should get filtered
read tools based on token scopes.
Uses JWT tokens because they embed scope information in claims,
enabling proper scope-based tool filtering.
"""
mcp_url = "http://localhost:8002/mcp"
logger.info(f"Creating write-only MCP client session for Keycloak at {mcp_url}")
async for session in create_mcp_client_session(
url=mcp_url,
token=keycloak_oauth_token_write_only,
client_name="Keycloak Write-Only MCP",
):
yield session
@pytest.fixture(scope="session")
async def nc_mcp_keycloak_client_no_custom_scopes(
anyio_backend, keycloak_oauth_token_no_custom_scopes
) -> AsyncGenerator[ClientSession, Any]:
"""
MCP client session authenticated with Keycloak token without custom scopes.
This client has only OIDC default scopes (openid, profile, email) without
application-specific scopes (notes:read, notes:write, etc.).
Expected behavior: Should see 0 tools (all tools require custom scopes).
Uses JWT tokens because they embed scope information in claims,
enabling proper scope-based tool filtering.
"""
mcp_url = "http://localhost:8002/mcp"
logger.info(
f"Creating no-custom-scopes MCP client session for Keycloak at {mcp_url}"
)
async for session in create_mcp_client_session(
url=mcp_url,
token=keycloak_oauth_token_no_custom_scopes,
client_name="Keycloak No Custom Scopes MCP",
):
yield session
# ========================================================================
# Astrolabe Dynamic Configuration Fixtures
# ========================================================================
# ===========================================================================================
@pytest.fixture(scope="session")
tests/unit/test_config_validators.py
-107
View File
@@ -22,16 +22,6 @@ from nextcloud_mcp_server.config_validators import (
class TestModeDetection:
"""Test auth mode detection from configuration."""
def test_token_exchange_mode_detection(self):
"""Test token exchange mode is detected."""
settings = Settings(
nextcloud_host="http://localhost",
enable_token_exchange=True,
)
mode = detect_auth_mode(settings)
assert mode == AuthMode.OAUTH_TOKEN_EXCHANGE
def test_multi_user_basic_mode_detection(self):
"""Test multi-user BasicAuth mode is detected."""
settings = Settings(
@@ -62,18 +52,6 @@ class TestModeDetection:
mode = detect_auth_mode(settings)
assert mode == AuthMode.OAUTH_SINGLE_AUDIENCE
def test_mode_priority_token_exchange_over_basic(self):
"""Test token exchange has priority over BasicAuth."""
settings = Settings(
nextcloud_host="http://localhost",
nextcloud_username="admin",
nextcloud_password="password",
enable_token_exchange=True,
)
mode = detect_auth_mode(settings)
assert mode == AuthMode.OAUTH_TOKEN_EXCHANGE
class TestSingleUserBasicValidation:
"""Test validation for single-user BasicAuth mode."""
@@ -165,21 +143,6 @@ class TestSingleUserBasicValidation:
# It will fail multi-user validation because username/password are forbidden
assert len(errors) > 0
def test_forbidden_token_exchange(self):
"""Test error when ENABLE_TOKEN_EXCHANGE is set."""
settings = Settings(
nextcloud_host="http://localhost",
nextcloud_username="admin",
nextcloud_password="password",
enable_token_exchange=True,
)
# Note: This will detect as OAUTH_TOKEN_EXCHANGE due to priority
mode, errors = validate_configuration(settings)
assert mode == AuthMode.OAUTH_TOKEN_EXCHANGE
# It will fail OAuth validation
def test_vector_sync_without_embedding_provider_uses_fallback(self):
"""Test that vector sync works with Simple provider fallback (no config needed)."""
settings = Settings(
@@ -419,51 +382,6 @@ class TestOAuthSingleAudienceValidation:
assert settings.enable_offline_access is True
class TestOAuthTokenExchangeValidation:
"""Test validation for OAuth token exchange mode."""
def test_valid_minimal_config(self):
"""Test valid minimal OAuth token exchange config."""
settings = Settings(
nextcloud_host="http://localhost",
enable_token_exchange=True,
)
mode, errors = validate_configuration(settings)
assert mode == AuthMode.OAUTH_TOKEN_EXCHANGE
assert len(errors) == 0
def test_valid_with_credentials(self):
"""Test valid config with OAuth credentials."""
settings = Settings(
nextcloud_host="http://localhost",
enable_token_exchange=True,
oidc_client_id="test-client",
oidc_client_secret="test-secret",
)
mode, errors = validate_configuration(settings)
assert mode == AuthMode.OAUTH_TOKEN_EXCHANGE
assert len(errors) == 0
def test_forbidden_username_password(self):
"""Test error when username/password are set."""
settings = Settings(
nextcloud_host="http://localhost",
enable_token_exchange=True,
nextcloud_username="admin",
nextcloud_password="password",
)
mode, errors = validate_configuration(settings)
assert mode == AuthMode.OAUTH_TOKEN_EXCHANGE
assert any("nextcloud_username" in err.lower() for err in errors)
assert any("nextcloud_password" in err.lower() for err in errors)
class TestModeSummary:
"""Test mode summary generation."""
@@ -477,14 +395,6 @@ class TestModeSummary:
assert "NEXTCLOUD_PASSWORD" in summary
assert "VECTOR_SYNC_ENABLED" in summary
def test_oauth_token_exchange_summary(self):
"""Test summary for OAuth token exchange mode."""
summary = get_mode_summary(AuthMode.OAUTH_TOKEN_EXCHANGE)
assert "oauth_exchange" in summary
assert "ENABLE_TOKEN_EXCHANGE" in summary
assert "RFC 8693" in summary
class TestEdgeCases:
"""Test edge cases and boundary conditions."""
@@ -800,23 +710,6 @@ class TestExplicitModeSelection:
assert mode == AuthMode.OAUTH_SINGLE_AUDIENCE
def test_explicit_oauth_token_exchange_mode(self):
"""Test explicit oauth_token_exchange mode selection."""
with patch.dict(
os.environ,
{
"NEXTCLOUD_HOST": "http://localhost:8080",
"MCP_DEPLOYMENT_MODE": "oauth_token_exchange",
},
clear=True,
):
from nextcloud_mcp_server.config import get_settings
settings = get_settings()
mode = detect_auth_mode(settings)
assert mode == AuthMode.OAUTH_TOKEN_EXCHANGE
def test_invalid_deployment_mode_raises_error(self):
"""Test invalid MCP_DEPLOYMENT_MODE raises ValueError."""
with patch.dict(
tests/unit/test_management_status_endpoint.py
-2
View File
@@ -37,7 +37,6 @@ def create_mock_settings(
oidc_issuer: str | None = None,
vector_sync_enabled: bool = False,
nextcloud_url: str = "http://localhost",
enable_token_exchange: bool = False,
mcp_client_id: str | None = None,
mcp_client_secret: str | None = None,
):
@@ -49,7 +48,6 @@ def create_mock_settings(
settings.oidc_issuer = oidc_issuer
settings.vector_sync_enabled = vector_sync_enabled
settings.nextcloud_url = nextcloud_url
settings.enable_token_exchange = enable_token_exchange
settings.mcp_client_id = mcp_client_id
settings.mcp_client_secret = mcp_client_secret
return settings
tests/unit/test_unified_verifier.py
+20 -29
View File
@@ -29,18 +29,9 @@ def base_settings():
nextcloud_resource_uri="http://localhost:8080",
jwks_uri="https://idp.example.com/jwks",
introspection_uri="https://idp.example.com/introspect",
enable_token_exchange=False, # Multi-audience mode
token_exchange_cache_ttl=300,
)
@pytest.fixture
def exchange_settings(base_settings):
"""Create settings for token exchange mode."""
base_settings.enable_token_exchange = True
return base_settings
class TestUnifiedTokenVerifierInit:
"""Test UnifiedTokenVerifier initialization."""
@@ -50,11 +41,11 @@ class TestUnifiedTokenVerifierInit:
assert verifier.mode == "multi-audience"
assert verifier.settings == base_settings
def test_init_exchange_mode(self, exchange_settings):
"""Test verifier initialization in token exchange mode."""
verifier = UnifiedTokenVerifier(exchange_settings)
assert verifier.mode == "exchange"
assert verifier.settings == exchange_settings
def test_init_always_multi_audience(self, base_settings):
"""Test verifier always initializes in multi-audience mode."""
verifier = UnifiedTokenVerifier(base_settings)
assert verifier.mode == "multi-audience"
assert verifier.settings == base_settings
class TestAudienceValidation:
@@ -117,9 +108,9 @@ class TestAudienceValidation:
# Should pass - we only validate MCP audience per RFC 7519
assert verifier._has_mcp_audience(payload) is True
def test_has_mcp_audience_with_client_id(self, exchange_settings):
def test_has_mcp_audience_with_client_id(self, base_settings):
"""Test MCP audience validation with client ID."""
verifier = UnifiedTokenVerifier(exchange_settings)
verifier = UnifiedTokenVerifier(base_settings)
payload = {
"aud": ["test-client-id"],
"sub": "testuser",
@@ -128,9 +119,9 @@ class TestAudienceValidation:
assert verifier._has_mcp_audience(payload) is True
def test_has_mcp_audience_with_server_url(self, exchange_settings):
def test_has_mcp_audience_with_server_url(self, base_settings):
"""Test MCP audience validation with server URL."""
verifier = UnifiedTokenVerifier(exchange_settings)
verifier = UnifiedTokenVerifier(base_settings)
payload = {
"aud": ["http://localhost:8000"],
"sub": "testuser",
@@ -139,9 +130,9 @@ class TestAudienceValidation:
assert verifier._has_mcp_audience(payload) is True
def test_has_mcp_audience_missing(self, exchange_settings):
def test_has_mcp_audience_missing(self, base_settings):
"""Test MCP audience validation fails without MCP audience."""
verifier = UnifiedTokenVerifier(exchange_settings)
verifier = UnifiedTokenVerifier(base_settings)
payload = {
"aud": ["http://localhost:8080"], # Wrong audience
"sub": "testuser",
@@ -292,12 +283,12 @@ class TestMultiAudienceVerification:
assert result.resource == "testuser"
class TestExchangeModeVerification:
"""Test token exchange mode verification."""
class TestMcpAudienceVerification:
"""Test MCP audience verification."""
async def test_verify_mcp_audience_only_success(self, exchange_settings):
async def test_verify_mcp_audience_only_success(self, base_settings):
"""Test MCP-only audience verification succeeds with MCP audience."""
verifier = UnifiedTokenVerifier(exchange_settings)
verifier = UnifiedTokenVerifier(base_settings)
# Mock introspection response with MCP audience only
introspection_response = {
@@ -318,9 +309,9 @@ class TestExchangeModeVerification:
assert result is not None
assert result.resource == "testuser"
async def test_verify_mcp_audience_only_fails_without_mcp(self, exchange_settings):
async def test_verify_mcp_audience_only_fails_without_mcp(self, base_settings):
"""Test MCP audience verification fails without MCP audience."""
verifier = UnifiedTokenVerifier(exchange_settings)
verifier = UnifiedTokenVerifier(base_settings)
# Mock introspection response without MCP audience
introspection_response = {
@@ -503,9 +494,9 @@ class TestVerifyTokenFlow:
assert result is not None
assert result.resource == "testuser"
async def test_verify_token_exchange_mode(self, exchange_settings):
"""Test verify_token in exchange mode."""
verifier = UnifiedTokenVerifier(exchange_settings)
async def test_verify_token_mcp_audience_only(self, base_settings):
"""Test verify_token with MCP audience only."""
verifier = UnifiedTokenVerifier(base_settings)
introspection_response = {
"active": True,