From c6316dbb911ad4a20bd4ae48ef286df03331e428 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Wed, 1 Apr 2026 17:43:44 +0200 Subject: [PATCH] =?UTF-8?q?fix:=20address=20PR=20review=20=E2=80=94=20remo?= =?UTF-8?q?ve=20token=20exchange=20tests,=20improve=20logging?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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) --- .../auth/test_token_exchange_legacy_v1.py | 308 -------------- .../auth/test_token_exchange_standard_v2.py | 222 ---------- tests/integration/conftest.py | 4 - .../test_keycloak_token_exchange.py | 380 ------------------ tests/manual/README.md | 47 --- tests/manual/TESTING_INSTRUCTIONS.md | 203 ---------- tests/manual/configure_impersonation.py | 195 --------- tests/manual/test_adr004_manual.py | 319 --------------- tests/manual/test_adr004_oauth_flow.py | 375 ----------------- tests/manual/test_impersonation.py | 289 ------------- tests/manual/test_nextcloud_impersonate.py | 227 ----------- tests/manual/test_token_exchange.py | 196 --------- tests/server/login_flow/conftest.py | 57 ++- 13 files changed, 40 insertions(+), 2782 deletions(-) delete mode 100644 tests/integration/auth/test_token_exchange_legacy_v1.py delete mode 100644 tests/integration/auth/test_token_exchange_standard_v2.py delete mode 100644 tests/integration/test_keycloak_token_exchange.py delete mode 100644 tests/manual/README.md delete mode 100644 tests/manual/TESTING_INSTRUCTIONS.md delete mode 100644 tests/manual/configure_impersonation.py delete mode 100644 tests/manual/test_adr004_manual.py delete mode 100644 tests/manual/test_adr004_oauth_flow.py delete mode 100644 tests/manual/test_impersonation.py delete mode 100644 tests/manual/test_nextcloud_impersonate.py delete mode 100644 tests/manual/test_token_exchange.py diff --git a/tests/integration/auth/test_token_exchange_legacy_v1.py b/tests/integration/auth/test_token_exchange_legacy_v1.py deleted file mode 100644 index 73d54837..00000000 --- a/tests/integration/auth/test_token_exchange_legacy_v1.py +++ /dev/null @@ -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)" - ) diff --git a/tests/integration/auth/test_token_exchange_standard_v2.py b/tests/integration/auth/test_token_exchange_standard_v2.py deleted file mode 100644 index 874491e4..00000000 --- a/tests/integration/auth/test_token_exchange_standard_v2.py +++ /dev/null @@ -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}" - ) diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 20e0ce4a..ab5b7f8f 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -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"] diff --git a/tests/integration/test_keycloak_token_exchange.py b/tests/integration/test_keycloak_token_exchange.py deleted file mode 100644 index 622ad8b7..00000000 --- a/tests/integration/test_keycloak_token_exchange.py +++ /dev/null @@ -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() diff --git a/tests/manual/README.md b/tests/manual/README.md deleted file mode 100644 index ba76d8bb..00000000 --- a/tests/manual/README.md +++ /dev/null @@ -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 -``` diff --git a/tests/manual/TESTING_INSTRUCTIONS.md b/tests/manual/TESTING_INSTRUCTIONS.md deleted file mode 100644 index 1fd8ea1b..00000000 --- a/tests/manual/TESTING_INSTRUCTIONS.md +++ /dev/null @@ -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=" \ - -d "code_verifier=" \ - -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. diff --git a/tests/manual/configure_impersonation.py b/tests/manual/configure_impersonation.py deleted file mode 100644 index df9c03f4..00000000 --- a/tests/manual/configure_impersonation.py +++ /dev/null @@ -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) diff --git a/tests/manual/test_adr004_manual.py b/tests/manual/test_adr004_manual.py deleted file mode 100644 index 1f504b19..00000000 --- a/tests/manual/test_adr004_manual.py +++ /dev/null @@ -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 = """ - - Authorization Success - -

✓ Authorization Successful

-

Authorization code received. You can close this window and return to the terminal.

- - {} - - - - """.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) diff --git a/tests/manual/test_adr004_oauth_flow.py b/tests/manual/test_adr004_oauth_flow.py deleted file mode 100644 index 74df9f24..00000000 --- a/tests/manual/test_adr004_oauth_flow.py +++ /dev/null @@ -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 = """ - - Authorization Success - -

✓ Authorization Successful

-

Authorization code received. You can close this window and return to the terminal.

- - {} - - - - - """.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) diff --git a/tests/manual/test_impersonation.py b/tests/manual/test_impersonation.py deleted file mode 100644 index 9437a740..00000000 --- a/tests/manual/test_impersonation.py +++ /dev/null @@ -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) diff --git a/tests/manual/test_nextcloud_impersonate.py b/tests/manual/test_nextcloud_impersonate.py deleted file mode 100644 index 99064b55..00000000 --- a/tests/manual/test_nextcloud_impersonate.py +++ /dev/null @@ -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) diff --git a/tests/manual/test_token_exchange.py b/tests/manual/test_token_exchange.py deleted file mode 100644 index 3c4b5858..00000000 --- a/tests/manual/test_token_exchange.py +++ /dev/null @@ -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) diff --git a/tests/server/login_flow/conftest.py b/tests/server/login_flow/conftest.py index d3abb125..73c65c03 100644 --- a/tests/server/login_flow/conftest.py +++ b/tests/server/login_flow/conftest.py @@ -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()