feat: Implement RFC 8693 token exchange for Keycloak (ADR-002 Tier 2)
Implements OAuth 2.0 Token Exchange (RFC 8693) enabling the MCP server to exchange service account tokens for user-scoped tokens. This provides an alternative to refresh tokens for background operations. **Core Implementation:** - Added `get_service_account_token()` method to KeycloakOAuthClient for client_credentials grant - Added `exchange_token_for_user()` method implementing RFC 8693 token exchange - Fixed Fernet encryption key handling in RefreshTokenStorage (was incorrectly base64 decoding already-encoded keys) - Updated OAuth configuration to support offline_access scope and refresh token storage infrastructure **Keycloak Configuration:** - Enabled `serviceAccountsEnabled` in realm-export.json - Added `token.exchange.grant.enabled` attribute - Added `client.token.exchange.standard.enabled` attribute (required for Keycloak 26.2+ Standard Token Exchange V2) - Fresh Keycloak imports now correctly enable token exchange **Docker Compose:** - Added TOKEN_ENCRYPTION_KEY and ENABLE_OFFLINE_ACCESS environment variables - Created oauth-tokens volume for refresh token storage - Configured both mcp-oauth and mcp-keycloak services **Testing & Documentation:** - Added tests/manual/test_token_exchange.py - Validates complete RFC 8693 flow - Added tests/manual/test_nextcloud_impersonate.py - Documents session-based impersonation limitations - Added docs/oauth-impersonation-findings.md - Comprehensive investigation findings and resolution documentation **Verified Working:** ✅ Service account token acquisition (client_credentials grant) ✅ RFC 8693 token exchange for internal-to-internal tokens ✅ Exchanged tokens validate with Nextcloud APIs ✅ Keycloak 26.4.2 Standard Token Exchange V2 support **Known Limitations:** - User impersonation (requested_subject) requires Keycloak Legacy V1 with preview features - Cross-client token exchange limited to same realm - Refresh token storage infrastructure ready but unused (MCP protocol limitation) Dependencies: aiosqlite>=0.20.0 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
@@ -0,0 +1,227 @@
|
||||
"""
|
||||
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 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
|
||||
import re
|
||||
|
||||
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)
|
||||
Reference in New Issue
Block a user