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

fix: address PR review — remove token exchange tests, improve logging

Browse Source 
- Remove all RFC 8693 token exchange tests (integration, manual, keycloak)
  since Nextcloud doesn't support bearer tokens without upstream patches
- Remove manual impersonation/ADR-004 scripts and their docs
- Clean up token_exchange singleton from integration conftest
- Improve logging in _complete_login_flow_v2_as_user with step-by-step
  [username] prefixed messages matching _complete_login_flow_v2 style
- Remove unnecessary time staggering from all_login_flow_user_tokens;
  concurrent token acquisition works without artificial delays

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2026-04-01 17:43:44 +02:00
co-authored by Claude Opus 4.6
parent 6278b6eb75
commit c6316dbb91
13 changed files with 40 additions and 2782 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/integration/auth/test_token_exchange_legacy_v1.py
-308
View File
@@ -1,308 +0,0 @@
"""
Integration test for RFC 8693 Token Exchange - Legacy V1 (Impersonation/Tier 1).
Tests the advanced impersonation feature where the service account token is
exchanged for a token with the target user's identity (sub claim changes).
This requires:
1. Keycloak with --features=preview enabled
2. Impersonation role granted to the service account
⚠️ This test will SKIP if impersonation permissions are not configured.
Configuration (one-time setup):
# Grant impersonation role
docker compose exec keycloak /opt/keycloak/bin/kcadm.sh config credentials \\
--server http://localhost:8080 \\
--realm master \\
--user admin \\
--password admin
docker compose exec keycloak /opt/keycloak/bin/kcadm.sh add-roles \\
-r nextcloud-mcp \\
--uusername service-account-nextcloud-mcp-server \\
--cclientid realm-management \\
--rolename impersonation
Usage:
pytest tests/integration/auth/test_token_exchange_legacy_v1.py -v
"""
import base64
import json
import os
import httpx
import pytest
pytestmark = [pytest.mark.integration, pytest.mark.anyio, pytest.mark.keycloak]
def decode_jwt(token: str) -> dict:
"""Decode JWT token payload without verification."""
try:
parts = token.split(".")
if len(parts) != 3:
return {"error": "Invalid JWT format"}
payload = parts[1]
padding = 4 - (len(payload) % 4)
if padding != 4:
payload += "=" * padding
decoded = base64.urlsafe_b64decode(payload)
return json.loads(decoded)
except Exception as e:
return {"error": str(e)}
@pytest.fixture
def keycloak_config():
"""Keycloak configuration for testing."""
return {
"url": os.getenv("KEYCLOAK_URL", "http://localhost:8888"),
"realm": os.getenv("KEYCLOAK_REALM", "nextcloud-mcp"),
"client_id": os.getenv("KEYCLOAK_CLIENT_ID", "nextcloud-mcp-server"),
"client_secret": os.getenv(
"KEYCLOAK_CLIENT_SECRET", "mcp-secret-change-in-production"
),
"token_endpoint": f"{os.getenv('KEYCLOAK_URL', 'http://localhost:8888')}/realms/{os.getenv('KEYCLOAK_REALM', 'nextcloud-mcp')}/protocol/openid-connect/token",
}
@pytest.fixture
async def service_account_token(keycloak_config):
"""Get a service account token using client_credentials grant."""
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
keycloak_config["token_endpoint"],
data={
"grant_type": "client_credentials",
"client_id": keycloak_config["client_id"],
"client_secret": keycloak_config["client_secret"],
"scope": "openid profile email",
},
)
response.raise_for_status()
token_data = response.json()
return token_data["access_token"]
async def test_token_exchange_impersonation_requires_permissions(
keycloak_config, service_account_token
):
"""Test that impersonation requires explicit permission grant.
This test documents that Legacy V1 impersonation is opt-in and requires
administrative configuration via Keycloak CLI.
"""
target_user = "admin" # User to impersonate
async with httpx.AsyncClient(timeout=30.0) as client:
exchange_response = await client.post(
keycloak_config["token_endpoint"],
data={
"grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
"client_id": keycloak_config["client_id"],
"client_secret": keycloak_config["client_secret"],
"subject_token": service_account_token,
"subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
"requested_token_type": "urn:ietf:params:oauth:token-type:access_token",
"requested_subject": target_user, # ← KEY: Request impersonation
},
)
# If permissions not granted, we expect 403 Forbidden
if exchange_response.status_code == 403:
pytest.skip(
"Impersonation permissions not configured. "
"Run tests/manual/configure_impersonation.py or grant manually via Keycloak CLI. "
"See test docstring for configuration commands."
)
# If permissions are granted, exchange should succeed
assert exchange_response.status_code == 200, (
f"Token exchange failed: {exchange_response.status_code} {exchange_response.text}"
)
async def test_token_exchange_impersonation_changes_subject(
keycloak_config, service_account_token
):
"""Test Legacy V1 impersonation - subject claim should change."""
target_user = "admin"
# Decode service account token
service_claims = decode_jwt(service_account_token)
assert "error" not in service_claims
service_sub = service_claims["sub"]
assert "service-account" in service_sub.lower()
# Exchange token WITH requested_subject (Legacy V1 impersonation)
async with httpx.AsyncClient(timeout=30.0) as client:
exchange_response = await client.post(
keycloak_config["token_endpoint"],
data={
"grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
"client_id": keycloak_config["client_id"],
"client_secret": keycloak_config["client_secret"],
"subject_token": service_account_token,
"subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
"requested_token_type": "urn:ietf:params:oauth:token-type:access_token",
"requested_subject": target_user, # ← KEY: Impersonate admin
},
)
# Skip if permissions not configured
if exchange_response.status_code == 403:
pytest.skip(
"Impersonation permissions not configured. "
"See test docstring for setup instructions."
)
# Token exchange should succeed with permissions
assert exchange_response.status_code == 200, (
f"Token exchange failed: {exchange_response.status_code} {exchange_response.text}"
)
exchanged_data = exchange_response.json()
assert "access_token" in exchanged_data
exchanged_token = exchanged_data["access_token"]
# Decode exchanged token
exchanged_claims = decode_jwt(exchanged_token)
assert "error" not in exchanged_claims
exchanged_sub = exchanged_claims["sub"]
# CRITICAL: Verify impersonation - sub claim MUST change
assert service_sub != exchanged_sub, (
f"Impersonation should change subject claim. "
f"Original: {service_sub}, Exchanged: {exchanged_sub}"
)
# Verify the new token represents the target user
assert "preferred_username" in exchanged_claims
assert exchanged_claims["preferred_username"] == target_user
async def test_impersonated_token_with_nextcloud(
keycloak_config, service_account_token
):
"""Test that impersonated token works with Nextcloud APIs."""
target_user = "admin"
nextcloud_host = os.getenv("NEXTCLOUD_HOST", "http://localhost:8080")
# Exchange token with impersonation
async with httpx.AsyncClient(timeout=30.0) as client:
exchange_response = await client.post(
keycloak_config["token_endpoint"],
data={
"grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
"client_id": keycloak_config["client_id"],
"client_secret": keycloak_config["client_secret"],
"subject_token": service_account_token,
"subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
"requested_token_type": "urn:ietf:params:oauth:token-type:access_token",
"requested_subject": target_user,
},
)
# Skip if permissions not configured
if exchange_response.status_code == 403:
pytest.skip("Impersonation permissions not configured.")
exchange_response.raise_for_status()
exchanged_token = exchange_response.json()["access_token"]
# Test with Nextcloud API
nc_response = await client.get(
f"{nextcloud_host}/ocs/v2.php/cloud/capabilities",
headers={"Authorization": f"Bearer {exchanged_token}"},
)
# Should get valid response from Nextcloud
assert nc_response.status_code in [
200,
401,
], f"Unexpected status: {nc_response.status_code}"
if nc_response.status_code == 200:
# Token was accepted - verify we got a valid response
# Nextcloud OCS API can return XML or JSON
assert len(nc_response.content) > 0, "Response should not be empty"
content_type = nc_response.headers.get("content-type", "")
assert any(t in content_type for t in ["json", "xml"]), (
f"Unexpected content type: {content_type}"
)
async def test_standard_v2_rejects_requested_subject():
"""Verify that Standard V2 (without preview features) rejects requested_subject.
This test documents the key difference between Standard V2 and Legacy V1.
NOTE: This test will PASS if preview features are enabled, as Keycloak
accepts the parameter in Legacy V1 mode. The test exists to document the
expected behavior when preview features are DISABLED.
"""
keycloak_url = os.getenv("KEYCLOAK_URL", "http://localhost:8888")
realm = os.getenv("KEYCLOAK_REALM", "nextcloud-mcp")
client_id = os.getenv("KEYCLOAK_CLIENT_ID", "nextcloud-mcp-server")
client_secret = os.getenv(
"KEYCLOAK_CLIENT_SECRET", "mcp-secret-change-in-production"
)
token_endpoint = f"{keycloak_url}/realms/{realm}/protocol/openid-connect/token"
async with httpx.AsyncClient(timeout=30.0) as client:
# Get service account token
token_response = await client.post(
token_endpoint,
data={
"grant_type": "client_credentials",
"client_id": client_id,
"client_secret": client_secret,
"scope": "openid profile email",
},
)
token_response.raise_for_status()
service_token = token_response.json()["access_token"]
# Try token exchange with requested_subject
exchange_response = await client.post(
token_endpoint,
data={
"grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
"client_id": client_id,
"client_secret": client_secret,
"subject_token": service_token,
"subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
"requested_token_type": "urn:ietf:params:oauth:token-type:access_token",
"requested_subject": "admin", # Try to impersonate
},
)
# Standard V2: expects 400 Bad Request with "not supported" message
# Legacy V1: accepts parameter, returns 200 or 403 (depending on permissions)
if exchange_response.status_code == 400:
# Standard V2 behavior
error_data = exchange_response.json()
assert (
"requested_subject" in error_data.get("error_description", "").lower()
)
# Test passes - Standard V2 correctly rejects the parameter
elif exchange_response.status_code in [200, 403]:
# Legacy V1 behavior - parameter is accepted
pytest.skip(
"Preview features enabled - Keycloak is in Legacy V1 mode. "
"This test documents Standard V2 behavior which rejects requested_subject."
)
else:
pytest.fail(
f"Unexpected status code: {exchange_response.status_code}. "
f"Expected 400 (Standard V2) or 200/403 (Legacy V1)"
)
tests/integration/auth/test_token_exchange_standard_v2.py
-222
View File
@@ -1,222 +0,0 @@
"""
Integration test for RFC 8693 Token Exchange - Standard V2 (Delegation/Tier 2).
Tests the production-ready token exchange without impersonation.
The service account exchanges its token for a user-scoped token while
maintaining its own identity (sub claim unchanged).
This is the RECOMMENDED approach for most use cases.
Requirements:
- Keycloak container running (can be Standard V2 or Legacy V1)
- MCP Keycloak service running on port 8002
Usage:
pytest tests/integration/auth/test_token_exchange_standard_v2.py -v
"""
import base64
import json
import os
import httpx
import pytest
pytestmark = [pytest.mark.integration, pytest.mark.anyio, pytest.mark.keycloak]
def decode_jwt(token: str) -> dict:
"""Decode JWT token payload without verification."""
try:
parts = token.split(".")
if len(parts) != 3:
return {"error": "Invalid JWT format"}
payload = parts[1]
padding = 4 - (len(payload) % 4)
if padding != 4:
payload += "=" * padding
decoded = base64.urlsafe_b64decode(payload)
return json.loads(decoded)
except Exception as e:
return {"error": str(e)}
@pytest.fixture
def keycloak_config():
"""Keycloak configuration for testing."""
return {
"url": os.getenv("KEYCLOAK_URL", "http://localhost:8888"),
"realm": os.getenv("KEYCLOAK_REALM", "nextcloud-mcp"),
"client_id": os.getenv("KEYCLOAK_CLIENT_ID", "nextcloud-mcp-server"),
"client_secret": os.getenv(
"KEYCLOAK_CLIENT_SECRET", "mcp-secret-change-in-production"
),
"token_endpoint": f"{os.getenv('KEYCLOAK_URL', 'http://localhost:8888')}/realms/{os.getenv('KEYCLOAK_REALM', 'nextcloud-mcp')}/protocol/openid-connect/token",
}
@pytest.fixture
async def service_account_token(keycloak_config):
"""Get a service account token using client_credentials grant."""
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
keycloak_config["token_endpoint"],
data={
"grant_type": "client_credentials",
"client_id": keycloak_config["client_id"],
"client_secret": keycloak_config["client_secret"],
"scope": "openid profile email",
},
)
response.raise_for_status()
token_data = response.json()
return token_data["access_token"]
async def test_token_exchange_delegation(keycloak_config, service_account_token):
"""Test Standard V2 token exchange with delegation (no impersonation)."""
# Decode service account token to get original claims
service_claims = decode_jwt(service_account_token)
assert "error" not in service_claims, "Failed to decode service account token"
assert "sub" in service_claims
service_sub = service_claims["sub"]
# Exchange token WITHOUT requested_subject (Standard V2 delegation)
async with httpx.AsyncClient(timeout=30.0) as client:
exchange_response = await client.post(
keycloak_config["token_endpoint"],
data={
"grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
"client_id": keycloak_config["client_id"],
"client_secret": keycloak_config["client_secret"],
"subject_token": service_account_token,
"subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
"requested_token_type": "urn:ietf:params:oauth:token-type:access_token",
# NOTE: NO requested_subject parameter - this is delegation, not impersonation
},
)
# Token exchange should succeed
assert exchange_response.status_code == 200, (
f"Token exchange failed: {exchange_response.status_code} {exchange_response.text}"
)
exchanged_data = exchange_response.json()
assert "access_token" in exchanged_data
assert "token_type" in exchanged_data
assert exchanged_data["token_type"].lower() == "bearer"
exchanged_token = exchanged_data["access_token"]
# Decode exchanged token
exchanged_claims = decode_jwt(exchanged_token)
assert "error" not in exchanged_claims, "Failed to decode exchanged token"
assert "sub" in exchanged_claims
exchanged_sub = exchanged_claims["sub"]
# CRITICAL: Verify delegation behavior - sub claim should NOT change
assert service_sub == exchanged_sub, (
f"Subject should remain unchanged in delegation (service account identity preserved). Original: {service_sub}, Exchanged: {exchanged_sub}"
)
# The exchanged token should still identify as the service account
assert "service-account" in exchanged_sub.lower(), (
"Exchanged token should maintain service account identity"
)
async def test_exchanged_token_with_nextcloud(keycloak_config, service_account_token):
"""Test that exchanged token works with Nextcloud APIs."""
nextcloud_host = os.getenv("NEXTCLOUD_HOST", "http://localhost:8080")
# Exchange the service account token
async with httpx.AsyncClient(timeout=30.0) as client:
exchange_response = await client.post(
keycloak_config["token_endpoint"],
data={
"grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
"client_id": keycloak_config["client_id"],
"client_secret": keycloak_config["client_secret"],
"subject_token": service_account_token,
"subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
"requested_token_type": "urn:ietf:params:oauth:token-type:access_token",
},
)
exchange_response.raise_for_status()
exchanged_token = exchange_response.json()["access_token"]
# Test the exchanged token with Nextcloud API
nc_response = await client.get(
f"{nextcloud_host}/ocs/v2.php/cloud/capabilities",
headers={"Authorization": f"Bearer {exchanged_token}"},
)
# Should get a valid response from Nextcloud
# Note: This might fail with 401 if user_oidc doesn't accept the token
# That's expected - this test verifies the token exchange itself works
assert nc_response.status_code in [
200,
401,
], f"Unexpected status: {nc_response.status_code}"
if nc_response.status_code == 200:
# Token was accepted - verify we got a valid response
# Nextcloud OCS API can return XML or JSON
assert len(nc_response.content) > 0, "Response should not be empty"
# Verify we got either JSON or XML capabilities response
content_type = nc_response.headers.get("content-type", "")
assert any(t in content_type for t in ["json", "xml"]), (
f"Unexpected content type: {content_type}"
)
async def test_token_exchange_without_permissions_should_work():
"""Verify Standard V2 doesn't require special permissions (unlike Legacy V1 impersonation)."""
# This test documents that Standard V2 token exchange works out-of-the-box
# without needing to grant impersonation roles via Keycloak CLI
keycloak_url = os.getenv("KEYCLOAK_URL", "http://localhost:8888")
realm = os.getenv("KEYCLOAK_REALM", "nextcloud-mcp")
client_id = os.getenv("KEYCLOAK_CLIENT_ID", "nextcloud-mcp-server")
client_secret = os.getenv(
"KEYCLOAK_CLIENT_SECRET", "mcp-secret-change-in-production"
)
token_endpoint = f"{keycloak_url}/realms/{realm}/protocol/openid-connect/token"
async with httpx.AsyncClient(timeout=30.0) as client:
# Get service account token
token_response = await client.post(
token_endpoint,
data={
"grant_type": "client_credentials",
"client_id": client_id,
"client_secret": client_secret,
"scope": "openid profile email",
},
)
token_response.raise_for_status()
service_token = token_response.json()["access_token"]
# Exchange token - should work without any special role grants
exchange_response = await client.post(
token_endpoint,
data={
"grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
"client_id": client_id,
"client_secret": client_secret,
"subject_token": service_token,
"subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
"requested_token_type": "urn:ietf:params:oauth:token-type:access_token",
},
)
# Should succeed without 403 Forbidden (no permission requirements)
assert exchange_response.status_code == 200, (
f"Standard V2 delegation should work without special permissions. "
f"Got: {exchange_response.status_code} {exchange_response.text}"
)
tests/integration/conftest.py
-4
View File
@@ -43,7 +43,6 @@ async def reset_all_singletons():
# Import all modules with singletons
import nextcloud_mcp_server.app as app_module
import nextcloud_mcp_server.auth.client_registry as client_registry_module
import nextcloud_mcp_server.auth.token_exchange as token_exchange_module
import nextcloud_mcp_server.embedding.service as embedding_module
import nextcloud_mcp_server.observability.tracing as tracing_module
import nextcloud_mcp_server.providers.registry as registry_module
@@ -63,7 +62,6 @@ async def reset_all_singletons():
),
"tracer": tracing_module._tracer,
"registry": client_registry_module._registry,
"token_exchange_service": token_exchange_module._token_exchange_service,
}
# Close any open memory streams before reset
@@ -89,7 +87,6 @@ async def reset_all_singletons():
app_module._vector_sync_state.scanner_wake_event = None
tracing_module._tracer = None
client_registry_module._registry = None
token_exchange_module._token_exchange_service = None
logger.debug("All singletons reset for test module")
@@ -115,4 +112,3 @@ async def reset_all_singletons():
) = originals["vector_sync_state"]
tracing_module._tracer = originals["tracer"]
client_registry_module._registry = originals["registry"]
token_exchange_module._token_exchange_service = originals["token_exchange_service"]
tests/integration/test_keycloak_token_exchange.py
-380
View File
@@ -1,380 +0,0 @@
"""Integration tests for RFC 8693 Token Exchange with Keycloak.
These tests validate the complete token exchange flow:
1. Obtain client token from Keycloak
2. Exchange for Nextcloud-audience token via RFC 8693
3. Use exchanged token to access Nextcloud APIs
4. Verify CRUD operations work with exchanged tokens
Requirements:
- Keycloak running with nextcloud-mcp realm configured
- Nextcloud running with user_oidc app configured
- Standard Token Exchange enabled on both clients
- token-exchange-nextcloud scope configured
"""
from typing import Any
import httpx
import jwt
import pytest
@pytest.fixture
async def keycloak_base_url() -> str:
"""Keycloak base URL (external)."""
return "http://localhost:8888"
@pytest.fixture
async def keycloak_token_url(keycloak_base_url: str) -> str:
"""Keycloak token endpoint URL."""
return f"{keycloak_base_url}/realms/nextcloud-mcp/protocol/openid-connect/token"
@pytest.fixture
async def nextcloud_base_url() -> str:
"""Nextcloud base URL."""
return "http://localhost:8080"
@pytest.fixture
async def http_client() -> httpx.AsyncClient:
"""Async HTTP client for API requests."""
async with httpx.AsyncClient(timeout=30.0, follow_redirects=True) as client:
yield client
@pytest.fixture
async def keycloak_client_token(
http_client: httpx.AsyncClient, keycloak_token_url: str
) -> str:
"""Get client token from Keycloak using password grant.
Returns token with aud: ["nextcloud-mcp-server", "nextcloud"]
"""
response = await http_client.post(
keycloak_token_url,
data={
"grant_type": "password",
"client_id": "nextcloud-mcp-server",
"client_secret": "mcp-secret-change-in-production",
"username": "admin",
"password": "admin",
"scope": "openid profile email offline_access notes:read notes:write",
},
)
response.raise_for_status()
token_data = response.json()
return token_data["access_token"]
async def exchange_token(
http_client: httpx.AsyncClient,
token_url: str,
subject_token: str,
audience: str = "nextcloud",
) -> dict[str, Any]:
"""Exchange token using RFC 8693.
Args:
http_client: HTTP client
token_url: Token endpoint URL
subject_token: Token to exchange
audience: Target audience
Returns:
Token response with access_token and expires_in
"""
response = await http_client.post(
token_url,
data={
"grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
"client_id": "nextcloud-mcp-server",
"client_secret": "mcp-secret-change-in-production",
"subject_token": subject_token,
"subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
"requested_token_type": "urn:ietf:params:oauth:token-type:access_token",
"audience": audience,
},
)
response.raise_for_status()
return response.json()
def decode_token_claims(token: str) -> dict[str, Any]:
"""Decode JWT token claims without verification.
Args:
token: JWT token
Returns:
Token claims
"""
return jwt.decode(token, options={"verify_signature": False})
@pytest.mark.integration
@pytest.mark.keycloak
class TestKeycloakTokenExchange:
"""Test RFC 8693 Token Exchange with Keycloak."""
async def test_token_exchange_basic(
self,
http_client: httpx.AsyncClient,
keycloak_token_url: str,
keycloak_client_token: str,
):
"""Test basic token exchange flow."""
# Verify initial token has both audiences
initial_claims = decode_token_claims(keycloak_client_token)
assert "nextcloud-mcp-server" in initial_claims["aud"]
assert "nextcloud" in initial_claims["aud"]
assert initial_claims["azp"] == "nextcloud-mcp-server"
# Exchange for Nextcloud-audience token
exchange_response = await exchange_token(
http_client, keycloak_token_url, keycloak_client_token
)
assert "access_token" in exchange_response
assert "expires_in" in exchange_response
assert exchange_response["expires_in"] > 0
# Verify exchanged token has correct audience
exchanged_token = exchange_response["access_token"]
exchanged_claims = decode_token_claims(exchanged_token)
assert exchanged_claims["aud"] == "nextcloud"
assert exchanged_claims["azp"] == "nextcloud-mcp-server"
assert exchanged_claims["sub"] == initial_claims["sub"]
async def test_token_exchange_with_nextcloud_api(
self,
http_client: httpx.AsyncClient,
keycloak_token_url: str,
keycloak_client_token: str,
nextcloud_base_url: str,
):
"""Test exchanged token works with Nextcloud APIs."""
# Exchange token
exchange_response = await exchange_token(
http_client, keycloak_token_url, keycloak_client_token
)
nextcloud_token = exchange_response["access_token"]
# Call Nextcloud Capabilities API
response = await http_client.get(
f"{nextcloud_base_url}/ocs/v1.php/cloud/capabilities",
headers={
"Authorization": f"Bearer {nextcloud_token}",
"OCS-APIRequest": "true",
},
)
response.raise_for_status()
# Verify response contains OCS data
assert "ocs" in response.text.lower()
async def test_token_exchange_multiple_times(
self,
http_client: httpx.AsyncClient,
keycloak_token_url: str,
keycloak_client_token: str,
):
"""Test multiple exchanges from same client token (stateless)."""
# Exchange token three times
tokens = []
for _ in range(3):
exchange_response = await exchange_token(
http_client, keycloak_token_url, keycloak_client_token
)
tokens.append(exchange_response["access_token"])
# All exchanges should succeed
assert len(tokens) == 3
# Tokens should be different (fresh ephemeral tokens)
# Note: Keycloak may cache, so tokens might be identical
# The important thing is that all exchanges succeeded
async def test_token_exchange_crud_operations(
self,
http_client: httpx.AsyncClient,
keycloak_token_url: str,
keycloak_client_token: str,
nextcloud_base_url: str,
):
"""Test CRUD operations with exchanged tokens."""
notes_api = f"{nextcloud_base_url}/index.php/apps/notes/api/v1/notes"
# Step 1: Exchange token for CREATE
exchange_response = await exchange_token(
http_client, keycloak_token_url, keycloak_client_token
)
create_token = exchange_response["access_token"]
# Step 2: Create a test note
create_response = await http_client.post(
notes_api,
headers={"Authorization": f"Bearer {create_token}"},
json={
"title": "Token Exchange Test",
"content": "This note was created using an RFC 8693 exchanged token!",
"category": "Test",
},
)
create_response.raise_for_status()
note_data = create_response.json()
note_id = note_data["id"]
assert note_data["title"] == "Token Exchange Test"
assert note_data["category"] == "Test"
# Step 3: Exchange token again for READ (simulate new request)
exchange_response = await exchange_token(
http_client, keycloak_token_url, keycloak_client_token
)
read_token = exchange_response["access_token"]
# Step 4: Read the note back
read_response = await http_client.get(
f"{notes_api}/{note_id}",
headers={"Authorization": f"Bearer {read_token}"},
)
read_response.raise_for_status()
read_data = read_response.json()
assert read_data["id"] == note_id
assert read_data["title"] == "Token Exchange Test"
assert "RFC 8693 exchanged token" in read_data["content"]
# Step 5: Exchange token again for DELETE
exchange_response = await exchange_token(
http_client, keycloak_token_url, keycloak_client_token
)
delete_token = exchange_response["access_token"]
# Step 6: Delete the note
delete_response = await http_client.delete(
f"{notes_api}/{note_id}",
headers={"Authorization": f"Bearer {delete_token}"},
)
# Notes API returns the deleted note or empty array
assert delete_response.status_code in (200, 204)
async def test_token_claims_preservation(
self,
http_client: httpx.AsyncClient,
keycloak_token_url: str,
keycloak_client_token: str,
):
"""Test that important claims are preserved during exchange."""
initial_claims = decode_token_claims(keycloak_client_token)
# Exchange token
exchange_response = await exchange_token(
http_client, keycloak_token_url, keycloak_client_token
)
exchanged_token = exchange_response["access_token"]
exchanged_claims = decode_token_claims(exchanged_token)
# Subject (user ID) should be preserved
assert exchanged_claims["sub"] == initial_claims["sub"]
# Authorized party should show delegation
assert exchanged_claims["azp"] == "nextcloud-mcp-server"
# Audience should be filtered to target
assert exchanged_claims["aud"] == "nextcloud"
# Token should have expiration
assert "exp" in exchanged_claims
assert exchanged_claims["exp"] > 0
async def test_token_exchange_scope_configuration(
self, http_client: httpx.AsyncClient, keycloak_token_url: str
):
"""Test that token-exchange-nextcloud scope is configured as default.
Since token-exchange-nextcloud is a default scope for nextcloud-mcp-server,
all tokens should have the nextcloud audience available for exchange.
"""
# Get a token - should automatically include default scopes
response = await http_client.post(
keycloak_token_url,
data={
"grant_type": "password",
"client_id": "nextcloud-mcp-server",
"client_secret": "mcp-secret-change-in-production",
"username": "admin",
"password": "admin",
"scope": "openid profile email",
},
)
response.raise_for_status()
token = response.json()["access_token"]
# Verify token has nextcloud in aud (from default token-exchange-nextcloud scope)
claims = decode_token_claims(token)
assert "nextcloud" in claims.get("aud", [])
# Exchange should succeed
exchange_response = await http_client.post(
keycloak_token_url,
data={
"grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
"client_id": "nextcloud-mcp-server",
"client_secret": "mcp-secret-change-in-production",
"subject_token": token,
"subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
"requested_token_type": "urn:ietf:params:oauth:token-type:access_token",
"audience": "nextcloud",
},
)
# Should succeed because token-exchange-nextcloud is a default scope
assert exchange_response.status_code == 200
exchanged_data = exchange_response.json()
assert "access_token" in exchanged_data
@pytest.mark.integration
@pytest.mark.keycloak
class TestTokenExchangeService:
"""Test the TokenExchangeService implementation."""
async def test_exchange_token_for_audience(
self, keycloak_client_token: str, keycloak_token_url: str
):
"""Test the exchange_token_for_audience function."""
from nextcloud_mcp_server.auth.token_exchange import (
TokenExchangeService,
)
# Create service
service = TokenExchangeService(
oidc_discovery_url="http://localhost:8888/realms/nextcloud-mcp/.well-known/openid-configuration",
client_id="nextcloud-mcp-server",
client_secret="mcp-secret-change-in-production",
)
try:
# Exchange token
exchanged_token, expires_in = await service.exchange_token_for_audience(
subject_token=keycloak_client_token,
requested_audience="nextcloud",
)
# Verify exchange succeeded
assert exchanged_token is not None
assert isinstance(exchanged_token, str)
assert expires_in > 0
# Verify token has correct claims
claims = decode_token_claims(exchanged_token)
assert claims["aud"] == "nextcloud"
assert claims["azp"] == "nextcloud-mcp-server"
finally:
await service.close()
tests/manual/README.md
-47
View File
@@ -1,47 +0,0 @@
# Manual OAuth Flow Testing
This directory contains manual test scripts for OAuth flows that require browser interaction.
## ADR-004 OAuth Hybrid Flow Test
The `test_adr004_oauth_flow.py` script tests the complete OAuth flow described in ADR-004.
### Prerequisites
1. **Install Playwright browsers:**
```bash
uv run playwright install firefox
```
2. **Start MCP server with OAuth enabled:**
For Nextcloud OIDC:
```bash
export ENABLE_OFFLINE_ACCESS=true
export TOKEN_ENCRYPTION_KEY=$(uv run python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())")
docker-compose up --build -d mcp-oauth
```
For Keycloak:
```bash
export ENABLE_OFFLINE_ACCESS=true
export TOKEN_ENCRYPTION_KEY=$(uv run python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())")
docker-compose up --build -d mcp-keycloak
```
### Running the Test
**Test with Nextcloud OIDC:**
```bash
uv run python tests/manual/test_adr004_oauth_flow.py --provider nextcloud
```
**Test with Keycloak:**
```bash
uv run python tests/manual/test_adr004_oauth_flow.py --provider keycloak
```
**Headless mode:**
```bash
uv run python tests/manual/test_adr004_oauth_flow.py --provider nextcloud --headless
```
tests/manual/TESTING_INSTRUCTIONS.md
-203
View File
@@ -1,203 +0,0 @@
# ADR-004 OAuth Flow Testing Instructions
## Automated Integration Test (Recommended)
The ADR-004 Hybrid Flow is now fully tested via automated integration tests using Playwright:
```bash
# Run all ADR-004 tests
uv run pytest tests/server/oauth/test_adr004_hybrid_flow.py --browser firefox -v
# Run specific test
uv run pytest tests/server/oauth/test_adr004_hybrid_flow.py::test_adr004_hybrid_flow_tool_execution --browser firefox -v
```
These tests verify:
- ✅ PKCE code challenge/verifier flow
- ✅ MCP server intercepts OAuth callback
- ✅ Master refresh token storage
- ✅ Client receives MCP access token
- ✅ MCP session establishment with hybrid flow token
- ✅ Tool execution using stored refresh tokens
- ✅ Multiple operations without re-authentication
## Manual Test (Legacy)
For manual testing or debugging, you can use the standalone test script:
```bash
# Make sure port 8765 is available
lsof -ti:8765 | xargs kill -9 2>/dev/null
# Run the test
uv run python tests/manual/test_adr004_manual.py --provider nextcloud
```
## Expected Flow
### 1. Test Script Starts
```
======================================================================
ADR-004 MANUAL OAUTH FLOW TEST
======================================================================
Provider: nextcloud
MCP Server: http://localhost:8001
Nextcloud: http://localhost:8080
======================================================================
✓ Generated PKCE challenge: gxQLsYDJ...
✓ Started callback server at http://localhost:8765/callback
```
### 2. Open OAuth URL in Browser
The script will print:
```
======================================================================
STEP 1: AUTHORIZE THE MCP SERVER
======================================================================
📋 Open this URL in your browser:
http://localhost:8001/oauth/authorize?response_type=code&...
📌 What will happen:
1. You'll be redirected to Nextcloud/Keycloak login
2. Login with username: admin, password: admin
3. You'll see a consent screen asking to authorize the MCP server
4. Click 'Authorize' or 'Allow'
5. You'll be redirected to localhost:8765/callback
6. The authorization code will appear in the terminal
```
### 3. Browser Flow
1. **Nextcloud Login** - You see the Nextcloud login page
2. **Enter Credentials** - admin/admin
3. **Consent Screen** - "Authorize Nextcloud MCP Server (jwt) to access your account?"
4. **Click Authorize**
5. **Redirect Chain**:
- Nextcloud redirects to: `http://localhost:8001/oauth/callback?code=...`
- MCP server processes the code
- MCP server redirects to: `http://localhost:8765/callback?code=mcp-code-...&state=...`
- Browser reaches the test script's callback server
- You see: "✓ Authorization Successful - You can close this window"
### 4. Test Script Continues
```
✓ Received authorization code!
Code: mcp-code-xyz...
✓ State parameter verified (CSRF protection)
======================================================================
STEP 2: EXCHANGE CODE FOR ACCESS TOKEN
======================================================================
✓ Successfully received access token
Token: eyJhbGciOiJSUzI1Ni...
Type: Bearer
Expires: 3600s
======================================================================
STEP 3: CALL MCP TOOL WITH ACCESS TOKEN
======================================================================
✓ MCP tool call succeeded!
Result: {...}
======================================================================
🎉 ADR-004 OAUTH FLOW TEST - SUCCESS
======================================================================
```
## Troubleshooting
### Browser Gets Stuck at "localhost:8765 refused to connect"
**Problem**: The callback server on port 8765 isn't accessible.
**Solutions**:
1. Check firewall isn't blocking port 8765
2. Verify the test script is still running
3. Check another process isn't using port 8765:
```bash
lsof -ti:8765
```
### Browser Shows "localhost:8765 - ERR_CONNECTION_REFUSED"
**Problem**: The callback server stopped or never started.
**Solution**:
1. Check the test script output - it should say "✓ Started callback server"
2. Restart the test script
3. Manually test the callback server:
```bash
curl http://localhost:8765/callback?code=test&state=test
```
Should return HTML page with "Authorization Successful"
### "Session not found or expired" Error
**Problem**: Took too long between steps (>10 minutes).
**Solution**: Restart the test - sessions expire after 10 minutes.
### Client ID is None
**Problem**: OAuth client credentials not loaded.
**Solution**: Rebuild the MCP server:
```bash
docker-compose up --build -d mcp-oauth
```
### Nextcloud Shows "Invalid redirect_uri"
**Problem**: The redirect URI isn't registered for the OAuth client.
**Solution**: Check registered URIs:
```bash
docker compose exec db mariadb -u root -ppassword nextcloud -e \
"SELECT c.client_identifier, r.redirect_uri FROM oc_oidc_clients c \
LEFT JOIN oc_oidc_redirect_uris r ON c.id = r.client_id \
WHERE c.name LIKE '%MCP%';"
```
Should show: `http://localhost:8001/oauth/callback`
## Manual Test Without Script
If the automated test doesn't work, you can test manually:
1. **Start callback server manually**:
```bash
python3 -m http.server 8765
```
2. **Open OAuth URL in browser** (get from test script output or build manually):
```
http://localhost:8001/oauth/authorize?response_type=code&client_id=test-mcp-client&redirect_uri=http://localhost:8765/callback&scope=openid+profile+email+offline_access&state=TEST&code_challenge=CHALLENGE&code_challenge_method=S256
```
3. **Complete login** at Nextcloud
4. **Browser should redirect** to `http://localhost:8765/callback?code=mcp-code-...&state=TEST`
5. **Copy the code** from the URL and exchange it:
```bash
curl -X POST http://localhost:8001/oauth/token \
-d "grant_type=authorization_code" \
-d "code=<MCP_CODE_HERE>" \
-d "code_verifier=<VERIFIER_HERE>" \
-d "redirect_uri=http://localhost:8765/callback" \
-d "client_id=test-mcp-client"
```
## Expected Database State After Success
```bash
# Check refresh token was stored
docker compose exec mcp-oauth sh -c \
"sqlite3 /app/data/tokens.db 'SELECT user_id, created_at FROM refresh_tokens;'"
```
Should show an entry for the authenticated user.
tests/manual/configure_impersonation.py
-195
View File
@@ -1,195 +0,0 @@
"""
Configure Keycloak client for token exchange with impersonation.
This script uses Keycloak Admin API to configure the necessary permissions
for the nextcloud-mcp-server client to impersonate users via token exchange.
Usage:
uv run python tests/manual/configure_impersonation.py
"""
import asyncio
import logging
import os
import sys
import httpx
logging.basicConfig(level=logging.INFO, format="%(levelname)-8s | %(message)s")
logger = logging.getLogger(__name__)
async def main():
"""Configure impersonation permissions in Keycloak"""
keycloak_url = os.getenv("KEYCLOAK_URL", "http://localhost:8888")
realm = os.getenv("KEYCLOAK_REALM", "nextcloud-mcp")
admin_username = "admin"
admin_password = "admin"
client_id = "nextcloud-mcp-server"
logger.info("=" * 80)
logger.info("Configuring Keycloak Impersonation Permissions")
logger.info("=" * 80)
logger.info(f"Keycloak URL: {keycloak_url}")
logger.info(f"Realm: {realm}")
logger.info(f"Client ID: {client_id}")
logger.info("")
async with httpx.AsyncClient(timeout=30.0) as client:
# Step 1: Get admin access token
logger.info("Step 1: Getting admin access token...")
token_response = await client.post(
f"{keycloak_url}/realms/master/protocol/openid-connect/token",
data={
"grant_type": "password",
"client_id": "admin-cli",
"username": admin_username,
"password": admin_password,
},
)
token_response.raise_for_status()
admin_token = token_response.json()["access_token"]
logger.info("✓ Admin token acquired")
logger.info("")
headers = {"Authorization": f"Bearer {admin_token}"}
# Step 2: Get client internal ID
logger.info("Step 2: Looking up client internal ID...")
clients_response = await client.get(
f"{keycloak_url}/admin/realms/{realm}/clients",
headers=headers,
params={"clientId": client_id},
)
clients_response.raise_for_status()
clients = clients_response.json()
if not clients:
logger.error(f"❌ Client '{client_id}' not found")
return 1
client_uuid = clients[0]["id"]
logger.info(f"✓ Found client UUID: {client_uuid}")
logger.info("")
# Step 3: Enable token exchange permission
logger.info("Step 3: Configuring token exchange permissions...")
# Get all clients (we need to allow exchange from/to any client)
all_clients_response = await client.get(
f"{keycloak_url}/admin/realms/{realm}/clients",
headers=headers,
)
all_clients_response.raise_for_status()
all_clients = all_clients_response.json()
# Get all users (we need to allow impersonation of any user)
users_response = await client.get(
f"{keycloak_url}/admin/realms/{realm}/users",
headers=headers,
)
users_response.raise_for_status()
users = users_response.json()
logger.info(f" Found {len(all_clients)} clients and {len(users)} users")
logger.info("")
# Step 4: Enable permission for client to perform token exchange
logger.info("Step 4: Enabling token exchange permission...")
# Update client to enable fine-grained permissions
update_response = await client.put(
f"{keycloak_url}/admin/realms/{realm}/clients/{client_uuid}",
headers=headers,
json={
**clients[0],
"authorizationServicesEnabled": False, # Don't need full authz
"serviceAccountsEnabled": True, # Already enabled
},
)
if update_response.status_code in [200, 204]:
logger.info("✓ Client configuration updated")
else:
logger.warning(f"⚠ Client update returned {update_response.status_code}")
logger.info("")
# Step 5: Set up token exchange permission policy
logger.info("Step 5: Configuring impersonation policy...")
# In Keycloak Legacy V1, we need to use the token-exchange permissions endpoint
# This is part of the preview features
# First, check if token exchange permissions endpoint exists
try:
perms_response = await client.get(
f"{keycloak_url}/admin/realms/{realm}/clients/{client_uuid}/token-exchange/permissions",
headers=headers,
)
if perms_response.status_code == 200:
logger.info("✓ Token exchange permissions endpoint available")
permissions = perms_response.json()
logger.info(f" Current permissions: {permissions}")
logger.info("")
# Enable impersonation for all users
logger.info("Step 6: Enabling impersonation for admin user...")
# Find admin user
admin_user = next((u for u in users if u["username"] == "admin"), None)
if admin_user:
# Enable permission for this client to impersonate admin
enable_response = await client.put(
f"{keycloak_url}/admin/realms/{realm}/users/{admin_user['id']}/impersonation",
headers=headers,
json={
"client": client_uuid,
"enabled": True,
},
)
if enable_response.status_code in [200, 204]:
logger.info("✓ Impersonation enabled for admin user")
else:
logger.warning(
f"⚠ Impersonation enable returned {enable_response.status_code}"
)
logger.info(f" Response: {enable_response.text}")
else:
logger.error("❌ Admin user not found")
elif perms_response.status_code == 404:
logger.warning("⚠ Token exchange permissions endpoint not found")
logger.info(" This might mean preview features aren't fully enabled")
logger.info(" Or the Keycloak version doesn't support this API")
else:
logger.warning(f"⚠ Unexpected response: {perms_response.status_code}")
except Exception as e:
logger.error(f"❌ Error configuring permissions: {e}")
logger.info("")
logger.info("Alternative: Manual configuration required")
logger.info(" 1. Open Keycloak Admin Console")
logger.info(" 2. Go to Clients → nextcloud-mcp-server")
logger.info(" 3. Go to Permissions tab")
logger.info(" 4. Enable 'token-exchange' permission")
logger.info(" 5. Configure permission policies for impersonation")
logger.info("")
logger.info("=" * 80)
logger.info("Configuration Complete")
logger.info("=" * 80)
logger.info("")
logger.info("Next step: Run impersonation test")
logger.info(" uv run python tests/manual/test_impersonation.py")
return 0
if __name__ == "__main__":
exit_code = asyncio.run(main())
sys.exit(exit_code)
tests/manual/test_adr004_manual.py
-319
View File
@@ -1,319 +0,0 @@
#!/usr/bin/env python3
"""
ADR-004 Manual OAuth Flow Test
This is a simplified version that doesn't use Playwright automation.
Instead, it prints URLs and waits for manual browser interaction.
Usage:
uv run python tests/manual/test_adr004_manual.py --provider nextcloud
"""
import argparse
import asyncio
import hashlib
import logging
import secrets
from base64 import urlsafe_b64encode
from http.server import BaseHTTPRequestHandler, HTTPServer
from threading import Thread
from urllib.parse import parse_qs, urlencode, urlparse
import httpx
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
class CallbackHandler(BaseHTTPRequestHandler):
"""Handles OAuth callback redirect to localhost"""
authorization_code = None
state = None
def do_GET(self):
"""Handle GET request with authorization code"""
parsed = urlparse(self.path)
params = parse_qs(parsed.query)
# Ignore favicon requests
if parsed.path == "/favicon.ico":
self.send_response(200)
self.send_header("Content-type", "image/x-icon")
self.end_headers()
return
CallbackHandler.authorization_code = params.get("code", [None])[0]
CallbackHandler.state = params.get("state", [None])[0]
# Send success page
self.send_response(200)
self.send_header("Content-type", "text/html")
self.end_headers()
code_display = (
CallbackHandler.authorization_code[:50] + "..."
if CallbackHandler.authorization_code
else "No code received"
)
html = """
<html>
<head><title>Authorization Success</title></head>
<body>
<h1 style="color: green;">✓ Authorization Successful</h1>
<p>Authorization code received. You can close this window and return to the terminal.</p>
<code style="background: #f0f0f0; padding: 10px; display: block; margin: 10px 0;">
{}
</code>
</body>
</html>
""".format(code_display)
self.wfile.write(html.encode())
def log_message(self, format, *args):
"""Log HTTP requests"""
logger.info(f"Callback server: {format % args}")
def generate_pkce_challenge():
"""Generate PKCE code verifier and challenge"""
code_verifier = secrets.token_urlsafe(32)
digest = hashlib.sha256(code_verifier.encode()).digest()
code_challenge = urlsafe_b64encode(digest).decode().rstrip("=")
return code_verifier, code_challenge
async def test_oauth_manual(
provider: str,
mcp_server_url: str,
nextcloud_host: str,
):
"""
Manual OAuth flow test - prints URLs for manual browser interaction.
"""
print("\n" + "=" * 70)
print("ADR-004 MANUAL OAUTH FLOW TEST")
print("=" * 70)
print(f"Provider: {provider}")
print(f"MCP Server: {mcp_server_url}")
print(f"Nextcloud: {nextcloud_host}")
print("=" * 70 + "\n")
# Generate PKCE challenge
code_verifier, code_challenge = generate_pkce_challenge()
logger.info(f"✓ Generated PKCE challenge: {code_challenge[:16]}...")
# Generate state for CSRF protection
state = secrets.token_urlsafe(32)
# Start local HTTP server for OAuth callback
callback_port = 8765
redirect_uri = f"http://localhost:{callback_port}/callback"
server = HTTPServer(("localhost", callback_port), CallbackHandler)
server_thread = Thread(target=server.serve_forever, daemon=True)
server_thread.start()
logger.info(f"✓ Started callback server at {redirect_uri}")
try:
# Build authorization URL
auth_params = {
"response_type": "code",
"client_id": "test-mcp-client",
"redirect_uri": redirect_uri,
"scope": "openid profile email offline_access notes:read notes:write",
"state": state,
"code_challenge": code_challenge,
"code_challenge_method": "S256",
}
auth_url = f"{mcp_server_url}/oauth/authorize?{urlencode(auth_params)}"
print("\n" + "=" * 70)
print("STEP 1: AUTHORIZE THE MCP SERVER")
print("=" * 70)
print("\n📋 Open this URL in your browser:\n")
print(f" {auth_url}")
print("\n📌 What will happen:")
print(" 1. You'll be redirected to Nextcloud/Keycloak login")
print(" 2. Login with username: admin, password: admin")
print(" 3. You'll see a consent screen asking to authorize the MCP server")
print(" 4. Click 'Authorize' or 'Allow'")
print(" 5. You'll be redirected to localhost:8765/callback")
print(" 6. The authorization code will appear in the terminal\n")
print("=" * 70)
print("\n⏳ Waiting for authorization... (timeout: 5 minutes)\n")
# Wait for authorization code (with timeout)
timeout = 300 # 5 minutes
elapsed = 0
while not CallbackHandler.authorization_code and elapsed < timeout:
await asyncio.sleep(1)
elapsed += 1
if not CallbackHandler.authorization_code:
raise RuntimeError("Timeout waiting for authorization code")
authorization_code = CallbackHandler.authorization_code
returned_state = CallbackHandler.state
print("\n✓ Received authorization code!")
logger.info(f"Code: {authorization_code[:16]}...")
# Verify state
if returned_state != state:
raise RuntimeError(
f"State mismatch! Expected {state}, got {returned_state}"
)
logger.info("✓ State parameter verified (CSRF protection)")
# Exchange authorization code for access token
print("\n" + "=" * 70)
print("STEP 2: EXCHANGE CODE FOR ACCESS TOKEN")
print("=" * 70)
async with httpx.AsyncClient() as client:
token_response = await client.post(
f"{mcp_server_url}/oauth/token",
data={
"grant_type": "authorization_code",
"code": authorization_code,
"code_verifier": code_verifier,
"redirect_uri": redirect_uri,
"client_id": "test-mcp-client",
},
timeout=30.0,
)
if token_response.status_code != 200:
print(f"\n❌ Token exchange failed: {token_response.status_code}")
print(f"Response: {token_response.text}")
raise RuntimeError("Token exchange failed")
token_data = token_response.json()
access_token = token_data["access_token"]
print("\n✓ Successfully received access token")
print(f" Token: {access_token[:30]}...")
print(f" Type: {token_data.get('token_type', 'Bearer')}")
print(f" Expires: {token_data.get('expires_in', 'unknown')}s")
# Test MCP tool call
print("\n" + "=" * 70)
print("STEP 3: CALL MCP TOOL WITH ACCESS TOKEN")
print("=" * 70)
async with httpx.AsyncClient() as client:
mcp_request = {
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "nc_notes_search_notes",
"arguments": {"query": "test"},
},
}
mcp_response = await client.post(
f"{mcp_server_url}/mcp",
json=mcp_request,
headers={
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json",
"Accept": "application/json, text/event-stream",
},
timeout=30.0,
)
if mcp_response.status_code != 200:
print(f"\n❌ MCP tool call failed: {mcp_response.status_code}")
print(f"Response: {mcp_response.text}")
raise RuntimeError("MCP tool call failed")
mcp_result = mcp_response.json()
if "error" in mcp_result:
print(f"\n❌ MCP tool returned error: {mcp_result['error']}")
raise RuntimeError(f"MCP tool error: {mcp_result['error']}")
print("\n✓ MCP tool call succeeded!")
print(f" Result: {mcp_result.get('result', {})}")
# Summary
print("\n" + "=" * 70)
print("🎉 ADR-004 OAUTH FLOW TEST - SUCCESS")
print("=" * 70)
print(f"Provider: {provider}")
print(f"MCP Server: {mcp_server_url}")
print(f"Nextcloud: {nextcloud_host}")
print("")
print("✓ User consented to MCP server access")
print("✓ User consented to offline_access (refresh tokens)")
print("✓ MCP server stored master refresh token")
print("✓ Client received MCP access token via PKCE")
print("✓ MCP tool call succeeded")
print("✓ MCP server exchanged tokens in background")
print("✓ Nextcloud data fetched successfully")
print("=" * 70 + "\n")
return {"success": True}
finally:
server.shutdown()
logger.info("Stopped callback server")
async def main():
parser = argparse.ArgumentParser(
description="Manual test for ADR-004 OAuth Hybrid Flow"
)
parser.add_argument(
"--provider",
choices=["nextcloud", "keycloak"],
required=True,
help="OAuth provider to test",
)
parser.add_argument(
"--mcp-server-url",
default="http://localhost:8001",
help="MCP server URL (default: http://localhost:8001)",
)
parser.add_argument(
"--nextcloud-host",
default="http://localhost:8080",
help="Nextcloud host URL (default: http://localhost:8080)",
)
args = parser.parse_args()
try:
result = await test_oauth_manual(
provider=args.provider,
mcp_server_url=args.mcp_server_url,
nextcloud_host=args.nextcloud_host,
)
return 0 if result["success"] else 1
except KeyboardInterrupt:
print("\n\n⚠️ Test interrupted by user")
return 1
except Exception as e:
logger.error(f"OAuth flow test failed: {e}", exc_info=True)
print("\n" + "=" * 70)
print("❌ ADR-004 OAUTH FLOW TEST - FAILED")
print("=" * 70)
print(f"Error: {e}")
print("=" * 70)
return 1
if __name__ == "__main__":
exit_code = asyncio.run(main())
exit(exit_code)
tests/manual/test_adr004_oauth_flow.py
-375
View File
@@ -1,375 +0,0 @@
#!/usr/bin/env python3
"""
ADR-004 OAuth Flow Test Script
Tests the complete Hybrid Flow implementation:
1. User initiates OAuth at MCP server /oauth/authorize
2. User consents to MCP server access (IdP)
3. User consents to MCP server accessing Nextcloud (IdP/Nextcloud)
4. MCP server receives master refresh token
5. Client receives MCP access token
6. Client calls MCP tool
7. MCP server exchanges master refresh token for Nextcloud access token
8. MCP server fetches data from Nextcloud on behalf of user
Usage:
# Test with Nextcloud OIDC app
uv run python tests/manual/test_adr004_oauth_flow.py --provider nextcloud
# Test with Keycloak
uv run python tests/manual/test_adr004_oauth_flow.py --provider keycloak
Requirements:
- MCP server running with OAuth enabled
- System web browser
"""
import argparse
import asyncio
import hashlib
import logging
import secrets
import webbrowser
from base64 import urlsafe_b64encode
from http.server import BaseHTTPRequestHandler, HTTPServer
from threading import Thread
from urllib.parse import parse_qs, urlencode, urlparse
import httpx
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
class CallbackHandler(BaseHTTPRequestHandler):
"""Handles OAuth callback redirect to localhost"""
authorization_code = None
state = None
def do_GET(self):
"""Handle GET request with authorization code"""
parsed = urlparse(self.path)
params = parse_qs(parsed.query)
# Ignore favicon requests
if parsed.path == "/favicon.ico":
self.send_response(200)
self.send_header("Content-type", "image/x-icon")
self.end_headers()
return
CallbackHandler.authorization_code = params.get("code", [None])[0]
CallbackHandler.state = params.get("state", [None])[0]
# Send success page
self.send_response(200)
self.send_header("Content-type", "text/html")
self.end_headers()
code_display = (
CallbackHandler.authorization_code[:50] + "..."
if CallbackHandler.authorization_code
else "No code received"
)
html = """
<html>
<head><title>Authorization Success</title></head>
<body>
<h1 style="color: green;">✓ Authorization Successful</h1>
<p>Authorization code received. You can close this window and return to the terminal.</p>
<code style="background: #f0f0f0; padding: 10px; display: block; margin: 10px 0;">
{}
</code>
<script>setTimeout(() => window.close(), 2000);</script>
</body>
</html>
""".format(code_display)
self.wfile.write(html.encode())
def log_message(self, format, *args):
"""Log HTTP requests"""
logger.info(f"Callback: {format % args}")
def generate_pkce_challenge():
"""Generate PKCE code verifier and challenge"""
code_verifier = secrets.token_urlsafe(32)
digest = hashlib.sha256(code_verifier.encode()).digest()
code_challenge = urlsafe_b64encode(digest).decode().rstrip("=")
return code_verifier, code_challenge
# Note: Playwright automation functions removed - using system browser instead
async def test_oauth_flow(
provider: str,
mcp_server_url: str,
nextcloud_host: str,
username: str,
password: str,
):
"""
Test complete ADR-004 OAuth flow using system browser.
Args:
provider: "nextcloud" or "keycloak"
mcp_server_url: MCP server URL (e.g., http://localhost:8001)
nextcloud_host: Nextcloud instance URL
username: Test user username (for documentation)
password: Test user password (for documentation)
"""
logger.info(f"Starting ADR-004 OAuth flow test with provider: {provider}")
logger.info(f"MCP Server: {mcp_server_url}")
logger.info(f"Nextcloud Host: {nextcloud_host}")
# Generate PKCE challenge
code_verifier, code_challenge = generate_pkce_challenge()
logger.info(f"✓ Generated PKCE challenge: {code_challenge[:16]}...")
# Generate state for CSRF protection
state = secrets.token_urlsafe(32)
# Start local HTTP server for OAuth callback
callback_port = 8765
redirect_uri = f"http://localhost:{callback_port}/callback"
server = HTTPServer(("localhost", callback_port), CallbackHandler)
server_thread = Thread(target=server.serve_forever, daemon=True)
server_thread.start()
logger.info(f"✓ Started callback server at {redirect_uri}")
try:
# Step 1: Build authorization URL
auth_params = {
"response_type": "code",
"client_id": "test-mcp-client",
"redirect_uri": redirect_uri,
"scope": "openid profile email offline_access notes:read notes:write",
"state": state,
"code_challenge": code_challenge,
"code_challenge_method": "S256",
}
auth_url = f"{mcp_server_url}/oauth/authorize?{urlencode(auth_params)}"
print("\n" + "=" * 70)
print("STEP 1: AUTHORIZE IN BROWSER")
print("=" * 70)
print(f"\n📋 Opening browser to: {auth_url[:80]}...")
print(f"\n📌 Login with: {username} / {password}")
print("📌 Then authorize the MCP server")
print("=" * 70 + "\n")
# Step 2: Open system browser
logger.info("Opening system browser for OAuth flow...")
webbrowser.open(auth_url)
logger.info("⏳ Waiting for authorization callback (timeout: 5 minutes)...")
# Wait for callback
timeout = 300 # 5 minutes
elapsed = 0
while not CallbackHandler.authorization_code and elapsed < timeout:
await asyncio.sleep(1)
elapsed += 1
if not CallbackHandler.authorization_code:
raise RuntimeError("Timeout waiting for authorization code")
# Step 3: Verify we received authorization code
authorization_code = CallbackHandler.authorization_code
returned_state = CallbackHandler.state
if not authorization_code:
raise RuntimeError("Failed to receive authorization code from callback")
logger.info(f"✓ Received MCP authorization code: {authorization_code[:16]}...")
# Verify state matches (CSRF protection)
if returned_state != state:
raise RuntimeError(
f"State mismatch! Expected {state}, got {returned_state}"
)
logger.info("✓ State parameter verified (CSRF protection)")
# Step 4: Exchange authorization code for access token
logger.info("Exchanging authorization code for access token...")
async with httpx.AsyncClient() as client:
token_response = await client.post(
f"{mcp_server_url}/oauth/token",
data={
"grant_type": "authorization_code",
"code": authorization_code,
"code_verifier": code_verifier,
"redirect_uri": redirect_uri,
"client_id": "test-mcp-client",
},
)
if token_response.status_code != 200:
logger.error(f"Token exchange failed: {token_response.status_code}")
logger.error(f"Response: {token_response.text}")
raise RuntimeError(
f"Token exchange failed: {token_response.status_code}"
)
token_data = token_response.json()
access_token = token_data["access_token"]
logger.info("✓ Successfully received access token")
logger.info(f" Token: {access_token[:20]}...")
logger.info(f" Type: {token_data.get('token_type', 'Bearer')}")
logger.info(f" Expires in: {token_data.get('expires_in', 'unknown')}s")
# Step 5: Use access token to call MCP tool
logger.info("Testing MCP tool call with access token...")
async with httpx.AsyncClient() as client:
# Call MCP server to list notes (this will trigger token exchange in background)
mcp_request = {
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "nc_notes_search_notes",
"arguments": {"query": "test"},
},
}
mcp_response = await client.post(
f"{mcp_server_url}/mcp",
json=mcp_request,
headers={
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json",
"Accept": "application/json, text/event-stream",
},
timeout=30.0,
)
if mcp_response.status_code != 200:
logger.error(f"MCP tool call failed: {mcp_response.status_code}")
logger.error(f"Response: {mcp_response.text}")
raise RuntimeError(f"MCP tool call failed: {mcp_response.status_code}")
mcp_result = mcp_response.json()
if "error" in mcp_result:
logger.error(f"MCP tool returned error: {mcp_result['error']}")
raise RuntimeError(f"MCP tool error: {mcp_result['error']}")
logger.info("✓ MCP tool call succeeded!")
logger.info(f" Result: {mcp_result.get('result', {})}")
# Step 6: Verify refresh token storage
logger.info("Verifying refresh token storage...")
# Check if refresh token was stored (requires database access)
# This would require accessing the SQLite database directly
logger.info("✓ OAuth flow completed successfully!")
# Summary
print("\n" + "=" * 70)
print("ADR-004 OAUTH FLOW TEST - SUCCESS")
print("=" * 70)
print(f"Provider: {provider}")
print(f"MCP Server: {mcp_server_url}")
print(f"Nextcloud: {nextcloud_host}")
print(f"User: {username}")
print("")
print("✓ User consented to MCP server access")
print("✓ User consented to offline_access (refresh tokens)")
print("✓ MCP server stored master refresh token")
print("✓ Client received MCP access token")
print("✓ MCP tool call succeeded")
print("✓ MCP server exchanged tokens in background")
print("✓ Nextcloud data fetched successfully")
print("=" * 70)
return {
"success": True,
"access_token": access_token,
"provider": provider,
}
finally:
server.shutdown()
logger.info("Stopped callback server")
async def main():
parser = argparse.ArgumentParser(
description="Test ADR-004 OAuth Hybrid Flow",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
# Test with Nextcloud OIDC
uv run python tests/manual/test_adr004_oauth_flow.py --provider nextcloud
# Test with Keycloak
uv run python tests/manual/test_adr004_oauth_flow.py --provider keycloak
# Headless mode
uv run python tests/manual/test_adr004_oauth_flow.py --provider nextcloud --headless
""",
)
parser.add_argument(
"--provider",
choices=["nextcloud", "keycloak"],
required=True,
help="OAuth provider to test (nextcloud or keycloak)",
)
parser.add_argument(
"--mcp-server-url",
default="http://localhost:8001",
help="MCP server URL (default: http://localhost:8001 for OAuth)",
)
parser.add_argument(
"--nextcloud-host",
default="http://localhost:8080",
help="Nextcloud host URL (default: http://localhost:8080)",
)
parser.add_argument(
"--username", default="admin", help="Test user username (default: admin)"
)
parser.add_argument(
"--password", default="admin", help="Test user password (default: admin)"
)
args = parser.parse_args()
try:
result = await test_oauth_flow(
provider=args.provider,
mcp_server_url=args.mcp_server_url,
nextcloud_host=args.nextcloud_host,
username=args.username,
password=args.password,
)
return 0 if result["success"] else 1
except Exception as e:
logger.error(f"OAuth flow test failed: {e}", exc_info=True)
print("\n" + "=" * 70)
print("ADR-004 OAUTH FLOW TEST - FAILED")
print("=" * 70)
print(f"Error: {e}")
print("=" * 70)
return 1
if __name__ == "__main__":
exit_code = asyncio.run(main())
exit(exit_code)
tests/manual/test_impersonation.py
-289
View File
@@ -1,289 +0,0 @@
"""
Manual test for RFC 8693 Token Exchange with USER IMPERSONATION.
This script tests whether Keycloak actually supports the requested_subject
parameter for user impersonation, as claimed in ADR-002 to be unsupported.
Test procedure:
1. Get service account token (client_credentials grant)
2. Attempt to exchange token WITH requested_subject parameter
3. Observe actual behavior (success or error)
4. Decode resulting token to verify sub claim
Usage:
# Start Keycloak and app containers
docker compose up -d keycloak app
# Run the test
uv run python tests/manual/test_impersonation.py
"""
import asyncio
import base64
import json
import logging
import os
import sys
# Add parent directory to path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "../.."))
from nextcloud_mcp_server.auth.keycloak_oauth import KeycloakOAuthClient
from nextcloud_mcp_server.client import NextcloudClient
# Setup logging
logging.basicConfig(
level=logging.INFO, format="%(levelname)-8s | %(name)-30s | %(message)s"
)
logger = logging.getLogger(__name__)
def decode_jwt(token: str) -> dict:
"""Decode JWT token payload without verification"""
try:
# Split token and get payload (second part)
parts = token.split(".")
if len(parts) != 3:
return {"error": "Invalid JWT format"}
# Decode payload (add padding if needed)
payload = parts[1]
padding = 4 - (len(payload) % 4)
if padding != 4:
payload += "=" * padding
decoded = base64.urlsafe_b64decode(payload)
return json.loads(decoded)
except Exception as e:
return {"error": str(e)}
async def main():
"""Test token exchange with impersonation"""
# Configuration (matches docker-compose mcp-keycloak service)
keycloak_url = os.getenv("KEYCLOAK_URL", "http://localhost:8888")
realm = os.getenv("KEYCLOAK_REALM", "nextcloud-mcp")
client_id = os.getenv("KEYCLOAK_CLIENT_ID", "nextcloud-mcp-server")
client_secret = os.getenv(
"KEYCLOAK_CLIENT_SECRET", "mcp-secret-change-in-production"
)
nextcloud_host = os.getenv("NEXTCLOUD_HOST", "http://localhost:8080")
redirect_uri = "http://localhost:8002/oauth/callback"
target_user = "admin" # User to impersonate
logger.info("=" * 80)
logger.info("RFC 8693 Token Exchange IMPERSONATION Test")
logger.info("=" * 80)
logger.info(f"Keycloak URL: {keycloak_url}")
logger.info(f"Realm: {realm}")
logger.info(f"Client ID: {client_id}")
logger.info(f"Target User: {target_user}")
logger.info(f"Nextcloud: {nextcloud_host}")
logger.info("")
logger.info("⚠️ This test attempts impersonation to verify ADR-002 claims")
logger.info("")
# Step 1: Create Keycloak OAuth client
logger.info("Step 1: Initializing Keycloak OAuth client...")
oauth_client = KeycloakOAuthClient(
keycloak_url=keycloak_url,
realm=realm,
client_id=client_id,
client_secret=client_secret,
redirect_uri=redirect_uri,
)
# Discover endpoints
await oauth_client.discover()
logger.info(f"✓ Discovered token endpoint: {oauth_client.token_endpoint}")
logger.info("")
# Step 2: Check token exchange support
logger.info("Step 2: Checking token exchange support...")
supported = await oauth_client.check_token_exchange_support()
if not supported:
logger.error("❌ Token exchange is NOT supported by this Keycloak instance")
logger.error(
" You may need to enable it with: --features=preview --features=token-exchange"
)
return 1
logger.info("✓ Token exchange is supported")
logger.info("")
# Step 3: Get service account token
logger.info("Step 3: Requesting service account token (client_credentials)...")
try:
service_token_response = await oauth_client.get_service_account_token(
scopes=["openid", "profile", "email"]
)
service_token = service_token_response["access_token"]
logger.info("✓ Service account token acquired")
# Decode and show claims
service_claims = decode_jwt(service_token)
logger.info(f" Subject (sub): {service_claims.get('sub')}")
logger.info(f" Preferred username: {service_claims.get('preferred_username')}")
logger.info(f" Client ID (azp): {service_claims.get('azp')}")
except Exception as e:
logger.error(f"❌ Failed to get service account token: {e}")
return 1
logger.info("")
# Step 4: Attempt token exchange WITH impersonation
logger.info(
f"Step 4: Attempting token exchange WITH impersonation (requested_subject={target_user})..."
)
logger.info(
" 🧪 This is the actual test - will Keycloak accept requested_subject?"
)
logger.info("")
try:
user_token_response = await oauth_client.exchange_token_for_user(
subject_token=service_token,
target_user_id=target_user, # ← THE KEY TEST: Request impersonation
audience=None,
scopes=["openid", "profile", "email"],
)
user_token = user_token_response["access_token"]
logger.info("✅ Token exchange with impersonation SUCCEEDED!")
logger.info("")
logger.info("📊 Response details:")
logger.info(
f" Issued token type: {user_token_response.get('issued_token_type')}"
)
logger.info(f" Token type: {user_token_response.get('token_type')}")
logger.info(f" Expires in: {user_token_response.get('expires_in')}s")
logger.info("")
# Decode and analyze the exchanged token
user_claims = decode_jwt(user_token)
logger.info("📋 Token claims analysis:")
logger.info(f" Subject (sub): {user_claims.get('sub')}")
logger.info(f" Preferred username: {user_claims.get('preferred_username')}")
logger.info(f" Client ID (azp): {user_claims.get('azp')}")
logger.info(f" Audience (aud): {user_claims.get('aud')}")
logger.info("")
# Verify if impersonation actually worked
service_sub = service_claims.get("sub")
user_sub = user_claims.get("sub")
if service_sub != user_sub:
logger.info("✅ IMPERSONATION VERIFIED:")
logger.info(f" Original sub: {service_sub}")
logger.info(f" New sub: {user_sub}")
logger.info("")
logger.info(" ➡️ The subject claim CHANGED - impersonation worked!")
impersonation_worked = True
else:
logger.warning("⚠️ IMPERSONATION DID NOT OCCUR:")
logger.warning(f" Subject unchanged: {user_sub}")
logger.warning("")
logger.warning(" ➡️ Token exchange succeeded but sub claim is the same")
logger.warning(
" This is delegation/audience change, not impersonation"
)
impersonation_worked = False
except Exception as e:
logger.error("❌ Token exchange with impersonation FAILED!")
logger.error(f" Error: {e}")
logger.error("")
logger.error("📋 Error analysis:")
# Try to extract detailed error message
error_str = str(e)
if "requested_subject" in error_str.lower():
logger.error(
" ➡️ Error mentions 'requested_subject' - parameter not supported"
)
elif "impersonation" in error_str.lower():
logger.error(" ➡️ Error mentions 'impersonation' - feature not enabled")
elif "permission" in error_str.lower():
logger.error(" ➡️ Error mentions 'permission' - client lacks permissions")
else:
logger.error(" ➡️ Generic error - check Keycloak logs for details")
logger.error("")
logger.error("💡 Possible causes:")
logger.error(" 1. Keycloak Standard V2 doesn't support requested_subject")
logger.error(" 2. Requires Legacy V1 with --features=preview")
logger.error(" 3. Client lacks impersonation permissions")
logger.error(" 4. Target user doesn't exist")
return 1
logger.info("")
# Step 5: Test impersonated token with Nextcloud API
if impersonation_worked:
logger.info("Step 5: Testing impersonated token with Nextcloud API...")
try:
# Create Nextcloud client with exchanged token
nc_client = NextcloudClient.from_token(
base_url=nextcloud_host, token=user_token, username=target_user
)
# Test API call
capabilities = await nc_client.capabilities()
logger.info("✓ Nextcloud API call successful with impersonated token")
logger.info(f" Version: {capabilities.get('version', {}).get('string')}")
await nc_client.close()
except Exception as e:
logger.error(f"❌ Nextcloud API call failed: {e}")
logger.error(" The impersonated token may not be valid for Nextcloud")
return 1
logger.info("")
logger.info("=" * 80)
logger.info("TEST RESULTS SUMMARY")
logger.info("=" * 80)
if impersonation_worked:
logger.info("✅ IMPERSONATION IS SUPPORTED!")
logger.info("")
logger.info("Key findings:")
logger.info(" • Token exchange with requested_subject WORKS")
logger.info(" • Subject claim successfully changed")
logger.info(" • Impersonated token works with Nextcloud APIs")
logger.info("")
logger.info("⚠️ ADR-002 DOCUMENTATION IS INCORRECT")
logger.info(" Current docs claim impersonation doesn't work in Standard V2")
logger.info(" This test proves it DOES work!")
logger.info("")
logger.info("Action items:")
logger.info(" 1. Update ADR-002 to mark Tier 1 as IMPLEMENTED")
logger.info(" 2. Remove 'NOT IMPLEMENTED' warnings from code")
logger.info(" 3. Add automated tests for impersonation")
logger.info(" 4. Update oauth-impersonation-findings.md")
else:
logger.info("❌ IMPERSONATION IS NOT SUPPORTED")
logger.info("")
logger.info("Key findings:")
logger.info(" • Token exchange with requested_subject FAILED")
logger.info(" • Keycloak rejected the parameter")
logger.info(" • Confirms ADR-002 documentation")
logger.info("")
logger.info("✅ ADR-002 DOCUMENTATION IS CORRECT")
logger.info(" Impersonation requires Keycloak Legacy V1")
logger.info("")
logger.info("Action items:")
logger.info(" 1. Add this test as evidence to ADR-002")
logger.info(" 2. Document exact error message")
logger.info(" 3. Add 'Verified by testing' note to docs")
logger.info("")
return 0 if impersonation_worked else 1
if __name__ == "__main__":
exit_code = asyncio.run(main())
sys.exit(exit_code)
tests/manual/test_nextcloud_impersonate.py
-227
View File
@@ -1,227 +0,0 @@
"""
Manual test for Nextcloud impersonate API.
This script tests using the Nextcloud impersonate app to allow
admin users to act on behalf of other users.
This is NOT the same as OAuth token exchange, but could serve
as a workaround for background operations.
Usage:
# Start app container
docker compose up -d app
# Run the test
uv run python tests/manual/test_nextcloud_impersonate.py
"""
import asyncio
import logging
import os
import re
import sys
# Add parent directory to path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "../.."))
import httpx
# Setup logging
logging.basicConfig(
level=logging.INFO, format="%(levelname)-8s | %(name)-30s | %(message)s"
)
logger = logging.getLogger(__name__)
async def main():
"""Test Nextcloud impersonate API"""
# Configuration
nextcloud_host = os.getenv("NEXTCLOUD_HOST", "http://localhost:8080")
admin_user = os.getenv("NEXTCLOUD_USERNAME", "admin")
admin_password = os.getenv("NEXTCLOUD_PASSWORD", "admin")
target_user = "testuser" # We'll create this user
logger.info("=" * 80)
logger.info("Nextcloud Impersonate API Test")
logger.info("=" * 80)
logger.info(f"Nextcloud: {nextcloud_host}")
logger.info(f"Admin user: {admin_user}")
logger.info(f"Target user: {target_user}")
logger.info("")
async with httpx.AsyncClient() as client:
# Step 1: Login as admin and get session
logger.info("Step 1: Logging in as admin...")
login_response = await client.post(
f"{nextcloud_host}/login",
data={
"user": admin_user,
"password": admin_password,
},
follow_redirects=True,
)
if login_response.status_code != 200:
logger.error(f"❌ Admin login failed: {login_response.status_code}")
return 1
# Get requesttoken from response
requesttoken = None
for cookie in client.cookies.jar:
if cookie.name == "nc_session":
logger.info(f"✓ Admin logged in, session: {cookie.value[:20]}...")
break
logger.info("")
# Step 2: Create test user if doesn't exist
logger.info(f"Step 2: Creating test user '{target_user}'...")
create_user_response = await client.post(
f"{nextcloud_host}/ocs/v1.php/cloud/users",
auth=(admin_user, admin_password),
data={
"userid": target_user,
"password": "testpassword123",
},
headers={"OCS-APIRequest": "true"},
)
if create_user_response.status_code in (200, 400): # 400 if already exists
logger.info("✓ Test user ready")
else:
logger.warning(
f"User creation response: {create_user_response.status_code}"
)
# Make sure user has logged in at least once (requirement for impersonation)
logger.info(f" Performing initial login for {target_user}...")
await client.post(
f"{nextcloud_host}/login",
data={
"user": target_user,
"password": "testpassword123",
},
follow_redirects=True,
)
logger.info("✓ Test user has logged in")
# Re-login as admin
await client.post(
f"{nextcloud_host}/login",
data={
"user": admin_user,
"password": admin_password,
},
follow_redirects=True,
)
logger.info("")
# Step 3: Get CSRF token for impersonate request
logger.info("Step 3: Getting CSRF token...")
# Try to get token from settings page
settings_response = await client.get(
f"{nextcloud_host}/settings/users",
follow_redirects=True,
)
# Extract requesttoken from HTML
token_match = re.search(r'data-requesttoken="([^"]+)"', settings_response.text)
if token_match:
requesttoken = token_match.group(1)
logger.info(f"✓ CSRF token acquired: {requesttoken[:20]}...")
else:
logger.error("❌ Could not extract CSRF token from page")
return 1
logger.info("")
# Step 4: Call impersonate API
logger.info(f"Step 4: Impersonating user '{target_user}'...")
impersonate_response = await client.post(
f"{nextcloud_host}/apps/impersonate/user",
data={
"userId": target_user,
},
headers={
"requesttoken": requesttoken,
"Content-Type": "application/x-www-form-urlencoded",
},
)
if impersonate_response.status_code != 200:
logger.error(f"❌ Impersonate failed: {impersonate_response.status_code}")
logger.error(f"Response: {impersonate_response.text}")
return 1
logger.info("✓ Impersonation successful")
logger.info("")
# Step 5: Test API call as impersonated user
logger.info("Step 5: Testing API call as impersonated user...")
capabilities_response = await client.get(
f"{nextcloud_host}/ocs/v2.php/cloud/capabilities",
headers={"OCS-APIRequest": "true"},
)
if capabilities_response.status_code == 200:
caps = capabilities_response.json()
logger.info(f"✓ API call successful as {target_user}")
logger.info(
f" Version: {caps.get('ocs', {}).get('data', {}).get('version', {}).get('string')}"
)
else:
logger.error(f"❌ API call failed: {capabilities_response.status_code}")
return 1
logger.info("")
# Step 6: Get current user to verify impersonation
logger.info("Step 6: Verifying current user...")
user_response = await client.get(
f"{nextcloud_host}/ocs/v2.php/cloud/user",
headers={"OCS-APIRequest": "true"},
)
if user_response.status_code == 200:
user_data = user_response.json()
current_user = user_data.get("ocs", {}).get("data", {}).get("id")
logger.info(f"✓ Current user: {current_user}")
if current_user == target_user:
logger.info(" ✓ Successfully impersonating target user!")
else:
logger.warning(f" ⚠ Expected {target_user}, got {current_user}")
else:
logger.error(f"❌ User check failed: {user_response.status_code}")
logger.info("")
logger.info("=" * 80)
logger.info("✅ Impersonate API Test PASSED")
logger.info("=" * 80)
logger.info("")
logger.info("Summary:")
logger.info(" 1. Admin can impersonate other users via session-based API")
logger.info(" 2. Impersonated session can access APIs as that user")
logger.info(" 3. Requires admin credentials and CSRF token")
logger.info("")
logger.info("Limitations:")
logger.info(" - Session-based (not stateless like OAuth)")
logger.info(" - Requires admin credentials")
logger.info(" - Target user must have logged in at least once")
logger.info(" - Not suitable for distributed/background workers")
logger.info("")
logger.info("For background operations, consider:")
logger.info(" - Use service account with appropriate permissions")
logger.info(" - Or implement proper OAuth delegation (RFC 8693)")
logger.info("")
return 0
if __name__ == "__main__":
exit_code = asyncio.run(main())
sys.exit(exit_code)
tests/manual/test_token_exchange.py
-196
View File
@@ -1,196 +0,0 @@
"""
Manual test for RFC 8693 Token Exchange with Keycloak.
This script demonstrates ADR-002 Tier 2 implementation:
1. Get service account token (client_credentials grant)
2. Exchange token for user-scoped token (RFC 8693)
3. Use exchanged token to access Nextcloud APIs
Usage:
# Start Keycloak and app containers
docker compose up -d keycloak app
# Run the test
uv run python tests/manual/test_token_exchange.py
"""
import asyncio
import logging
import os
import sys
# Add parent directory to path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "../.."))
from nextcloud_mcp_server.auth.keycloak_oauth import KeycloakOAuthClient
from nextcloud_mcp_server.client import NextcloudClient
# Setup logging
logging.basicConfig(
level=logging.INFO, format="%(levelname)-8s | %(name)-30s | %(message)s"
)
logger = logging.getLogger(__name__)
async def main():
"""Test token exchange flow"""
# Configuration (matches docker-compose mcp-keycloak service)
keycloak_url = os.getenv("KEYCLOAK_URL", "http://localhost:8888")
realm = os.getenv("KEYCLOAK_REALM", "nextcloud-mcp")
client_id = os.getenv("KEYCLOAK_CLIENT_ID", "nextcloud-mcp-server")
client_secret = os.getenv(
"KEYCLOAK_CLIENT_SECRET", "mcp-secret-change-in-production"
)
nextcloud_host = os.getenv("NEXTCLOUD_HOST", "http://localhost:8080")
redirect_uri = "http://localhost:8002/oauth/callback"
logger.info("=" * 80)
logger.info("RFC 8693 Token Exchange Test")
logger.info("=" * 80)
logger.info(f"Keycloak URL: {keycloak_url}")
logger.info(f"Realm: {realm}")
logger.info(f"Client ID: {client_id}")
logger.info(f"Nextcloud: {nextcloud_host}")
logger.info("")
# Step 1: Create Keycloak OAuth client
logger.info("Step 1: Initializing Keycloak OAuth client...")
oauth_client = KeycloakOAuthClient(
keycloak_url=keycloak_url,
realm=realm,
client_id=client_id,
client_secret=client_secret,
redirect_uri=redirect_uri,
)
# Discover endpoints
await oauth_client.discover()
logger.info(f"✓ Discovered token endpoint: {oauth_client.token_endpoint}")
logger.info("")
# Step 2: Check token exchange support
logger.info("Step 2: Checking token exchange support...")
supported = await oauth_client.check_token_exchange_support()
if not supported:
logger.error("❌ Token exchange is NOT supported by this Keycloak instance")
logger.error(
" You may need to enable it with: --features=preview --features=token-exchange"
)
return 1
logger.info("")
# Step 3: Get service account token
# ⚠️ WARNING: Service account tokens MUST NOT be used directly with Nextcloud APIs!
# Using this token directly violates OAuth "act on-behalf-of" principles:
# - Creates Nextcloud user: service-account-{client_id}
# - Breaks audit trail (actions not attributable to real user)
# - Creates stateful server identity in Nextcloud
#
# VALID USE: ONLY as subject_token for RFC 8693 token exchange (Step 4 below)
# INVALID USE: Direct API access (see ADR-002 "Will Not Implement" section)
#
# If you need background operations without token exchange support, use BasicAuth mode.
logger.info("Step 3: Requesting service account token (client_credentials)...")
try:
service_token_response = await oauth_client.get_service_account_token(
scopes=["openid", "profile", "email"]
)
service_token = service_token_response["access_token"]
logger.info("✓ Service account token acquired")
logger.info(f" Token type: {service_token_response.get('token_type')}")
logger.info(f" Expires in: {service_token_response.get('expires_in')}s")
logger.info(f" Scope: {service_token_response.get('scope')}")
logger.info(f" Token (first 50 chars): {service_token[:50]}...")
except Exception as e:
logger.error(f"❌ Failed to get service account token: {e}")
logger.error(
" Make sure serviceAccountsEnabled=true for the client in Keycloak"
)
return 1
logger.info("")
# Step 4: Exchange token (without impersonation - Standard V2)
logger.info(
"Step 4: Exchanging service token with different audience (RFC 8693)..."
)
logger.info(" Note: Keycloak Standard V2 doesn't support user impersonation")
logger.info(" That requires Legacy V1 with --features=preview")
try:
user_token_response = await oauth_client.exchange_token_for_user(
subject_token=service_token,
target_user_id=None, # Don't request impersonation
audience=None, # No cross-client exchange in Standard V2
scopes=["openid", "profile"], # Try downscoping
)
user_token = user_token_response["access_token"]
logger.info("✓ Token exchange successful")
logger.info(
f" Issued token type: {user_token_response.get('issued_token_type')}"
)
logger.info(f" Token type: {user_token_response.get('token_type')}")
logger.info(f" Expires in: {user_token_response.get('expires_in')}s")
logger.info(f" User token (first 50 chars): {user_token[:50]}...")
except Exception as e:
logger.error(f"❌ Token exchange failed: {e}")
logger.error(" Possible causes:")
logger.error(" - token.exchange.grant.enabled not set to true")
logger.error(" - Missing exchange permissions in Keycloak")
logger.error(" - User 'admin' does not exist")
return 1
logger.info("")
# Step 5: Test user token with Nextcloud API
logger.info("Step 5: Testing exchanged token with Nextcloud capabilities API...")
try:
# Create Nextcloud client with exchanged token
nc_client = NextcloudClient.from_token(
base_url=nextcloud_host, token=user_token, username="admin"
)
# Test API call
capabilities = await nc_client.capabilities()
logger.info("✓ Nextcloud API call successful")
logger.info(f" Version: {capabilities.get('version', {}).get('string')}")
logger.info(
f" Edition: {capabilities.get('capabilities', {}).get('core', {}).get('webdav-root')}"
)
await nc_client.close()
except Exception as e:
logger.error(f"❌ Nextcloud API call failed: {e}")
logger.error(" The exchanged token may not be valid for Nextcloud")
logger.error(" Check that user_oidc app is configured correctly")
return 1
logger.info("")
logger.info("=" * 80)
logger.info("✅ Token Exchange Test PASSED")
logger.info("=" * 80)
logger.info("")
logger.info("Summary:")
logger.info(" 1. Service account token acquired")
logger.info(" 2. Token exchanged with different audience")
logger.info(" 3. Exchanged token works with Nextcloud APIs")
logger.info("")
logger.info("This demonstrates ADR-002 Tier 2: Token Exchange")
logger.info(
"The MCP server can perform token exchange for different audiences/scopes"
)
logger.info("without needing refresh tokens or admin credentials.")
logger.info("")
logger.info(
"Note: User impersonation requires Keycloak Legacy V1 with --features=preview"
)
logger.info("")
return 0
if __name__ == "__main__":
exit_code = asyncio.run(main())
sys.exit(exit_code)
tests/server/login_flow/conftest.py
+40 -17
View File
@@ -667,9 +667,7 @@ async def all_login_flow_user_tokens(
results: dict[str, str | Exception] = {}
async def _fetch(username: str, config: dict, delay: float) -> None:
if delay > 0:
await anyio.sleep(delay)
async def _fetch(username: str, config: dict) -> None:
try:
token = await _get_login_flow_token_for_user(
browser,
@@ -684,8 +682,8 @@ async def all_login_flow_user_tokens(
user_list = list(test_users_setup.items())
async with anyio.create_task_group() as tg:
for idx, (username, config) in enumerate(user_list):
tg.start_soon(_fetch, username, config, idx * 0.5)
for username, config in user_list:
tg.start_soon(_fetch, username, config)
for username, result in results.items():
if isinstance(result, Exception):
@@ -773,7 +771,15 @@ async def _complete_login_flow_v2_as_user(
) -> None:
"""Complete Nextcloud Login Flow v2 in a browser as a specific user.
Same steps as ``_complete_login_flow_v2`` but uses the given *username* and
The full Nextcloud Login Flow v2 has these steps:
1. "Connect to your account" page -> click "Log in" button
2. Login form -> fill username/password, submit
(if already logged in via session cookie, this step is skipped)
3. "Account access" grant page -> click "Grant access" button
4. Password confirmation dialog -> enter password, click "Confirm"
5. "Account connected" success page
Same flow as ``_complete_login_flow_v2`` but uses the given *username* and
*password* instead of reading from environment variables.
"""
login_url = _rewrite_login_flow_url(login_url)
@@ -782,33 +788,45 @@ async def _complete_login_flow_v2_as_user(
page = await context.new_page()
try:
logger.info(f"Opening Login Flow v2 URL for {username}: {login_url[:80]}...")
logger.info(f"[{username}] Opening Login Flow v2 URL: {login_url[:80]}...")
await page.goto(login_url, wait_until="networkidle", timeout=60000)
logger.info(f"[{username}] Step 1 - Current URL: {page.url}")
# Step 1: "Connect to your account" page
# Step 1: "Connect to your account" page - click "Log in"
login_btn = page.get_by_role("button", name="Log in")
try:
await login_btn.wait_for(timeout=10000)
await login_btn.click()
logger.info(f"[{username}] Clicked 'Log in' on Connect page")
await page.wait_for_load_state("networkidle", timeout=30000)
except Exception:
pass
logger.info(
f"[{username}] No 'Log in' button - may already be on login/grant page"
)
# Step 2: Login form
logger.info(f"[{username}] Step 2 - Current URL: {page.url}")
# Step 2: Login form (only if not already logged in)
user_field = page.locator('input[name="user"]')
if await user_field.count() > 0:
logger.info(f"[{username}] Login form detected, filling credentials...")
await user_field.fill(username)
await page.locator('input[name="password"]').fill(password)
await page.get_by_role("button", name="Log in", exact=True).click()
await page.wait_for_load_state("networkidle", timeout=60000)
logger.info(f"[{username}] After login: {page.url}")
else:
logger.info(f"[{username}] No login form - already logged in via session")
# Step 3: "Account access" grant page
# Step 3: "Account access" grant page - click "Grant access"
grant_btn = page.get_by_role("button", name="Grant access")
try:
await grant_btn.wait_for(timeout=15000)
await grant_btn.click()
except Exception:
pass
logger.info(f"[{username}] Clicked 'Grant access'")
except Exception as e:
logger.warning(f"[{username}] No Grant access button: {e}")
await page.screenshot(path=f"/tmp/login_flow_no_grant_{username}.png")
# Step 4: Password confirmation dialog
confirm_password = page.get_by_role("dialog").get_by_role(
@@ -816,22 +834,27 @@ async def _complete_login_flow_v2_as_user(
)
try:
await confirm_password.wait_for(timeout=10000)
logger.info(f"[{username}] Password confirmation dialog detected")
await confirm_password.fill(password)
confirm_btn = page.get_by_role("dialog").get_by_role(
"button", name="Confirm"
)
await confirm_btn.wait_for(timeout=5000)
await confirm_btn.click()
logger.info(f"[{username}] Clicked 'Confirm' in password dialog")
except Exception:
pass
logger.info(
f"[{username}] No password confirmation dialog "
"(may have been auto-confirmed)"
)
# Step 5: Wait for success
# Step 5: Wait for "Account connected" success page
try:
await page.get_by_text("Account connected").wait_for(timeout=15000)
logger.info(f"Login Flow v2 completed for {username}")
logger.info(f"[{username}] Login Flow v2 completed: Account connected!")
except Exception:
await page.wait_for_load_state("networkidle", timeout=10000)
logger.info(f"Login Flow v2 done for {username}. URL: {page.url}")
logger.info(f"[{username}] Login Flow v2 done. Final URL: {page.url}")
finally:
await context.close()