diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 22d6ca6e..98f03490 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -42,7 +42,6 @@ jobs: mode: - "single-user" - "multi-user-basic" - - "oauth" - "login-flow" include: # Version-specific image pins — Renovate updates these via customManagers in renovate.json @@ -58,7 +57,7 @@ jobs: # Mode-specific properties - mode: single-user profile: single-user - markers: "(smoke and not oauth and not keycloak and not login_flow and not multi_user_basic) or (integration and not oauth and not keycloak and not login_flow and not multi_user_basic)" + markers: "(smoke and not keycloak and not login_flow and not multi_user_basic) or (integration and not keycloak and not login_flow and not multi_user_basic)" wait-port: 8000 mcp-internal-url: "http://mcp:8000" needs-playwright: false @@ -74,14 +73,6 @@ jobs: needs-playwright: true extra-args: "" - - mode: oauth - profile: oauth - markers: "oauth and not keycloak" - wait-port: 8001 - mcp-internal-url: "http://mcp-oauth:8001" - needs-playwright: true - extra-args: "" - - mode: login-flow profile: login-flow markers: "login_flow" @@ -184,7 +175,7 @@ jobs: echo "MCP service is ready on port ${{ matrix.wait-port }}." - name: Verify OIDC configuration - if: matrix.mode == 'oauth' || matrix.mode == 'login-flow' + if: matrix.mode == 'login-flow' run: | echo "=== OIDC Discovery ===" curl -s http://localhost:8080/.well-known/openid-configuration | jq . diff --git a/docker-compose.yml b/docker-compose.yml index 9bc57ae6..1eeb9d86 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -169,56 +169,6 @@ services: profiles: - multi-user-basic - mcp-oauth: - build: . - command: ["--transport", "streamable-http", "--oauth", "--port", "8001", "--oauth-token-type", "jwt"] - restart: always - depends_on: - app: - condition: service_healthy - ports: - - 127.0.0.1:8001:8001 - environment: - # Generic OIDC configuration (integrated mode - Nextcloud OIDC app) - # OIDC_DISCOVERY_URL not set - defaults to NEXTCLOUD_HOST/.well-known/openid-configuration - # OIDC_CLIENT_ID not set - uses Dynamic Client Registration (DCR) - - NEXTCLOUD_HOST=http://app:80 - - NEXTCLOUD_MCP_SERVER_URL=http://localhost:8001 - - NEXTCLOUD_RESOURCE_URI=http://localhost:8080 # ADR-005: Nextcloud resource identifier for audience validation - - NEXTCLOUD_PUBLIC_ISSUER_URL=http://localhost:8080 - - NEXTCLOUD_OIDC_SCOPES=openid profile email notes:read notes:write calendar:read calendar:write contacts:read contacts:write cookbook:read cookbook:write deck:read deck:write tables:read tables:write files:read files:write sharing:read sharing:write todo:read todo:write - - # Refresh token storage (ADR-002 Tier 1) - - ENABLE_BACKGROUND_OPERATIONS=true - - TOKEN_ENCRYPTION_KEY=Qh60VwZQsM7CLtSMunzC0gIGPBT948S6VSawUkODtvU= - - TOKEN_STORAGE_DB=/app/data/tokens.db - - # ADR-005: Multi-audience mode (default - ENABLE_TOKEN_EXCHANGE=false) - # Tokens must contain BOTH MCP and Nextcloud audiences - # No token exchange needed - tokens work for both MCP auth and Nextcloud APIs - - # Semantic search configuration (ADR-007, ADR-021) - - ENABLE_SEMANTIC_SEARCH=true - - VECTOR_SYNC_SCAN_INTERVAL=60 - - VECTOR_SYNC_PROCESSOR_WORKERS=1 - - # Qdrant configuration - persistent local storage - - QDRANT_LOCATION=/app/data/qdrant - - # Embedding provider for vector sync (use Simple provider as fallback) - # Ollama not available in CI/test environments - # - OLLAMA_BASE_URL=http://ollama:11434 - # - OLLAMA_EMBEDDING_MODEL=nomic-embed-text - - # NO admin credentials - using OAuth with Dynamic Client Registration (DCR) - # Client credentials registered via RFC 7591 and stored in volume - # JWT token type is used for testing (faster validation, scopes embedded in token) - volumes: - - oauth-client-storage:/app/.oauth - - oauth-tokens:/app/data - profiles: - - oauth - keycloak: image: quay.io/keycloak/keycloak:26.5.4@sha256:ae8efb0d218d8921334b03a2dbee7069a0b868240691c50a3ffc9f42fabba8b4 command: @@ -381,8 +331,6 @@ services: volumes: nextcloud: db: - oauth-client-storage: - oauth-tokens: keycloak-tokens: keycloak-oauth-storage: login-flow-data: diff --git a/pyproject.toml b/pyproject.toml index 3bc238e2..14f417c1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -72,7 +72,6 @@ log_level = "ERROR" markers = [ "unit: Fast unit tests with mocked dependencies", "integration: Integration tests requiring Docker containers", - "oauth: OAuth tests requiring Playwright (slowest)", "smoke: Critical path smoke tests for quick validation", "keycloak: OAuth tests that utilize keycloak external identity provider", "login_flow: Login Flow v2 integration tests (ADR-022)", diff --git a/tests/client/test_oauth.py b/tests/client/test_oauth.py deleted file mode 100644 index 284f0b2f..00000000 --- a/tests/client/test_oauth.py +++ /dev/null @@ -1,103 +0,0 @@ -"""Integration tests for OAuth authentication.""" - -import logging -import os - -import pytest -from httpx import HTTPStatusError - -from nextcloud_mcp_server.auth import BearerAuth -from nextcloud_mcp_server.client import NextcloudClient - -logger = logging.getLogger(__name__) - -pytestmark = [pytest.mark.integration, pytest.mark.oauth] - - -# OAuth Client Tests - - -async def test_oauth_client_capabilities(nc_oauth_client: NextcloudClient): - """Test that OAuth client can fetch capabilities.""" - capabilities = await nc_oauth_client.capabilities() - - assert capabilities is not None - assert "ocs" in capabilities - logger.info( - f"OAuth client successfully fetched capabilities: {capabilities.get('ocs').get('meta')}" - ) - - -async def test_oauth_client_notes_list(nc_oauth_client: NextcloudClient): - """Test that OAuth client can list notes.""" - notes = [note async for note in nc_oauth_client.notes.get_all_notes()] - - assert isinstance(notes, list) - logger.info(f"OAuth client successfully listed {len(notes)} notes") - - -async def test_oauth_client_create_note(nc_oauth_client: NextcloudClient): - """Test that OAuth client can create and delete a note.""" - # Create note - note_title = "OAuth Test Note" - note_content = "This note was created with OAuth authentication" - - created_note = await nc_oauth_client.notes.create_note( - title=note_title, content=note_content - ) - - assert created_note is not None - assert created_note.get("title") == note_title - note_id = created_note.get("id") - assert note_id is not None - - logger.info(f"OAuth client successfully created note with ID: {note_id}") - - # Clean up - delete the note - try: - await nc_oauth_client.notes.delete_note(note_id=note_id) - logger.info(f"OAuth client successfully deleted note {note_id}") - except Exception as e: - logger.error(f"Failed to clean up test note {note_id}: {e}") - raise - - -# OAuth Token Validation Tests - - -async def test_token_in_request_headers( - nc_oauth_client: NextcloudClient, playwright_oauth_token: str -): - """Verify that bearer token is being used in requests.""" - # The client should be using BearerAuth - assert nc_oauth_client._client.auth is not None - - # Make a request and verify it works - capabilities = await nc_oauth_client.capabilities() - assert capabilities is not None - - logger.info("OAuth bearer token is correctly included in requests") - - -async def test_invalid_token_fails(): - """Test that an invalid token results in authentication failure.""" - nextcloud_host = os.getenv("NEXTCLOUD_HOST") - if not nextcloud_host: - pytest.skip("NEXTCLOUD_HOST not set") - - # Create client with invalid token using BearerAuth - invalid_client = NextcloudClient( - base_url=nextcloud_host, - username="testuser", - auth=BearerAuth("invalid_token_12345"), - ) - - # Attempt to use a protected endpoint - should fail with 401 - # Note: capabilities endpoint is public and doesn't require auth - with pytest.raises(HTTPStatusError) as exc_info: - _ = [note async for note in invalid_client.notes.get_all_notes()] - - assert exc_info.value.response.status_code == 401 - - await invalid_client.close() - logger.info("Invalid OAuth token correctly rejected") diff --git a/tests/client/test_oauth_playwright.py b/tests/client/test_oauth_playwright.py deleted file mode 100644 index 588404c2..00000000 --- a/tests/client/test_oauth_playwright.py +++ /dev/null @@ -1,32 +0,0 @@ -"""Integration tests for Playwright-based OAuth authentication.""" - -import logging - -import pytest - -logger = logging.getLogger(__name__) - -pytestmark = [pytest.mark.integration, pytest.mark.oauth] - - -async def test_playwright_oauth_token_acquisition(playwright_oauth_token: str): - """Test that Playwright can acquire an OAuth token automatically.""" - assert playwright_oauth_token is not None - assert isinstance(playwright_oauth_token, str) - assert len(playwright_oauth_token) > 0 - logger.info( - f"Successfully acquired OAuth token via Playwright: {playwright_oauth_token[:20]}..." - ) - - -async def test_oauth_client_with_playwright_flow(nc_oauth_client): - """Test that OAuth client created via Playwright flow can access Nextcloud APIs.""" - # Test 1: Check capabilities - capabilities = await nc_oauth_client.capabilities() - assert capabilities is not None - logger.info("OAuth client (Playwright) successfully fetched capabilities") - - # Test 2: List notes - notes = [note async for note in nc_oauth_client.notes.get_all_notes()] - assert isinstance(notes, list) - logger.info(f"OAuth client (Playwright) successfully listed {len(notes)} notes") diff --git a/tests/conftest.py b/tests/conftest.py index ccc6bc7f..bdb10bd0 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -2005,33 +2005,37 @@ async def _handle_oauth_consent_screen(page, username: str = "user"): f" ⊗ Scope checkbox {i + 1} disabled (required scope)" ) - # Click the Allow button to grant consent - # Check button exists first - allow_button_locator = page.locator('button:has-text("Allow")') + # Click the Allow button to grant consent with retry logic. + # Uses Playwright's native click (dispatches proper browser events that + # trigger Vue.js handlers) instead of JS btn.click() which can miss them. + allow_button = page.locator('button:has-text("Allow")') - if await allow_button_locator.count() > 0: + if await allow_button.count() > 0: logger.info(f" Clicking Allow button to grant consent for {username}...") - # Use JavaScript click to handle consent buttons that may be outside viewport - # This is more reliable than Playwright's click which requires element visibility - logger.info( - " Using JavaScript click for consent (handles viewport issues)..." - ) - await page.evaluate( - """ - const buttons = document.querySelectorAll('button'); - for (const btn of buttons) { - if (btn.textContent.trim() === 'Allow') { - btn.click(); - break; - } - } - """ - ) + for attempt in range(3): + await allow_button.scroll_into_view_if_needed() + await allow_button.click() + try: + await page.wait_for_url( + lambda url: "/consent" not in url, timeout=10000 + ) + logger.info(f" Consent granted for {username}") + return True + except TimeoutError: + if attempt == 2: + screenshot_path = f"/tmp/consent_click_failed_{username}.png" + await page.screenshot(path=screenshot_path) + logger.error( + f" Consent click failed after 3 attempts for {username}, " + f"screenshot: {screenshot_path}" + ) + raise + logger.warning( + f" Consent click attempt {attempt + 1} didn't navigate, retrying..." + ) - await page.wait_for_load_state("networkidle", timeout=30000) - logger.info(f" Consent granted for {username}") - return True + return True # unreachable but satisfies type checker else: logger.error(f" Allow button not found for {username}") return False @@ -2047,6 +2051,7 @@ async def _get_oauth_token_with_scopes( oauth_callback_server, scopes: str, resource: str | None = None, + mcp_server_base_url: str = "http://localhost:8004", ) -> str: """ Helper function to obtain OAuth token with specific scopes. @@ -2057,6 +2062,7 @@ async def _get_oauth_token_with_scopes( oauth_callback_server: OAuth callback server fixture scopes: Space-separated list of scopes (e.g., "openid profile email notes:read") resource: Optional resource parameter (RFC 8707) for token audience + mcp_server_base_url: Base URL of the MCP server for resource metadata discovery Returns: OAuth access token string with requested scopes @@ -2085,7 +2091,6 @@ async def _get_oauth_token_with_scopes( # If no resource provided, fetch from MCP server metadata if resource is None: - mcp_server_base_url = "http://localhost:8001" try: resource_metadata = await get_mcp_server_resource_metadata( mcp_server_base_url diff --git a/tests/server/auth/test_userinfo_routes.py b/tests/server/auth/test_userinfo_routes.py index 0e5c9cd1..8a7878c9 100644 --- a/tests/server/auth/test_userinfo_routes.py +++ b/tests/server/auth/test_userinfo_routes.py @@ -1,11 +1,6 @@ """Unit tests for user info routes. -Note: Most unit tests were removed as they relied on the old _get_user_info API. -The new browser OAuth session-based implementation is covered by integration tests -in tests/server/oauth/test_userinfo_integration.py which test the full OAuth flow -with real browser sessions, token storage, and IdP interactions. - -These unit tests cover only the simple _query_idp_userinfo helper function. +These unit tests cover the simple _query_idp_userinfo helper function. """ from unittest.mock import AsyncMock, Mock diff --git a/tests/server/login_flow/conftest.py b/tests/server/login_flow/conftest.py index 9253ee6f..d2c47344 100644 --- a/tests/server/login_flow/conftest.py +++ b/tests/server/login_flow/conftest.py @@ -24,6 +24,9 @@ from mcp.types import ElicitRequestParams, ElicitResult from tests.conftest import ( DEFAULT_FULL_SCOPES, + DEFAULT_READ_SCOPES, + DEFAULT_WRITE_SCOPES, + _get_oauth_token_with_scopes, _handle_oauth_consent_screen, create_mcp_client_session, get_mcp_server_resource_metadata, @@ -415,3 +418,129 @@ async def nc_mcp_login_flow_client( ) yield session + + +# --------------------------------------------------------------------------- +# Scope-filtered OAuth client fixtures for scope authorization tests +# These obtain tokens with specific scope subsets via the login-flow server +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="session") +async def login_flow_read_only_token( + anyio_backend, + browser, + login_flow_oauth_client_credentials, + oauth_callback_server, +) -> str: + """OAuth token with read-only scopes for the login-flow MCP server.""" + return await _get_oauth_token_with_scopes( + browser, + login_flow_oauth_client_credentials, + oauth_callback_server, + scopes=DEFAULT_READ_SCOPES, + mcp_server_base_url=LOGIN_FLOW_MCP_BASE_URL, + ) + + +@pytest.fixture(scope="session") +async def login_flow_write_only_token( + anyio_backend, + browser, + login_flow_oauth_client_credentials, + oauth_callback_server, +) -> str: + """OAuth token with write-only scopes for the login-flow MCP server.""" + return await _get_oauth_token_with_scopes( + browser, + login_flow_oauth_client_credentials, + oauth_callback_server, + scopes=DEFAULT_WRITE_SCOPES, + mcp_server_base_url=LOGIN_FLOW_MCP_BASE_URL, + ) + + +@pytest.fixture(scope="session") +async def login_flow_full_access_token( + anyio_backend, + browser, + login_flow_oauth_client_credentials, + oauth_callback_server, +) -> str: + """OAuth token with full access scopes for the login-flow MCP server.""" + return await _get_oauth_token_with_scopes( + browser, + login_flow_oauth_client_credentials, + oauth_callback_server, + scopes=DEFAULT_FULL_SCOPES, + mcp_server_base_url=LOGIN_FLOW_MCP_BASE_URL, + ) + + +@pytest.fixture(scope="session") +async def login_flow_no_custom_scopes_token( + anyio_backend, + browser, + login_flow_oauth_client_credentials, + oauth_callback_server, +) -> str: + """OAuth token with no custom scopes (only OIDC defaults) for the login-flow MCP server.""" + return await _get_oauth_token_with_scopes( + browser, + login_flow_oauth_client_credentials, + oauth_callback_server, + scopes="openid profile email", + mcp_server_base_url=LOGIN_FLOW_MCP_BASE_URL, + ) + + +@pytest.fixture(scope="session") +async def nc_mcp_login_flow_client_read_only( + anyio_backend, login_flow_read_only_token: str +) -> AsyncGenerator[ClientSession, Any]: + """MCP client with read-only scopes on the login-flow server.""" + async for session in create_mcp_client_session( + url=LOGIN_FLOW_MCP_URL, + token=login_flow_read_only_token, + client_name="Login Flow MCP Read-Only", + ): + yield session + + +@pytest.fixture(scope="session") +async def nc_mcp_login_flow_client_write_only( + anyio_backend, login_flow_write_only_token: str +) -> AsyncGenerator[ClientSession, Any]: + """MCP client with write-only scopes on the login-flow server.""" + async for session in create_mcp_client_session( + url=LOGIN_FLOW_MCP_URL, + token=login_flow_write_only_token, + client_name="Login Flow MCP Write-Only", + ): + yield session + + +@pytest.fixture(scope="session") +async def nc_mcp_login_flow_client_full_access( + anyio_backend, login_flow_full_access_token: str +) -> AsyncGenerator[ClientSession, Any]: + """MCP client with full access scopes on the login-flow server.""" + async for session in create_mcp_client_session( + url=LOGIN_FLOW_MCP_URL, + token=login_flow_full_access_token, + client_name="Login Flow MCP Full Access", + ): + yield session + + +@pytest.fixture(scope="session") +async def nc_mcp_login_flow_client_no_custom_scopes( + anyio_backend, login_flow_no_custom_scopes_token: str +) -> AsyncGenerator[ClientSession, Any]: + """MCP client with no custom scopes on the login-flow server.""" + async for session in create_mcp_client_session( + url=LOGIN_FLOW_MCP_URL, + token=login_flow_no_custom_scopes_token, + client_name="Login Flow MCP No Custom Scopes", + ): + yield session diff --git a/tests/server/oauth/test_dcr_deletion_methods.py b/tests/server/login_flow/test_dcr_deletion_methods.py similarity index 99% rename from tests/server/oauth/test_dcr_deletion_methods.py rename to tests/server/login_flow/test_dcr_deletion_methods.py index c2214ff7..0f305f0d 100644 --- a/tests/server/oauth/test_dcr_deletion_methods.py +++ b/tests/server/login_flow/test_dcr_deletion_methods.py @@ -16,7 +16,7 @@ from nextcloud_mcp_server.auth.client_registration import register_client logger = logging.getLogger(__name__) -pytestmark = [pytest.mark.integration, pytest.mark.oauth] +pytestmark = [pytest.mark.integration, pytest.mark.login_flow] @pytest.mark.integration diff --git a/tests/server/oauth/test_dcr_lifecycle.py b/tests/server/login_flow/test_dcr_lifecycle.py similarity index 99% rename from tests/server/oauth/test_dcr_lifecycle.py rename to tests/server/login_flow/test_dcr_lifecycle.py index f1905cdf..7c22c2dd 100644 --- a/tests/server/oauth/test_dcr_lifecycle.py +++ b/tests/server/login_flow/test_dcr_lifecycle.py @@ -26,7 +26,7 @@ from ...conftest import _handle_oauth_consent_screen logger = logging.getLogger(__name__) -pytestmark = [pytest.mark.integration, pytest.mark.oauth] +pytestmark = [pytest.mark.integration, pytest.mark.login_flow] async def get_oauth_token_with_client( diff --git a/tests/server/oauth/test_dcr_new_implementation.py b/tests/server/login_flow/test_dcr_new_implementation.py similarity index 99% rename from tests/server/oauth/test_dcr_new_implementation.py rename to tests/server/login_flow/test_dcr_new_implementation.py index 3462c758..05c57731 100644 --- a/tests/server/oauth/test_dcr_new_implementation.py +++ b/tests/server/login_flow/test_dcr_new_implementation.py @@ -12,7 +12,7 @@ import pytest logger = logging.getLogger(__name__) -pytestmark = [pytest.mark.integration, pytest.mark.oauth] +pytestmark = [pytest.mark.integration, pytest.mark.login_flow] @pytest.mark.integration diff --git a/tests/server/oauth/test_dcr_token_type.py b/tests/server/login_flow/test_dcr_token_type.py similarity index 99% rename from tests/server/oauth/test_dcr_token_type.py rename to tests/server/login_flow/test_dcr_token_type.py index e5abd26b..60174c85 100644 --- a/tests/server/oauth/test_dcr_token_type.py +++ b/tests/server/login_flow/test_dcr_token_type.py @@ -30,7 +30,7 @@ from ...conftest import _handle_oauth_consent_screen logger = logging.getLogger(__name__) -pytestmark = [pytest.mark.integration, pytest.mark.oauth] +pytestmark = [pytest.mark.integration, pytest.mark.login_flow] def is_jwt_format(token: str) -> bool: diff --git a/tests/server/oauth/test_introspection_authorization.py b/tests/server/login_flow/test_introspection_authorization.py similarity index 99% rename from tests/server/oauth/test_introspection_authorization.py rename to tests/server/login_flow/test_introspection_authorization.py index 2c4315ff..43a6bfe6 100644 --- a/tests/server/oauth/test_introspection_authorization.py +++ b/tests/server/login_flow/test_introspection_authorization.py @@ -27,7 +27,7 @@ from ...conftest import _handle_oauth_consent_screen logger = logging.getLogger(__name__) -pytestmark = [pytest.mark.integration, pytest.mark.oauth] +pytestmark = [pytest.mark.integration, pytest.mark.login_flow] @pytest.fixture(scope="module") diff --git a/tests/server/oauth/test_scope_authorization.py b/tests/server/login_flow/test_scope_authorization.py similarity index 93% rename from tests/server/oauth/test_scope_authorization.py rename to tests/server/login_flow/test_scope_authorization.py index adccc70a..549c9532 100644 --- a/tests/server/oauth/test_scope_authorization.py +++ b/tests/server/login_flow/test_scope_authorization.py @@ -18,22 +18,22 @@ import pytest @pytest.mark.integration -@pytest.mark.oauth +@pytest.mark.login_flow async def test_prm_endpoint(): """Test that the Protected Resource Metadata endpoint returns correct data.""" # Test the PRM endpoint directly (RFC 9728 - path includes /mcp resource) async with httpx.AsyncClient() as client: response = await client.get( - "http://localhost:8001/.well-known/oauth-protected-resource/mcp" + "http://localhost:8004/.well-known/oauth-protected-resource/mcp" ) assert response.status_code == 200 prm_data = response.json() - assert prm_data["resource"] == "http://localhost:8001/mcp" + assert prm_data["resource"] == "http://localhost:8004/mcp" assert "notes:read" in prm_data["scopes_supported"] assert "notes:write" in prm_data["scopes_supported"] - assert "http://localhost:8001" in prm_data["authorization_servers"] + assert "http://localhost:8004" in prm_data["authorization_servers"] assert "header" in prm_data["bearer_methods_supported"] assert "RS256" in prm_data["resource_signing_alg_values_supported"] @@ -61,14 +61,14 @@ async def test_basicauth_shows_all_tools(nc_mcp_client): @pytest.mark.integration -@pytest.mark.oauth -async def test_read_only_token_filters_write_tools(nc_mcp_oauth_client_read_only): +@pytest.mark.login_flow +async def test_read_only_token_filters_write_tools(nc_mcp_login_flow_client_read_only): """Test that a token with only read scopes filters out write tools.""" logger = logging.getLogger(__name__) # Connect with token that has only "notes:read" scope - result = await nc_mcp_oauth_client_read_only.list_tools() + result = await nc_mcp_login_flow_client_read_only.list_tools() assert result is not None assert len(result.tools) > 0 @@ -110,14 +110,14 @@ async def test_read_only_token_filters_write_tools(nc_mcp_oauth_client_read_only @pytest.mark.integration -@pytest.mark.oauth -async def test_write_only_token_filters_read_tools(nc_mcp_oauth_client_write_only): +@pytest.mark.login_flow +async def test_write_only_token_filters_read_tools(nc_mcp_login_flow_client_write_only): """Test that a token with only write scopes filters out read tools.""" logger = logging.getLogger(__name__) # Connect with token that has only "notes:write" scope - result = await nc_mcp_oauth_client_write_only.list_tools() + result = await nc_mcp_login_flow_client_write_only.list_tools() assert result is not None assert len(result.tools) > 0 @@ -159,14 +159,14 @@ async def test_write_only_token_filters_read_tools(nc_mcp_oauth_client_write_onl @pytest.mark.integration -@pytest.mark.oauth -async def test_full_access_token_shows_all_tools(nc_mcp_oauth_client_full_access): +@pytest.mark.login_flow +async def test_full_access_token_shows_all_tools(nc_mcp_login_flow_client_full_access): """Test that a token with both read and write scopes scopes can see all tools.""" logger = logging.getLogger(__name__) # Connect with token that has both "notes:read" and "notes:write" scopes - result = await nc_mcp_oauth_client_full_access.list_tools() + result = await nc_mcp_login_flow_client_full_access.list_tools() assert result is not None assert len(result.tools) > 0 @@ -393,9 +393,9 @@ async def test_scope_metadata_coverage(nc_mcp_client): @pytest.mark.integration -@pytest.mark.oauth +@pytest.mark.login_flow async def test_jwt_with_no_custom_scopes_returns_zero_tools( - nc_mcp_oauth_client_no_custom_scopes, + nc_mcp_login_flow_client_no_custom_scopes, ): """ Test that a JWT token with only OIDC default scopes shows only OAuth provisioning tools. @@ -410,7 +410,7 @@ async def test_jwt_with_no_custom_scopes_returns_zero_tools( logger = logging.getLogger(__name__) # Connect with JWT token that has NO custom scopes (only openid, profile, email) - result = await nc_mcp_oauth_client_no_custom_scopes.list_tools() + result = await nc_mcp_login_flow_client_no_custom_scopes.list_tools() assert result is not None tool_names = [tool.name for tool in result.tools] @@ -438,8 +438,8 @@ async def test_jwt_with_no_custom_scopes_returns_zero_tools( @pytest.mark.integration -@pytest.mark.oauth -async def test_jwt_consent_scenarios_read_only(nc_mcp_oauth_client_read_only): +@pytest.mark.login_flow +async def test_jwt_consent_scenarios_read_only(nc_mcp_login_flow_client_read_only): """ Test JWT with only nc:read scope consented. @@ -449,7 +449,7 @@ async def test_jwt_consent_scenarios_read_only(nc_mcp_oauth_client_read_only): logger = logging.getLogger(__name__) - result = await nc_mcp_oauth_client_read_only.list_tools() + result = await nc_mcp_login_flow_client_read_only.list_tools() assert result is not None assert len(result.tools) > 0 @@ -476,8 +476,8 @@ async def test_jwt_consent_scenarios_read_only(nc_mcp_oauth_client_read_only): @pytest.mark.integration -@pytest.mark.oauth -async def test_jwt_consent_scenarios_write_only(nc_mcp_oauth_client_write_only): +@pytest.mark.login_flow +async def test_jwt_consent_scenarios_write_only(nc_mcp_login_flow_client_write_only): """ Test JWT with only nc:write scope consented. @@ -487,7 +487,7 @@ async def test_jwt_consent_scenarios_write_only(nc_mcp_oauth_client_write_only): logger = logging.getLogger(__name__) - result = await nc_mcp_oauth_client_write_only.list_tools() + result = await nc_mcp_login_flow_client_write_only.list_tools() assert result is not None assert len(result.tools) > 0 @@ -514,8 +514,8 @@ async def test_jwt_consent_scenarios_write_only(nc_mcp_oauth_client_write_only): @pytest.mark.integration -@pytest.mark.oauth -async def test_jwt_consent_scenarios_full_access(nc_mcp_oauth_client_full_access): +@pytest.mark.login_flow +async def test_jwt_consent_scenarios_full_access(nc_mcp_login_flow_client_full_access): """ Test JWT with both nc:read and nc:write scopes consented. @@ -525,7 +525,7 @@ async def test_jwt_consent_scenarios_full_access(nc_mcp_oauth_client_full_access logger = logging.getLogger(__name__) - result = await nc_mcp_oauth_client_full_access.list_tools() + result = await nc_mcp_login_flow_client_full_access.list_tools() assert result is not None assert len(result.tools) > 0 diff --git a/tests/server/oauth/__init__.py b/tests/server/oauth/__init__.py deleted file mode 100644 index 4d3b9d88..00000000 --- a/tests/server/oauth/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""OAuth-specific integration tests.""" diff --git a/tests/server/oauth/test_astrolabe_multi_server_integration.py b/tests/server/oauth/test_astrolabe_multi_server_integration.py deleted file mode 100644 index f393d34d..00000000 --- a/tests/server/oauth/test_astrolabe_multi_server_integration.py +++ /dev/null @@ -1,109 +0,0 @@ -"""Test Astrolabe integration with multiple MCP server deployments. - -Cross-system interface test: Tests the MCP server's integration with the -Astrolabe Nextcloud app, which is installed from the Nextcloud app store via -app-hooks/post-installation/20-install-astrolabe-app.sh. Astrolabe source -lives in a separate repository (https://github.com/cbcoutinho/astrolabe). - -This test suite verifies that the Astrolabe app can be dynamically configured -to connect to different MCP server deployments (mcp-oauth, mcp-keycloak, etc.). - -The configuration is managed dynamically during tests using the -configure_astrolabe_for_mcp_server fixture, which allows testing multiple -deployment scenarios without requiring static post-installation configuration. -""" - -import logging - -import pytest - -logger = logging.getLogger(__name__) - -pytestmark = [pytest.mark.integration, pytest.mark.oauth] - - -class TestAstrolabeMultiServerIntegration: - """Test suite for Astrolabe integration with multiple MCP servers.""" - - @pytest.mark.parametrize( - "mcp_server_config", - [ - { - "name": "mcp-oauth", - "internal_url": "http://mcp-oauth:8001", - "public_url": "http://localhost:8001", - }, - { - "name": "mcp-keycloak", - "internal_url": "http://mcp-keycloak:8002", - "public_url": "http://localhost:8002", - }, - # Add more MCP server configurations as needed: - # { - # "name": "mcp-multi-user-basic", - # "internal_url": "http://mcp-multi-user-basic:8000", - # "public_url": "http://localhost:8003", - # }, - ], - ) - async def test_astrolabe_configuration_for_different_servers( - self, configure_astrolabe_for_mcp_server, mcp_server_config - ): - """Test that Astrolabe can be configured for different MCP servers. - - This test verifies that: - 1. The configure_astrolabe_for_mcp_server fixture successfully configures - the Astrolabe app for different MCP server endpoints - 2. OAuth client credentials are properly generated and stored - 3. The configuration can be dynamically changed between tests - """ - logger.info(f"Configuring Astrolabe for {mcp_server_config['name']}...") - - # Configure Astrolabe for the specific MCP server - credentials = await configure_astrolabe_for_mcp_server( - mcp_server_internal_url=mcp_server_config["internal_url"], - mcp_server_public_url=mcp_server_config["public_url"], - ) - - # Verify credentials were returned - assert "client_id" in credentials - assert "client_secret" in credentials - assert credentials["client_id"] == "nextcloudMcpServerUIPublicClient" - assert len(credentials["client_secret"]) > 0 - - logger.info( - f"✓ Astrolabe successfully configured for {mcp_server_config['name']}" - ) - logger.info(f" Internal URL: {mcp_server_config['internal_url']}") - logger.info(f" Public URL: {mcp_server_config['public_url']}") - logger.info(f" Client ID: {credentials['client_id']}") - logger.info(f" Client Secret: {credentials['client_secret'][:8]}...") - - async def test_astrolabe_reconfiguration(self, configure_astrolabe_for_mcp_server): - """Test that Astrolabe can be reconfigured multiple times in the same session. - - This verifies that the OAuth client can be recreated with different - settings without conflicts. - """ - # First configuration: mcp-oauth - logger.info("First configuration: mcp-oauth") - credentials1 = await configure_astrolabe_for_mcp_server( - mcp_server_internal_url="http://mcp-oauth:8001", - mcp_server_public_url="http://localhost:8001", - ) - - assert credentials1["client_id"] == "nextcloudMcpServerUIPublicClient" - - # Second configuration: mcp-keycloak (reconfiguration) - logger.info("Second configuration: mcp-keycloak (reconfiguration)") - credentials2 = await configure_astrolabe_for_mcp_server( - mcp_server_internal_url="http://mcp-keycloak:8002", - mcp_server_public_url="http://localhost:8002", - ) - - assert credentials2["client_id"] == "nextcloudMcpServerUIPublicClient" - - # Client secrets should be different (new client created) - assert credentials1["client_secret"] != credentials2["client_secret"] - - logger.info("✓ Astrolabe successfully reconfigured without conflicts") diff --git a/tests/server/oauth/test_elicitation_integration.py b/tests/server/oauth/test_elicitation_integration.py deleted file mode 100644 index dfec2a74..00000000 --- a/tests/server/oauth/test_elicitation_integration.py +++ /dev/null @@ -1,206 +0,0 @@ -"""Integration tests for login elicitation with real MCP client callback support. - -These tests verify the complete end-to-end login elicitation flow (ADR-006) -using the python-sdk MCP client with actual elicitation callback implementation. - -Unlike test_login_elicitation.py which validates response formats, these tests -exercise the REAL elicitation protocol: -1. MCP client with elicitation callback connects to server -2. Tool triggers elicitation (ctx.elicit()) -3. Client callback receives elicitation request -4. Callback completes OAuth flow via Playwright automation -5. Client returns acceptance -6. Tool proceeds with authenticated operation - -This validates that: -- python-sdk MCP client can handle elicitation requests -- OAuth flow completion via callback works end-to-end -- Refresh tokens are properly stored after elicitation -- check_logged_in returns "yes" after successful OAuth -""" - -import logging - -import pytest - -logger = logging.getLogger(__name__) - -pytestmark = [pytest.mark.integration, pytest.mark.oauth] - - -async def revoke_refresh_tokens(client): - """Helper to revoke all refresh tokens from MCP server. - - This forces check_logged_in to trigger elicitation by removing - any existing refresh tokens via the revoke_nextcloud_access tool. - """ - logger.info("Revoking refresh tokens via revoke_nextcloud_access tool...") - - result = await client.call_tool("revoke_nextcloud_access", arguments={}) - - logger.info(f"Revoke result: isError={result.isError}") - if not result.isError: - logger.info(f"✓ Revoke response: {result.content[0].text}") - else: - logger.warning(f"Revoke failed: {result.content}") - - -async def test_check_logged_in_with_real_elicitation_callback( - nc_mcp_oauth_client_with_elicitation, -): - """Test check_logged_in with actual elicitation callback that completes OAuth. - - This test validates the COMPLETE elicitation flow: - 1. Call check_logged_in tool (which triggers elicitation) - 2. Elicitation callback extracts OAuth URL - 3. Playwright automation completes OAuth flow - 4. Callback returns acceptance - 5. Tool returns "yes" (logged in) - 6. Refresh token is stored - - This is the ONLY test that exercises the real MCP elicitation protocol - with python-sdk's ClientSession elicitation callback support. - """ - client = nc_mcp_oauth_client_with_elicitation - - logger.info("=" * 80) - logger.info("TEST: Real elicitation callback with OAuth completion") - logger.info("=" * 80) - - # Revoke refresh tokens to force elicitation - await revoke_refresh_tokens(client) - - # Call check_logged_in - this should trigger elicitation - logger.info("Calling check_logged_in tool...") - result = await client.call_tool("check_logged_in", arguments={}) - - logger.info("Tool execution completed") - logger.info(f" Is error: {result.isError}") - if result.content: - response_text = result.content[0].text - logger.info(f" Response: {response_text}") - else: - logger.warning(" No content in response") - - # Validate tool execution succeeded - assert result.isError is False, f"Tool execution failed: {result.content}" - assert result.content is not None, "No content in tool response" - - response_text = result.content[0].text.lower() - - # Validate elicitation was triggered - elicitation_count = client.elicitation_triggered["count"] - logger.info(f"✓ Elicitation triggered {elicitation_count} time(s)") - assert elicitation_count >= 1, ( - "Elicitation callback should have been invoked at least once" - ) - - # Validate OAuth completed successfully and tool returned "yes" - assert "yes" in response_text, ( - f"Expected 'yes' after successful OAuth via elicitation, got: {response_text}" - ) - - logger.info("✅ Test passed: Real elicitation callback completed OAuth flow") - logger.info("=" * 80) - - -async def test_elicitation_callback_url_extraction( - nc_mcp_oauth_client_with_elicitation, -): - """Test that elicitation callback correctly extracts OAuth URL. - - This validates the URL extraction logic in the callback by examining - the elicitation message format returned by check_logged_in. - """ - client = nc_mcp_oauth_client_with_elicitation - - logger.info("Testing OAuth URL extraction from elicitation message...") - - # Revoke refresh tokens to force elicitation - await revoke_refresh_tokens(client) - - # Call check_logged_in to trigger elicitation - result = await client.call_tool("check_logged_in", arguments={}) - - # Should succeed (callback extracts URL and completes OAuth) - assert result.isError is False - assert "yes" in result.content[0].text.lower() - - # Elicitation should have been triggered - assert client.elicitation_triggered["count"] >= 1 - - logger.info("✓ URL extraction and OAuth completion successful") - - -async def test_elicitation_stores_refresh_token( - nc_mcp_oauth_client_with_elicitation, -): - """Test that refresh token is stored after elicitation completes. - - Validates that after successful OAuth via elicitation: - 1. check_logged_in returns "yes" - 2. check_provisioning_status shows is_provisioned=true - """ - client = nc_mcp_oauth_client_with_elicitation - - logger.info("Testing refresh token storage after elicitation...") - - # Revoke refresh tokens to force elicitation - await revoke_refresh_tokens(client) - - # Complete OAuth via elicitation - result = await client.call_tool("check_logged_in", arguments={}) - assert result.isError is False - assert "yes" in result.content[0].text.lower() - - # Verify refresh token was stored - logger.info("Checking provisioning status...") - status_result = await client.call_tool("check_provisioning_status", arguments={}) - - assert status_result.isError is False - status_text = status_result.content[0].text.lower() - - # Server should report provisioning complete - assert "is_provisioned" in status_text or "offline" in status_text, ( - f"Expected provisioning status, got: {status_text}" - ) - - logger.info("✓ Refresh token stored successfully after elicitation") - - -async def test_second_check_logged_in_does_not_elicit( - nc_mcp_oauth_client_with_elicitation, -): - """Test that second call to check_logged_in does not trigger elicitation. - - After successful OAuth via elicitation: - - First call: triggers elicitation, completes OAuth, returns "yes" - - Second call: no elicitation (already logged in), returns "yes" - """ - client = nc_mcp_oauth_client_with_elicitation - - logger.info("Testing that already-logged-in users don't get elicited...") - - # First call: triggers elicitation - result1 = await client.call_tool("check_logged_in", arguments={}) - assert result1.isError is False - assert "yes" in result1.content[0].text.lower() - - elicitation_count_after_first = client.elicitation_triggered["count"] - logger.info(f"After first call: {elicitation_count_after_first} elicitations") - - # Second call: should NOT trigger elicitation (already logged in) - result2 = await client.call_tool("check_logged_in", arguments={}) - assert result2.isError is False - assert "yes" in result2.content[0].text.lower() - - elicitation_count_after_second = client.elicitation_triggered["count"] - logger.info(f"After second call: {elicitation_count_after_second} elicitations") - - # Elicitation count should be the same (no new elicitation) - assert elicitation_count_after_second == elicitation_count_after_first, ( - "Second check_logged_in should not trigger elicitation " - "(user is already logged in)" - ) - - logger.info("✓ Already-logged-in users don't get redundant elicitations") diff --git a/tests/server/oauth/test_keycloak_dcr.py b/tests/server/oauth/test_keycloak_dcr.py deleted file mode 100644 index 59113bb1..00000000 --- a/tests/server/oauth/test_keycloak_dcr.py +++ /dev/null @@ -1,630 +0,0 @@ -""" -Tests for Dynamic Client Registration (DCR) with Keycloak external IdP. - -These tests verify that DCR (RFC 7591) and client deletion (RFC 7592) -work correctly with Keycloak as an external identity provider: - -1. Client registration via Keycloak's DCR endpoint -2. Token acquisition with dynamically registered client -3. MCP tool execution with Keycloak-issued tokens -4. Client deletion via RFC 7592 -5. Error handling for DCR operations - -This validates ADR-002 external IdP integration where clients are -dynamically provisioned rather than pre-configured. - -Architecture: - MCP Client → Keycloak DCR → Keycloak OAuth → MCP Server → Nextcloud APIs -""" - -import json -import logging -import os -import secrets -import time -from urllib.parse import quote - -import anyio -import httpx -import pytest - -from nextcloud_mcp_server.auth.client_registration import delete_client, register_client - -logger = logging.getLogger(__name__) - -pytestmark = [pytest.mark.integration, pytest.mark.keycloak] - - -# ============================================================================ -# Helper Functions -# ============================================================================ - - -async def handle_keycloak_login(page, username: str, password: str): - """ - Handle Keycloak login page. - - Keycloak uses: - - input#username for username field - - input#password for password field - - Form submission via JavaScript (more reliable than clicking button) - """ - logger.info(f"Handling Keycloak login for user: {username}") - logger.info(f"Current URL before login: {page.url}") - - # Wait for username field and fill it - await page.wait_for_selector("input#username", timeout=10000) - await page.fill("input#username", username) - - # Fill password field - await page.wait_for_selector("input#password", timeout=10000) - await page.fill("input#password", password) - - # Submit form using JavaScript (more reliable than clicking button) - logger.info("Submitting Keycloak login form...") - async with page.expect_navigation(timeout=60000): - await page.evaluate("document.querySelector('form').submit()") - - logger.info(f"✓ Keycloak login completed, redirected to: {page.url}") - - -async def handle_keycloak_consent(page, client_name: str): - """ - Handle Keycloak OAuth consent screen. - - Keycloak consent screen has: - - Checkbox inputs for each scope - - Button with name="accept" to grant consent - - Button with name="cancel" to deny consent - """ - logger.info(f"Handling Keycloak consent for client: {client_name}") - - try: - # Wait for consent screen (button with name="accept") - await page.wait_for_selector('button[name="accept"]', timeout=5000) - - # Click accept button and wait for navigation - async with page.expect_navigation(timeout=60000): - await page.click('button[name="accept"]') - - logger.info("✓ Keycloak consent granted") - except Exception as e: - # Consent screen might not appear if already consented - logger.debug(f"No consent screen or already authorized: {e}") - - -async def get_keycloak_oauth_token_with_client( - browser, - client_id: str, - client_secret: str, - token_endpoint: str, - authorization_endpoint: str, - callback_url: str, - auth_states: dict, - scopes: str = "openid profile email notes:read notes:write", - username: str = "admin", - password: str = "admin", -) -> str: - """ - Obtain OAuth access token from Keycloak using dynamically registered client. - - Args: - browser: Playwright browser instance - client_id: OAuth client ID (from DCR registration) - client_secret: OAuth client secret (from DCR registration) - token_endpoint: Keycloak token endpoint URL - authorization_endpoint: Keycloak authorization endpoint URL - callback_url: Callback URL for OAuth redirect - auth_states: Dict for storing auth codes (from callback server) - scopes: Space-separated list of scopes to request - username: Keycloak username (default: admin) - password: Keycloak password (default: admin) - - Returns: - Access token string - """ - # Generate unique state parameter - state = secrets.token_urlsafe(32) - - # URL-encode scopes - scopes_encoded = quote(scopes, safe="") - - # Construct authorization URL - auth_url = ( - f"{authorization_endpoint}?" - f"response_type=code&" - f"client_id={client_id}&" - f"redirect_uri={quote(callback_url, safe='')}&" - f"state={state}&" - f"scope={scopes_encoded}" - ) - - logger.info("Starting OAuth flow with Keycloak...") - logger.info(f"Authorization URL: {auth_url[:100]}...") - - # Browser automation - context = await browser.new_context(ignore_https_errors=True) - page = await context.new_page() - - try: - await page.goto(auth_url, wait_until="networkidle", timeout=60000) - current_url = page.url - logger.info(f"Current URL after navigation: {current_url[:100]}...") - - # Check if we're on Keycloak login page - if "/realms/" in current_url and "/protocol/openid-connect/auth" in current_url: - # We're on the Keycloak authorization page, might need to login - try: - # Check if login form is present - await page.wait_for_selector("input#username", timeout=3000) - await handle_keycloak_login(page, username, password) - except Exception as e: - logger.debug(f"No login form found, might already be logged in: {e}") - - # Handle consent screen if present - await handle_keycloak_consent(page, "DCR Test Client") - - # Wait for callback - logger.info("Waiting for OAuth callback...") - timeout_seconds = 30 - start_time = time.time() - while state not in auth_states: - if time.time() - start_time > timeout_seconds: - raise TimeoutError( - f"Timeout waiting for OAuth callback (state={state[:16]}...)" - ) - await anyio.sleep(0.5) - - auth_code = auth_states[state] - logger.info(f"Got auth code: {auth_code[:20]}...") - - finally: - await context.close() - - # Exchange code for token - logger.info("Exchanging authorization code for access token...") - async with httpx.AsyncClient(timeout=30.0) as http_client: - token_response = await http_client.post( - token_endpoint, - data={ - "grant_type": "authorization_code", - "code": auth_code, - "redirect_uri": callback_url, - "client_id": client_id, - "client_secret": client_secret, - }, - ) - - token_response.raise_for_status() - token_data = token_response.json() - access_token = token_data.get("access_token") - - if not access_token: - raise ValueError(f"No access_token in response: {token_data}") - - logger.info("Successfully obtained access token from Keycloak") - return access_token - - -# ============================================================================ -# DCR Registration Tests -# ============================================================================ - - -@pytest.mark.integration -async def test_keycloak_dcr_registration(anyio_backend, oauth_callback_server): - """ - Test that DCR registration works with Keycloak. - - Verifies: - - Keycloak's DCR endpoint is discoverable via OIDC discovery - - Client registration succeeds (RFC 7591) - - Registration response includes client_id, client_secret - - Registration response includes RFC 7592 fields (registration_access_token, registration_client_uri) - """ - keycloak_discovery_url = os.getenv( - "OIDC_DISCOVERY_URL", - "http://localhost:8888/realms/nextcloud-mcp/.well-known/openid-configuration", - ) - - auth_states, callback_url = oauth_callback_server - - # OIDC Discovery - logger.info("Discovering Keycloak OIDC endpoints...") - async with httpx.AsyncClient(timeout=30.0) as client: - discovery_response = await client.get(keycloak_discovery_url) - discovery_response.raise_for_status() - oidc_config = discovery_response.json() - - registration_endpoint = oidc_config.get("registration_endpoint") - - if not registration_endpoint: - pytest.skip( - "Keycloak DCR not enabled (no registration_endpoint in discovery)" - ) - - logger.info(f"✓ Found registration endpoint: {registration_endpoint}") - - # Register client - logger.info("Registering OAuth client via Keycloak DCR...") - client_info = await register_client( - nextcloud_url=keycloak_discovery_url.replace( - "/.well-known/openid-configuration", "" - ), - registration_endpoint=registration_endpoint, - client_name="Keycloak DCR Test Client", - redirect_uris=[callback_url], - scopes="openid profile email notes:read notes:write", - token_type=None, # Keycloak doesn't support token_type field - ) - - assert client_info.client_id, "Registration should return client_id" - assert client_info.client_secret, "Registration should return client_secret" - logger.info(f"✓ Client registered: {client_info.client_id[:16]}...") - - # Verify RFC 7592 fields are present - assert client_info.registration_access_token, ( - "Keycloak should return registration_access_token for RFC 7592 deletion" - ) - assert client_info.registration_client_uri, ( - "Keycloak should return registration_client_uri for RFC 7592 operations" - ) - logger.info("✓ RFC 7592 fields present in registration response") - - # Cleanup: Delete the client - logger.info("Cleaning up: deleting test client...") - keycloak_host = keycloak_discovery_url.replace( - "/.well-known/openid-configuration", "" - ) - success = await delete_client( - nextcloud_url=keycloak_host, - client_id=client_info.client_id, - registration_access_token=client_info.registration_access_token, - client_secret=client_info.client_secret, - registration_client_uri=client_info.registration_client_uri, - ) - - assert success, "Cleanup deletion should succeed" - logger.info("✓ Test client deleted successfully") - - -# ============================================================================ -# Complete DCR Lifecycle Tests -# ============================================================================ - - -@pytest.mark.integration -async def test_keycloak_dcr_complete_lifecycle( - anyio_backend, - browser, - oauth_callback_server, - nc_mcp_keycloak_client, -): - """ - Test the complete DCR lifecycle with Keycloak: - 1. Register client via DCR (RFC 7591) - 2. Obtain OAuth token with registered client - 3. Use token to access MCP tools - 4. Delete client via RFC 7592 - - This is the end-to-end test that validates DCR works for external IdPs. - """ - keycloak_discovery_url = os.getenv( - "OIDC_DISCOVERY_URL", - "http://localhost:8888/realms/nextcloud-mcp/.well-known/openid-configuration", - ) - - auth_states, callback_url = oauth_callback_server - - # Step 1: OIDC Discovery - logger.info("Step 1: Discovering Keycloak OIDC endpoints...") - async with httpx.AsyncClient(timeout=30.0) as client: - discovery_response = await client.get(keycloak_discovery_url) - discovery_response.raise_for_status() - oidc_config = discovery_response.json() - - registration_endpoint = oidc_config.get("registration_endpoint") - token_endpoint = oidc_config.get("token_endpoint") - authorization_endpoint = oidc_config.get("authorization_endpoint") - - if not registration_endpoint: - pytest.skip( - "Keycloak DCR not enabled (no registration_endpoint in discovery)" - ) - - logger.info(f"✓ Registration endpoint: {registration_endpoint}") - logger.info(f"✓ Token endpoint: {token_endpoint}") - logger.info(f"✓ Authorization endpoint: {authorization_endpoint}") - - # Step 2: Register client - logger.info("Step 2: Registering OAuth client via Keycloak DCR...") - keycloak_host = keycloak_discovery_url.replace( - "/.well-known/openid-configuration", "" - ) - client_info = await register_client( - nextcloud_url=keycloak_host, - registration_endpoint=registration_endpoint, - client_name="Keycloak DCR Lifecycle Test", - redirect_uris=[callback_url], - scopes="openid profile email notes:read notes:write calendar:read", - token_type=None, # Keycloak doesn't support token_type field - ) - - logger.info(f"✓ Client registered: {client_info.client_id[:16]}...") - logger.info(f" Client secret: {client_info.client_secret[:16]}...") - logger.info( - f" Registration token: {client_info.registration_access_token[:16]}..." - ) - - # Step 3: Obtain OAuth token - logger.info("Step 3: Obtaining OAuth token with registered client...") - access_token = await get_keycloak_oauth_token_with_client( - browser=browser, - client_id=client_info.client_id, - client_secret=client_info.client_secret, - token_endpoint=token_endpoint, - authorization_endpoint=authorization_endpoint, - callback_url=callback_url, - auth_states=auth_states, - scopes="openid profile email notes:read notes:write calendar:read", - username="admin", - password="admin", - ) - - assert access_token, "Failed to obtain access token" - logger.info(f"✓ Access token obtained: {access_token[:30]}...") - - # Step 4: Verify token works with MCP server (optional - requires MCP client setup) - # This step is optional since we already have nc_mcp_keycloak_client fixture - # that uses the pre-configured client. For a full test, you'd create a new - # MCP client with the dynamically registered client, but that's complex. - logger.info("✓ Token can be used with MCP server (verified in other tests)") - - # Step 5: Delete client - logger.info("Step 4: Deleting OAuth client via RFC 7592...") - success = await delete_client( - nextcloud_url=keycloak_host, - client_id=client_info.client_id, - registration_access_token=client_info.registration_access_token, - client_secret=client_info.client_secret, - registration_client_uri=client_info.registration_client_uri, - ) - - assert success, "Client deletion should succeed" - logger.info(f"✓ Client deleted successfully: {client_info.client_id[:16]}...") - - # Step 6: Verify deleted client cannot be used - logger.info("Step 5: Verifying deleted client cannot obtain new tokens...") - async with httpx.AsyncClient(timeout=30.0) as http_client: - try: - # Try to use client credentials grant (should fail) - token_response = await http_client.post( - token_endpoint, - data={ - "grant_type": "client_credentials", - "client_id": client_info.client_id, - "client_secret": client_info.client_secret, - }, - ) - - # Accept 400 or 401 as valid rejection - if token_response.status_code in [400, 401]: - logger.info( - f"✓ Deleted client correctly rejected ({token_response.status_code})" - ) - else: - pytest.fail( - f"Deleted client should not be able to obtain tokens, " - f"but got status {token_response.status_code}" - ) - - except httpx.HTTPStatusError as e: - if e.response.status_code in [400, 401]: - logger.info("✓ Deleted client correctly rejected") - else: - raise - - logger.info("✅ Complete Keycloak DCR lifecycle test passed!") - - -# ============================================================================ -# Error Handling Tests -# ============================================================================ - - -@pytest.mark.integration -async def test_keycloak_dcr_delete_with_wrong_token( - anyio_backend, - oauth_callback_server, -): - """ - Test that deletion fails with wrong registration_access_token. - - Verifies: - 1. Client registration succeeds - 2. Deletion with wrong registration_access_token fails - 3. Deletion with correct registration_access_token succeeds - """ - keycloak_discovery_url = os.getenv( - "OIDC_DISCOVERY_URL", - "http://localhost:8888/realms/nextcloud-mcp/.well-known/openid-configuration", - ) - - auth_states, callback_url = oauth_callback_server - - # OIDC Discovery - async with httpx.AsyncClient(timeout=30.0) as client: - discovery_response = await client.get(keycloak_discovery_url) - discovery_response.raise_for_status() - oidc_config = discovery_response.json() - - registration_endpoint = oidc_config.get("registration_endpoint") - - if not registration_endpoint: - pytest.skip("Keycloak DCR not enabled") - - # Register client - logger.info("Registering OAuth client for wrong token test...") - keycloak_host = keycloak_discovery_url.replace( - "/.well-known/openid-configuration", "" - ) - client_info = await register_client( - nextcloud_url=keycloak_host, - registration_endpoint=registration_endpoint, - client_name="Keycloak DCR Wrong Token Test", - redirect_uris=[callback_url], - scopes="openid profile email", - token_type=None, # Keycloak doesn't support token_type field - ) - - logger.info(f"Client registered: {client_info.client_id[:16]}...") - - # Try to delete with wrong registration_access_token - logger.info("Attempting deletion with wrong registration_access_token...") - wrong_token = "wrong_token_" + secrets.token_urlsafe(32) - - success = await delete_client( - nextcloud_url=keycloak_host, - client_id=client_info.client_id, - registration_access_token=wrong_token, - client_secret=client_info.client_secret, - registration_client_uri=client_info.registration_client_uri, - ) - - assert not success, "Deletion with wrong token should fail" - logger.info("✓ Deletion correctly failed with wrong token") - - # Clean up: Delete with correct token - logger.info("Cleaning up: deleting with correct registration_access_token...") - success = await delete_client( - nextcloud_url=keycloak_host, - client_id=client_info.client_id, - registration_access_token=client_info.registration_access_token, - client_secret=client_info.client_secret, - registration_client_uri=client_info.registration_client_uri, - ) - - assert success, "Deletion with correct token should succeed" - logger.info("✓ Cleanup successful") - - -@pytest.mark.integration -async def test_keycloak_dcr_deletion_is_idempotent( - anyio_backend, - oauth_callback_server, -): - """ - Test that deleting the same client twice fails gracefully on second attempt. - - Verifies: - 1. First deletion succeeds - 2. Second deletion fails gracefully (no exception, returns False) - """ - keycloak_discovery_url = os.getenv( - "OIDC_DISCOVERY_URL", - "http://localhost:8888/realms/nextcloud-mcp/.well-known/openid-configuration", - ) - - auth_states, callback_url = oauth_callback_server - - # OIDC Discovery - async with httpx.AsyncClient(timeout=30.0) as client: - discovery_response = await client.get(keycloak_discovery_url) - discovery_response.raise_for_status() - oidc_config = discovery_response.json() - - registration_endpoint = oidc_config.get("registration_endpoint") - - if not registration_endpoint: - pytest.skip("Keycloak DCR not enabled") - - # Register client - logger.info("Registering OAuth client for idempotency test...") - keycloak_host = keycloak_discovery_url.replace( - "/.well-known/openid-configuration", "" - ) - client_info = await register_client( - nextcloud_url=keycloak_host, - registration_endpoint=registration_endpoint, - client_name="Keycloak DCR Idempotency Test", - redirect_uris=[callback_url], - scopes="openid profile email", - token_type=None, # Keycloak doesn't support token_type field - ) - - logger.info(f"Client registered: {client_info.client_id[:16]}...") - - # First deletion - logger.info("First deletion attempt...") - success = await delete_client( - nextcloud_url=keycloak_host, - client_id=client_info.client_id, - registration_access_token=client_info.registration_access_token, - client_secret=client_info.client_secret, - registration_client_uri=client_info.registration_client_uri, - ) - - assert success, "First deletion should succeed" - logger.info("✓ First deletion succeeded") - - # Second deletion (should fail gracefully) - logger.info("Second deletion attempt (should fail)...") - success = await delete_client( - nextcloud_url=keycloak_host, - client_id=client_info.client_id, - registration_access_token=client_info.registration_access_token, - client_secret=client_info.client_secret, - registration_client_uri=client_info.registration_client_uri, - ) - - assert not success, "Second deletion should fail (client already deleted)" - logger.info("✓ Second deletion correctly failed (client already deleted)") - - -# ============================================================================ -# Documentation Tests -# ============================================================================ - - -async def test_keycloak_dcr_architecture(): - """ - Document the Keycloak DCR architecture for reference. - - This test captures the design and flow for DCR with external IdPs. - """ - architecture = { - "flow": [ - "1. MCP client discovers Keycloak OIDC endpoints via .well-known/openid-configuration", - "2. MCP client registers via Keycloak DCR endpoint (RFC 7591)", - "3. Keycloak returns client_id, client_secret, registration_access_token", - "4. MCP client uses credentials to obtain OAuth token", - "5. MCP client uses token to authenticate with MCP server", - "6. MCP server validates token via Nextcloud user_oidc app", - "7. When done, MCP client deletes registration via RFC 7592", - ], - "components": { - "keycloak_dcr": "Dynamic Client Registration endpoint (RFC 7591)", - "keycloak_oauth": "OAuth/OIDC provider for authentication", - "mcp_server": "MCP server with external IdP config", - "nextcloud": "API server with user_oidc app for token validation", - }, - "advantages": [ - "No manual client pre-configuration required", - "Clients can self-register and self-cleanup", - "Standards-based (RFC 7591, RFC 7592)", - "Works with any compliant OIDC provider", - "Supports dynamic callback URL registration", - ], - "security": [ - "Registration tokens protect client management operations", - "Clients can only delete themselves (not others)", - "Token validation ensures only authorized access", - "Automatic cleanup prevents client sprawl", - ], - } - - logger.info("Keycloak DCR Architecture:") - - logger.info(json.dumps(architecture, indent=2)) - - assert True diff --git a/tests/server/oauth/test_keycloak_external_idp.py b/tests/server/oauth/test_keycloak_external_idp.py deleted file mode 100644 index 99da439a..00000000 --- a/tests/server/oauth/test_keycloak_external_idp.py +++ /dev/null @@ -1,566 +0,0 @@ -"""Keycloak External IdP Integration Tests. - -Tests verify ADR-002 external identity provider integration where: -1. Keycloak acts as external OAuth/OIDC provider -2. MCP server validates tokens via Nextcloud user_oidc app -3. Nextcloud auto-provisions users from Keycloak token claims -4. MCP tools execute successfully with Keycloak tokens - -Architecture: - MCP Client → Keycloak (OAuth) → MCP Server → Nextcloud user_oidc (validates) → APIs - -Tests: -1. Keycloak OAuth token acquisition via Playwright -2. MCP client connection to mcp-keycloak service (port 8002) -3. Token validation through Nextcloud user_oidc app -4. MCP tool execution with Keycloak tokens -5. User auto-provisioning from Keycloak claims -6. Scope-based tool filtering with Keycloak JWT tokens -""" - -import json -import logging - -import pytest - -from nextcloud_mcp_server.client import NextcloudClient - -logger = logging.getLogger(__name__) - -pytestmark = [pytest.mark.integration, pytest.mark.keycloak] - - -# ============================================================================ -# OAuth Token Acquisition Tests -# ============================================================================ - - -async def test_keycloak_oauth_token_acquisition(keycloak_oauth_token): - """Test that Playwright can obtain OAuth token from Keycloak. - - Verifies: - - Playwright automation handles Keycloak login page (input#username, input#password) - - Keycloak consent screen is handled correctly - - Authorization code is exchanged for access token - - Token is returned successfully - - This is a foundational test - if this fails, all other Keycloak tests will fail. - """ - assert keycloak_oauth_token is not None - assert isinstance(keycloak_oauth_token, str) - assert len(keycloak_oauth_token) > 100 # Tokens should be substantial length - - logger.info( - f"✓ Keycloak OAuth token acquired (length: {len(keycloak_oauth_token)})" - ) - logger.info(f" Token prefix: {keycloak_oauth_token[:50]}...") - - -async def test_keycloak_oauth_client_credentials_discovery( - keycloak_oauth_client_credentials, -): - """Test Keycloak OIDC discovery and credential loading. - - Verifies: - - OIDC discovery endpoint is accessible - - Token and authorization endpoints are discovered - - Static client credentials are loaded from environment - - Callback server is initialized - """ - ( - client_id, - client_secret, - callback_url, - token_endpoint, - authorization_endpoint, - ) = keycloak_oauth_client_credentials - - assert client_id == "nextcloud-mcp-server" - assert client_secret == "mcp-secret-change-in-production" - assert callback_url.startswith("http://") - # With --hostname-backchannel-dynamic, external clients see localhost:8888 - assert "localhost:8888" in token_endpoint or "keycloak" in token_endpoint - assert ( - "localhost:8888" in authorization_endpoint - or "keycloak" in authorization_endpoint - ) - assert "/realms/nextcloud-mcp/" in token_endpoint - - logger.info("✓ Keycloak OIDC discovery successful") - logger.info(f" Client ID: {client_id}") - logger.info(f" Token endpoint: {token_endpoint}") - logger.info(f" Authorization endpoint: {authorization_endpoint}") - - -# ============================================================================ -# MCP Server Connectivity Tests -# ============================================================================ - - -async def test_mcp_client_connects_to_keycloak_server(nc_mcp_keycloak_client): - """Test MCP client can connect to mcp-keycloak service (port 8002). - - Verifies: - - MCP client session is established - - Server responds to list_tools request - - Tools are available for use - """ - result = await nc_mcp_keycloak_client.list_tools() - - assert result is not None - assert len(result.tools) > 0 - - logger.info( - f"✓ MCP client connected to Keycloak server with {len(result.tools)} tools" - ) - - -async def test_external_idp_server_initialization(nc_mcp_keycloak_client): - """Test that MCP server correctly initializes with external IdP configuration. - - Verifies: - - Server auto-detects external IdP mode (issuer != Nextcloud host) - - Server reports correct provider type - - All expected tools are registered - - The server should log messages like: - - "✓ Detected external IdP mode (issuer: http://keycloak:8080/realms/nextcloud-mcp != Nextcloud: http://app:80)" - """ - result = await nc_mcp_keycloak_client.list_tools() - - # Verify we have a full set of tools (not filtered to specific apps) - tool_names = [tool.name for tool in result.tools] - - # Should have tools from multiple apps - has_notes = any("notes" in name for name in tool_names) - has_calendar = any("calendar" in name for name in tool_names) - has_files = any("webdav" in name for name in tool_names) - - assert has_notes, "Missing Notes tools" - assert has_calendar, "Missing Calendar tools" - assert has_files, "Missing WebDAV/Files tools" - - logger.info("✓ MCP server initialized with external IdP mode") - logger.info(f" Tools from multiple apps detected: {len(result.tools)} total") - - -# ============================================================================ -# Token Validation Tests -# ============================================================================ - - -async def test_external_idp_token_validation(nc_mcp_keycloak_client): - """Test that Keycloak tokens are validated via Nextcloud user_oidc app. - - Token flow: - 1. Keycloak issues OAuth token - 2. MCP client sends token to MCP server - 3. MCP server passes token to Nextcloud user_oidc app - 4. user_oidc validates token with Keycloak (JWKS or introspection) - 5. Nextcloud returns user info to MCP server - 6. MCP server uses token to access Nextcloud APIs - - This test verifies the entire flow works. - """ - # Execute a read operation (requires token validation) - result = await nc_mcp_keycloak_client.call_tool( - "nc_notes_search_notes", arguments={"query": ""} - ) - - assert result.isError is False, f"Tool execution failed: {result.content}" - assert result.content is not None - response_data = json.loads(result.content[0].text) - - # Successful response means token was validated and user was authenticated - assert "results" in response_data - assert isinstance(response_data["results"], list) - - logger.info("✓ Keycloak token validated successfully via Nextcloud user_oidc app") - logger.info(f" Tool execution returned {len(response_data['results'])} results") - - -# ============================================================================ -# Tool Execution Tests -# ============================================================================ - - -async def test_tools_work_with_keycloak_token(nc_mcp_keycloak_client): - """Test that MCP tools execute successfully with Keycloak OAuth tokens. - - Verifies end-to-end functionality: - - Read operations work (nc_notes_search_notes) - - Write operations work (nc_notes_create_note) - - Different apps work (Notes, Calendar, Files) - """ - # Test 1: Read operation (Notes) - search_result = await nc_mcp_keycloak_client.call_tool( - "nc_notes_search_notes", arguments={"query": ""} - ) - assert search_result.isError is False - logger.info("✓ Read operation successful (nc_notes_search_notes)") - - # Test 2: Write operation (Notes) - create_result = await nc_mcp_keycloak_client.call_tool( - "nc_notes_create_note", - arguments={ - "title": "Keycloak Test Note", - "content": "Created via external IdP token", - "category": "Test", - }, - ) - assert create_result.isError is False - create_data = json.loads(create_result.content[0].text) - note_id = create_data["id"] - logger.info(f"✓ Write operation successful (created note {note_id})") - - # Test 3: Different app (Calendar) - calendar_result = await nc_mcp_keycloak_client.call_tool( - "nc_calendar_list_calendars", arguments={} - ) - assert calendar_result.isError is False - logger.info("✓ Calendar tool execution successful") - - # Test 4: File operations (WebDAV) - files_result = await nc_mcp_keycloak_client.call_tool( - "nc_webdav_list_directory", arguments={"path": "/"} - ) - assert files_result.isError is False - logger.info("✓ WebDAV tool execution successful") - - # Cleanup: Delete test note - await nc_mcp_keycloak_client.call_tool( - "nc_notes_delete_note", arguments={"note_id": note_id} - ) - logger.info(f"✓ Cleanup: Deleted test note {note_id}") - - -async def test_keycloak_token_persistence(nc_mcp_keycloak_client): - """Test that Keycloak token works across multiple operations. - - Verifies: - - Token is properly cached by MCP server - - Token can be reused for multiple API calls - - No re-authentication is required between calls - """ - # Execute multiple operations with same session - operations = [ - ("nc_notes_search_notes", {"query": ""}), - ("nc_calendar_list_calendars", {}), - ("nc_webdav_list_directory", {"path": "/"}), - ] - - for tool_name, arguments in operations: - result = await nc_mcp_keycloak_client.call_tool(tool_name, arguments=arguments) - assert result.isError is False, f"Failed to execute {tool_name}" - logger.info(f"✓ {tool_name} executed successfully") - - logger.info("✓ Keycloak token persistence verified (3 operations with same token)") - - -# ============================================================================ -# User Provisioning Tests -# ============================================================================ - - -async def test_user_auto_provisioning(nc_client: NextcloudClient, keycloak_oauth_token): - """Test that Nextcloud validates users from Keycloak token claims. - - When a user authenticates with Keycloak, Nextcloud's user_oidc app - validates the token and authenticates the user. In this test setup, - the Keycloak 'admin' user maps to the Nextcloud 'admin' user. - - Verification: - 1. User exists in Nextcloud after OAuth authentication - 2. User can access Nextcloud APIs with Keycloak token - 3. Bearer token validation is working correctly - - Note: With bearer-provisioning enabled, user_oidc would auto-provision - new users from token claims, but since we use 'admin' in both Keycloak - and Nextcloud, they map to the same user. - """ - # Get list of users (returns List[str] of user IDs) - user_ids = await nc_client.users.search_users() - - logger.info(f"Found {len(user_ids)} users in Nextcloud") - logger.info(f"Users: {user_ids}") - - # Verify the admin user exists (used for authentication) - assert "admin" in user_ids, "Expected 'admin' user to exist in Nextcloud" - - # Verify we can access APIs with the Keycloak token (already tested in previous tests) - # The fact that we got this far means bearer token validation is working - - logger.info("✓ User authentication and bearer token validation verified") - logger.info(f" Total users: {len(user_ids)}") - logger.info(" Bearer provisioning is enabled and working correctly") - - -# ============================================================================ -# Scope-Based Authorization Tests -# ============================================================================ - - -async def test_scope_filtering_with_keycloak(nc_mcp_keycloak_client): - """Test that tool filtering works correctly with Keycloak JWT scopes. - - Keycloak tokens should include scopes in JWT payload (if JWT format). - The MCP server should filter tools based on these scopes. - - Expected scopes (from docker-compose.yml): - - openid profile email offline_access - - notes:read notes:write - - calendar:read calendar:write - - contacts:read contacts:write - - etc. - - Tools should be filtered accordingly. - """ - result = await nc_mcp_keycloak_client.list_tools() - tool_names = [tool.name for tool in result.tools] - - # With full scopes, all app tools should be available - expected_tools = [ - "nc_notes_get_note", # notes:read - "nc_notes_create_note", # notes:write - "nc_calendar_list_calendars", # calendar:read - "nc_calendar_create_event", # calendar:write - "nc_webdav_list_directory", # files:read - "nc_webdav_write_file", # files:write - ] - - for tool_name in expected_tools: - assert tool_name in tool_names, f"Expected tool {tool_name} not found" - - logger.info("✓ Scope-based tool filtering working with Keycloak tokens") - logger.info(f" Available tools: {len(tool_names)}") - - -# ============================================================================ -# Error Handling Tests -# ============================================================================ - - -async def test_keycloak_error_handling(nc_mcp_keycloak_client): - """Test error handling with Keycloak tokens. - - Verifies: - - Invalid operations return proper errors - - Token validation errors are handled correctly - - API errors propagate correctly through the chain - """ - # Try to get a non-existent note - result = await nc_mcp_keycloak_client.call_tool( - "nc_notes_get_note", arguments={"note_id": 999999} - ) - - # Should get an error (note doesn't exist) - assert result.isError is True - logger.info( - "✓ Keycloak OAuth server correctly handles errors for invalid operations" - ) - - -# ============================================================================ -# Documentation Tests -# ============================================================================ - - -async def test_external_idp_architecture(): - """Document the external IdP architecture (ADR-002). - - This test captures the design and flow for reference. - """ - architecture = { - "flow": [ - "1. User authenticates with Keycloak (external IdP)", - "2. Keycloak issues OAuth access token with scopes", - "3. MCP client uses token to authenticate with MCP server", - "4. MCP server receives token and passes to Nextcloud", - "5. Nextcloud user_oidc app validates token with Keycloak", - "6. Nextcloud auto-provisions user from token claims (if first login)", - "7. Nextcloud returns validated user info to MCP server", - "8. MCP server executes tool using validated token", - ], - "components": { - "keycloak": "External OAuth/OIDC provider (port 8888)", - "mcp_server": "MCP server with external IdP config (port 8002)", - "nextcloud": "API server with user_oidc app (port 8080)", - "user_oidc": "Nextcloud app that validates external IdP tokens", - }, - "configuration": { - "keycloak_realm": "nextcloud-mcp", - "keycloak_client": "nextcloud-mcp-server", - "nextcloud_provider": "keycloak (via user_oidc app)", - "token_validation": "Keycloak JWKS or introspection endpoint", - }, - "advantages": [ - "No admin credentials needed in MCP server", - "Centralized identity management", - "Standards-based (RFC 6749, RFC 7662, RFC 9068)", - "Supports enterprise IdPs (Keycloak, Auth0, Okta, etc.)", - "User auto-provisioning from IdP claims", - ], - } - - logger.info("External IdP Architecture (ADR-002):") - logger.info(json.dumps(architecture, indent=2)) - - assert True - - -# ============================================================================ -# Scope-Based Authorization Tests (JWT Token Filtering) -# ============================================================================ - - -async def test_keycloak_read_only_token_filters_write_tools( - nc_mcp_keycloak_client_read_only, -): - """Test that a Keycloak token with only read scopes filters out write tools.""" - # Connect with token that has only read scopes - result = await nc_mcp_keycloak_client_read_only.list_tools() - assert result is not None - assert len(result.tools) > 0 - - tool_names = [tool.name for tool in result.tools] - logger.info(f"Keycloak read-only token sees {len(tool_names)} tools") - - # Verify read tools are present - expected_read_tools = [ - "nc_notes_get_note", # notes:read - "nc_notes_search_notes", # notes:read - "nc_calendar_list_calendars", # calendar:read - "nc_calendar_get_event", # calendar:read - ] - - for tool in expected_read_tools: - assert tool in tool_names, f"Expected read tool {tool} not found in tool list" - - # Verify write tools are NOT present (filtered out) - write_tools_should_be_filtered = [ - "nc_notes_create_note", # notes:write - "nc_notes_update_note", # notes:write - "nc_notes_delete_note", # notes:write - "nc_calendar_create_event", # calendar:write - "nc_calendar_update_event", # calendar:write - "nc_calendar_delete_event", # calendar:write - ] - - for tool in write_tools_should_be_filtered: - assert tool not in tool_names, ( - f"Write tool {tool} should be filtered out but was found in tool list" - ) - - logger.info( - f"✅ Keycloak read-only token properly filters tools: {len(tool_names)} read tools visible, " - f"write tools hidden" - ) - - -async def test_keycloak_write_only_token_filters_read_tools( - nc_mcp_keycloak_client_write_only, -): - """Test that a Keycloak token with only write scopes filters out read tools.""" - # Connect with token that has only write scopes - result = await nc_mcp_keycloak_client_write_only.list_tools() - assert result is not None - assert len(result.tools) > 0 - - tool_names = [tool.name for tool in result.tools] - logger.info(f"Keycloak write-only token sees {len(tool_names)} tools") - - # Verify write tools are present - expected_write_tools = [ - "nc_notes_create_note", # notes:write - "nc_notes_update_note", # notes:write - "nc_notes_delete_note", # notes:write - "nc_calendar_create_event", # calendar:write - "nc_calendar_update_event", # calendar:write - "nc_calendar_delete_event", # calendar:write - ] - - for tool in expected_write_tools: - assert tool in tool_names, f"Expected write tool {tool} not found in tool list" - - # Verify read-only tools are NOT present (write-only scope) - read_tools_should_be_filtered = [ - "nc_notes_get_note", # notes:read - "nc_notes_search_notes", # notes:read - "nc_calendar_list_calendars", # calendar:read - "nc_calendar_get_event", # calendar:read - ] - - for tool in read_tools_should_be_filtered: - assert tool not in tool_names, ( - f"Read tool {tool} should be filtered out but was found in tool list" - ) - - logger.info( - f"✅ Keycloak write-only token properly filters tools: {len(tool_names)} write tools visible, " - f"read tools hidden" - ) - - -async def test_keycloak_full_access_token_shows_all_tools(nc_mcp_keycloak_client): - """Test that a Keycloak token with both read and write scopes sees all tools.""" - # Connect with token that has both read and write scopes - result = await nc_mcp_keycloak_client.list_tools() - assert result is not None - assert len(result.tools) > 0 - - tool_names = [tool.name for tool in result.tools] - logger.info(f"Keycloak full access token sees {len(tool_names)} tools") - - # Verify both read and write tools are present - expected_read_tools = [ - "nc_notes_get_note", # notes:read - "nc_notes_search_notes", # notes:read - "nc_calendar_list_calendars", # calendar:read - ] - - expected_write_tools = [ - "nc_notes_create_note", # notes:write - "nc_calendar_create_event", # calendar:write - ] - - for tool in expected_read_tools: - assert tool in tool_names, f"Expected read tool {tool} not found" - - for tool in expected_write_tools: - assert tool in tool_names, f"Expected write tool {tool} not found" - - # Should have all 90+ tools (both read and write) - assert len(tool_names) >= 90 - - logger.info( - f"✅ Keycloak full access token sees all tools: {len(tool_names)} total (read + write)" - ) - - -async def test_keycloak_no_custom_scopes_returns_zero_tools( - nc_mcp_keycloak_client_no_custom_scopes, -): - """ - Test that a Keycloak JWT token with only OIDC default scopes returns 0 tools. - - This tests the security behavior when a user declines to grant custom scopes during consent. - Expected: JWT token has scopes=['openid', 'profile', 'email'] but no custom scopes. - All tools require at least one custom scope, so they should all be filtered out. - """ - # Connect with JWT token that has NO custom scopes (only openid, profile, email) - result = await nc_mcp_keycloak_client_no_custom_scopes.list_tools() - assert result is not None - - tool_names = [tool.name for tool in result.tools] - logger.info( - f"Keycloak JWT token with no custom scopes sees {len(tool_names)} tools (should be 0)" - ) - - # All tools require custom scopes, so should be filtered out - assert len(tool_names) == 0, ( - f"Expected 0 tools but got {len(tool_names)}: {tool_names[:10]}" - ) - - logger.info( - "✅ Keycloak JWT token without custom scopes correctly returns 0 tools (all filtered out)" - ) diff --git a/tests/server/oauth/test_login_elicitation.py b/tests/server/oauth/test_login_elicitation.py deleted file mode 100644 index 738b7c86..00000000 --- a/tests/server/oauth/test_login_elicitation.py +++ /dev/null @@ -1,246 +0,0 @@ -"""Integration tests for login elicitation flow (ADR-006 Interim Implementation). - -Tests verify: -1. check_logged_in tool with elicitation for unauthenticated users -2. Elicitation contains login URL in message -3. User can complete login via OAuth -4. After login, check_logged_in returns "yes" -5. Already-authenticated users get immediate "yes" response -6. Elicitation decline/cancel handling -""" - -import logging -import re - -import pytest - -logger = logging.getLogger(__name__) - -pytestmark = [pytest.mark.integration, pytest.mark.oauth] - - -async def test_check_logged_in_elicitation_flow( - nc_mcp_oauth_client, browser, oauth_callback_server -): - """Test that check_logged_in elicits login for unauthenticated user. - - This test validates the complete elicitation flow: - 1. Call check_logged_in on authenticated client (already has refresh token) - 2. Verify tool returns "yes" without elicitation - 3. Extract and validate the elicitation URL format from response - 4. Verify refresh token exists after successful OAuth flow - - Note: Actual elicitation handling requires MCP protocol support in the test client. - This test validates the response format and token storage. - """ - # Call check_logged_in tool on authenticated client - logger.info("Calling check_logged_in on authenticated client") - result = await nc_mcp_oauth_client.call_tool("check_logged_in", arguments={}) - - assert result.isError is False, f"Tool execution failed: {result.content}" - assert result.content is not None - - response_text = result.content[0].text - logger.info(f"check_logged_in response: {response_text}") - - # Since nc_mcp_oauth_client fixture already completes OAuth during setup, - # the user should already be provisioned and we expect "yes" - # For unauthenticated users, the response would contain an elicitation URL - # Note: Test framework may return "elicitation not supported" if MCP elicitation is unavailable - assert ( - "yes" in response_text.lower() - or "http" in response_text.lower() - or "elicitation not supported" in response_text.lower() - ), f"Unexpected response: {response_text}" - - # If response contains a URL (elicitation case), validate its format - if "http" in response_text: - url_pattern = r"https?://[^\s]+" - urls = re.findall(url_pattern, response_text) - assert len(urls) > 0, "Expected elicitation URL in response" - - login_url = urls[0] - logger.info(f"Elicitation URL: {login_url}") - - # Validate URL points to MCP server's Flow 2 endpoint - assert "/oauth/authorize-nextcloud" in login_url, ( - f"Expected URL to point to MCP server Flow 2 endpoint, got: {login_url}" - ) - # Validate URL contains state parameter - assert "state=" in login_url, "Expected state parameter in elicitation URL" - elif "elicitation not supported" in response_text.lower(): - logger.info( - "✓ Test client doesn't support elicitation - this is expected in test environment" - ) - - -async def test_check_logged_in_already_authenticated(nc_mcp_oauth_client): - """Test that check_logged_in returns 'yes' for authenticated user. - - This test verifies that if the user has already completed Flow 2 - (resource provisioning), the tool immediately returns "yes" without - elicitation. - """ - logger.info("Calling check_logged_in on authenticated client") - - # Since we're using the nc_mcp_oauth_client fixture which completes - # OAuth during setup, the user should already be provisioned - result = await nc_mcp_oauth_client.call_tool("check_logged_in", arguments={}) - - assert result.isError is False, f"Tool execution failed: {result.content}" - assert result.content is not None - - response_text = result.content[0].text - logger.info(f"Response: {response_text}") - - # Check for valid responses: - # - "yes" (already logged in) - # - "not enabled" (offline access not enabled) - # - "not configured" (MCP_SERVER_CLIENT_ID not set) - # - "elicitation not supported" (test environment limitation) - assert ( - "yes" in response_text.lower() - or "not enabled" in response_text.lower() - or "not configured" in response_text.lower() - or "elicitation not supported" in response_text.lower() - ) - - -async def test_check_logged_in_url_format(nc_mcp_oauth_client): - """Test that login URL (when needed) follows correct OAuth format. - - This test verifies that if the tool needs to provide a login URL, - the URL contains the correct OAuth parameters for Flow 2. - """ - # Call the tool - result = await nc_mcp_oauth_client.call_tool("check_logged_in", arguments={}) - - assert result.isError is False, f"Tool execution failed: {result.content}" - assert result.content is not None - - response_text = result.content[0].text - logger.info(f"Response: {response_text}") - - # If response contains a URL, validate it - url_pattern = r"https?://[^\s]+" - urls = re.findall(url_pattern, response_text) - - if urls: - login_url = urls[0] - logger.info(f"Found login URL: {login_url}") - - # Validate OAuth parameters - assert "response_type=code" in login_url - assert "client_id=" in login_url - assert "redirect_uri=" in login_url - assert "scope=" in login_url - assert "state=" in login_url - assert "openid" in login_url # Should request openid scope - - # Validate callback URL (unified endpoint without query params) - # Note: redirect_uri should be /oauth/callback (no query params) - # Flow type is determined by session lookup, not URL params - assert ( - "/oauth/callback" in login_url - or "callback-nextcloud" in login_url # Legacy support - or "authorize-nextcloud" in login_url - ) - - -async def test_check_logged_in_with_user_id(nc_mcp_oauth_client): - """Test that check_logged_in accepts optional user_id parameter. - - This verifies the tool can be called with an explicit user_id. - """ - result = await nc_mcp_oauth_client.call_tool( - "check_logged_in", arguments={"user_id": "testuser"} - ) - - assert result.isError is False, f"Tool execution failed: {result.content}" - assert result.content is not None - - response_text = result.content[0].text - logger.info(f"Response with user_id: {response_text}") - - # Should get some response (either yes or not logged in) - assert len(response_text) > 0 - - -async def test_check_logged_in_tool_metadata(nc_mcp_oauth_client): - """Test that check_logged_in tool has correct metadata.""" - tools = await nc_mcp_oauth_client.list_tools() - assert tools is not None - - # Find the check_logged_in tool - check_logged_in_tool = None - for tool in tools.tools: - if tool.name == "check_logged_in": - check_logged_in_tool = tool - break - - assert check_logged_in_tool is not None, "check_logged_in tool not found" - logger.info(f"Tool: {check_logged_in_tool.name}") - logger.info(f"Description: {check_logged_in_tool.description}") - - # Verify description mentions login - assert "login" in check_logged_in_tool.description.lower() - - # Tool should have openid scope requirement - # (This would need to be verified via tool schema if exposed) - - -async def test_elicitation_url_and_refresh_token_flow(nc_mcp_oauth_client): - """Test that MCP server validates refresh tokens after OAuth completion. - - This test validates the server's refresh token handling through its API: - 1. Call check_provisioning_status to verify server-side token validation - 2. Server responses indicate token state: - - is_provisioned=True: Server has valid refresh token - - is_provisioned=False: No token or invalid token - - Error response: Token validation failed - - The test does NOT directly access refresh token storage - it relies on - the MCP server to validate tokens internally and report status via API. - """ - logger.info("Testing server-side refresh token validation via API") - - # Call check_provisioning_status - the server will internally: - # 1. Check if refresh token exists for the user - # 2. Validate the refresh token is not expired - # 3. Return provisioning status - result = await nc_mcp_oauth_client.call_tool( - "check_provisioning_status", arguments={} - ) - - assert result.isError is False, f"Tool execution failed: {result.content}" - assert result.content is not None - - response_text = result.content[0].text - logger.info(f"Provisioning status response: {response_text}") - - # Parse the response to validate server's token validation - # Expected responses: - # 1. "is_provisioned: true" - server validated token successfully - # 2. "is_provisioned: false" - no token or invalid token - # 3. Error message - token validation failed - - if "is_provisioned" in response_text.lower(): - if "true" in response_text.lower(): - logger.info("✓ Server validated refresh token: is_provisioned=True") - logger.info(" This confirms the server has a valid refresh token stored") - else: - logger.info("Server reports: is_provisioned=False (no valid token)") - elif "error" in response_text.lower(): - logger.warning( - f"Server returned error during token validation: {response_text}" - ) - else: - logger.info(f"Server response: {response_text}") - - # The key validation: Server must return a valid response - # (not an error), proving it can check its own refresh token state - assert ( - "is_provisioned" in response_text.lower() or "offline" in response_text.lower() - ), f"Expected provisioning status response from server, got: {response_text}" - - logger.info("✓ Server successfully validated refresh token state via API") diff --git a/tests/server/oauth/test_nc_php_app_debug.py b/tests/server/oauth/test_nc_php_app_debug.py deleted file mode 100644 index c2bce01f..00000000 --- a/tests/server/oauth/test_nc_php_app_debug.py +++ /dev/null @@ -1,92 +0,0 @@ -"""Debug test to capture what's on the NC PHP app settings page.""" - -import logging -import os - -import pytest - -logger = logging.getLogger(__name__) - -pytestmark = [pytest.mark.integration, pytest.mark.oauth] - - -async def test_capture_settings_page(browser, configure_astrolabe_for_mcp_server): - """Capture what's actually rendered on the personal settings page.""" - # Configure Astrolabe for mcp-oauth server - await configure_astrolabe_for_mcp_server( - mcp_server_internal_url="http://mcp-oauth:8001", - mcp_server_public_url="http://localhost:8001", - ) - - nextcloud_host = os.getenv("NEXTCLOUD_HOST", "http://localhost:8080") - username = os.getenv("NEXTCLOUD_USERNAME", "admin") - password = os.getenv("NEXTCLOUD_PASSWORD", "admin") - - context = await browser.new_context() - page = await context.new_page() - - try: - # Login - logger.info(f"Logging in to {nextcloud_host} as {username}...") - await page.goto(f"{nextcloud_host}/login") - await page.fill('input[name="user"]', username) - await page.fill('input[name="password"]', password) - await page.click('button[type="submit"]') - await page.wait_for_url(f"{nextcloud_host}/apps/dashboard/", timeout=10000) - logger.info("✓ Logged in") - - # Navigate to settings - logger.info("Navigating to personal MCP settings...") - await page.goto(f"{nextcloud_host}/settings/user/astrolabe") - await page.wait_for_load_state("networkidle") - - # Capture page content - page_content = await page.content() - - # Save screenshot - screenshot_path = "/tmp/nc-php-app-settings-debug.png" - await page.screenshot(path=screenshot_path, full_page=True) - logger.info(f"Screenshot saved to: {screenshot_path}") - - # Log what we found - logger.info(f"Page URL: {page.url}") - logger.info(f"Page title: {await page.title()}") - - # Check for key strings (Vue 3 UI) - checks = [ - "Enable Semantic Search", # oauth-required.php authorization button - "Service Status", # personal.php when authorized - "Background Sync Access", # personal.php when authorized - "What happens next?", # oauth-required.php steps - "Astrolabe", # Header - ] - - for check in checks: - found = check in page_content - logger.info(f" '{check}': {'FOUND' if found else 'NOT FOUND'}") - - # Print first 500 chars of body - body = await page.locator("body").text_content() - logger.info(f"Body text (first 500 chars): {body[:500] if body else 'NO BODY'}") - - # Try to find links - links = await page.locator("a").all_text_contents() - logger.info(f"Found {len(links)} links on page") - for i, link_text in enumerate(links[:10]): - logger.info(f" Link {i}: {link_text}") - - # Check the Enable Semantic Search button href - try: - btn = page.locator('a:has-text("Enable Semantic Search")') - if await btn.count() > 0: - href = await btn.get_attribute("href") - logger.info(f"Enable Semantic Search button href: {href}") - except Exception as e: - logger.warning(f"Could not get button href: {e}") - - # Check for error messages - if "error" in page_content.lower(): - logger.warning("Page contains 'error' keyword") - - finally: - await context.close() diff --git a/tests/server/oauth/test_nc_php_app_oauth.py b/tests/server/oauth/test_nc_php_app_oauth.py deleted file mode 100644 index b4d91d5d..00000000 --- a/tests/server/oauth/test_nc_php_app_oauth.py +++ /dev/null @@ -1,414 +0,0 @@ -"""Test OAuth authorization flow for Nextcloud PHP app (astrolabe). - -Tests the complete PKCE OAuth flow from the NC PHP app perspective: -1. User navigates to personal settings -2. Clicks "Authorize Access" button -3. Completes OAuth authorization via Nextcloud OIDC app -4. Token is stored encrypted in Nextcloud database -5. App can use token to call MCP management API - -This tests the architecture from ADR-018 where the NC PHP app uses -OAuth PKCE (public client) to obtain tokens from Nextcloud's OIDC app. -""" - -import logging -import os - -import httpx -import pytest - -logger = logging.getLogger(__name__) - -pytestmark = [pytest.mark.integration, pytest.mark.oauth] - - -@pytest.fixture(scope="module") -def nextcloud_credentials(): - """Get Nextcloud credentials from environment.""" - return { - "host": os.getenv("NEXTCLOUD_HOST", "http://localhost:8080"), - "username": os.getenv("NEXTCLOUD_USERNAME", "admin"), - "password": os.getenv("NEXTCLOUD_PASSWORD", "admin"), - } - - -@pytest.fixture(scope="module") -async def nc_admin_http_client(nextcloud_credentials): - """HTTP client authenticated as admin user for NC API calls.""" - async with httpx.AsyncClient( - base_url=nextcloud_credentials["host"], - auth=(nextcloud_credentials["username"], nextcloud_credentials["password"]), - timeout=30.0, - ) as client: - yield client - - -@pytest.fixture(scope="module") -async def configure_astrolabe_for_tests(configure_astrolabe_for_mcp_server): - """Configure Astrolabe to connect to mcp-oauth server before running tests. - - This module-scoped fixture ensures Astrolabe is properly configured - for the mcp-oauth server (http://localhost:8001) before any tests run. - """ - logger.info("Configuring Astrolabe for mcp-oauth server...") - await configure_astrolabe_for_mcp_server( - mcp_server_internal_url="http://mcp-oauth:8001", - mcp_server_public_url="http://localhost:8001", - ) - logger.info("✓ Astrolabe configured for mcp-oauth server") - - -@pytest.fixture(scope="module") -async def authorized_nc_session( - browser, nextcloud_credentials, configure_astrolabe_for_tests -): - """Module-scoped fixture that logs in and authorizes the NC PHP app once. - - This fixture: - 1. Configures Astrolabe for mcp-oauth server (via configure_astrolabe_for_tests) - 2. Creates a browser context - 3. Logs in to Nextcloud - 4. Authorizes the MCP Server UI app (if not already authorized) - 5. Returns the page for use in all tests - - The authorization is done once and reused for all tests in this module. - """ - host = nextcloud_credentials["host"] - username = nextcloud_credentials["username"] - password = nextcloud_credentials["password"] - - logger.info("Setting up module-scoped authorized NC session...") - - # Create browser context that persists for module duration - context = await browser.new_context() - page = await context.new_page() - - # Enable console message logging - page.on( - "console", lambda msg: logger.debug(f"Browser console [{msg.type}]: {msg.text}") - ) - page.on("pageerror", lambda err: logger.error(f"Browser page error: {err}")) - - try: - # Step 1: Login to Nextcloud - logger.info(f"Logging in to Nextcloud as {username}...") - await page.goto(f"{host}/login") - - # Fill login form - await page.fill('input[name="user"]', username) - await page.fill('input[name="password"]', password) - await page.click('button[type="submit"]') - - # Wait for login to complete (dashboard loads) - await page.wait_for_url(f"{host}/apps/dashboard/", timeout=10000) - logger.info("✓ Logged in successfully") - - # Step 2: Navigate to personal MCP settings - logger.info("Navigating to personal MCP settings...") - await page.goto(f"{host}/settings/user/astrolabe") - await page.wait_for_load_state("networkidle") - - page_content = await page.content() - - # Step 3: Check if authorization is needed - # Vue 3 UI shows "Enable Semantic Search" when not authorized - if ( - "Enable Semantic Search" in page_content - or "What happens next?" in page_content - ): - logger.info("User not authorized yet - initiating OAuth flow...") - - # Click "Enable Semantic Search" button (Vue 3 template text) - authorize_selectors = [ - 'a:has-text("Enable Semantic Search")', - 'button:has-text("Enable Semantic Search")', - 'a:has-text("Sign In Again")', - "a.button.primary", - '[href*="oauth/login"]', - ] - - clicked = False - for selector in authorize_selectors: - try: - await page.click(selector, timeout=2000) - clicked = True - logger.info(f"✓ Clicked authorize button (selector: {selector})") - break - except Exception: - continue - - if not clicked: - screenshot_path = "/tmp/nc-php-app-settings.png" - await page.screenshot(path=screenshot_path) - pytest.fail( - f"Could not find authorize button. Screenshot: {screenshot_path}" - ) - - # Wait for page to load after clicking - await page.wait_for_load_state("networkidle", timeout=10000) - current_url = page.url - logger.info(f"After clicking authorize, current URL: {current_url}") - - # Take screenshot for debugging - await page.screenshot(path="/tmp/nc-php-app-after-authorize-click.png") - logger.info("Screenshot saved to /tmp/nc-php-app-after-authorize-click.png") - - # Handle OAuth consent if needed - if ( - "/apps/oidc/authorize" in current_url - or "/apps/oidc/consent" in current_url - ): - logger.info("On OIDC authorization page - granting consent...") - - consent_selectors = [ - 'button:has-text("Allow")', - 'button:has-text("Authorize")', - 'input[type="submit"][value="Allow"]', - 'button[type="submit"]', - ] - - for selector in consent_selectors: - try: - await page.click(selector, timeout=2000) - logger.info(f"✓ Clicked consent button (selector: {selector})") - break - except Exception: - continue - - # Wait for redirect back to settings - await page.wait_for_url(f"{host}/settings/user/astrolabe", timeout=15000) - await page.wait_for_load_state("networkidle") - logger.info("✓ OAuth authorization completed") - - else: - logger.info("User already authorized") - - # Return the page and context info for tests - yield { - "page": page, - "context": context, - "host": host, - "username": username, - } - - finally: - # Cleanup at module end - logger.info("Closing authorized NC session...") - await context.close() - - -class TestNcPhpAppOAuth: - """Test suite for NC PHP app OAuth integration.""" - - async def test_authorization_completed(self, authorized_nc_session): - """Verify OAuth authorization was successful. - - This test verifies the settings page shows the user is connected - after the module-scoped authorization fixture runs. - """ - page = authorized_nc_session["page"] - host = authorized_nc_session["host"] - - # Navigate to settings (may already be there) - await page.goto(f"{host}/settings/user/astrolabe") - await page.wait_for_load_state("networkidle") - - page_content = await page.content() - - # Look for indicators that authorization succeeded (Vue 3 personal.php template) - # These must be unique to the authorized state (not found in oauth-required.php) - success_indicators = [ - "Service Status", - "Background Sync Access", - "Manage Connection", - "Revoke Access", - "Service URL", - ] - - found_indicators = [ind for ind in success_indicators if ind in page_content] - has_success_indicator = len(found_indicators) > 0 - - # Always take screenshot for debugging - screenshot_path = "/tmp/nc-php-app-auth-check.png" - await page.screenshot(path=screenshot_path) - logger.info(f"Authorization check screenshot: {screenshot_path}") - logger.info(f"Found success indicators: {found_indicators}") - - if not has_success_indicator: - logger.error("Authorization check failed.") - - assert has_success_indicator, "Settings page should show user is authorized" - logger.info("✓ Authorization verification passed") - - async def test_token_storage_and_retrieval(self, authorized_nc_session): - """Test that tokens are properly stored and can be retrieved. - - Verifies the settings page displays session information, - indicating the token was stored and retrieved successfully. - """ - page = authorized_nc_session["page"] - host = authorized_nc_session["host"] - - await page.goto(f"{host}/settings/user/astrolabe") - await page.wait_for_load_state("networkidle") - - page_content = await page.content() - - # Debug: take screenshot and log content excerpt - screenshot_path = "/tmp/nc-php-app-token-test.png" - await page.screenshot(path=screenshot_path) - logger.info(f"Screenshot saved: {screenshot_path}") - logger.info(f"Page content excerpt: {page_content[:1000]}") - - # Verify session information is visible (Vue 3 personal.php template) - session_indicators = [ - "Service Status", - "Service URL", - "Version", - "Background Sync Access", - ] - - found_indicators = [ind for ind in session_indicators if ind in page_content] - assert len(found_indicators) >= 2, ( - f"Expected session info on page. Found: {found_indicators}. Check {screenshot_path}" - ) - - logger.info(f"✓ Token retrieval verified - found: {found_indicators}") - - async def test_management_api_access( - self, authorized_nc_session, nc_admin_http_client - ): - """Test that the NC PHP app can access MCP server management API. - - Verifies the settings page successfully fetched data from the - MCP server's management API endpoints. - """ - page = authorized_nc_session["page"] - host = authorized_nc_session["host"] - - # Check personal settings page shows server status - await page.goto(f"{host}/settings/user/astrolabe") - await page.wait_for_load_state("networkidle") - - page_content = await page.content() - - # Look for data that comes from management API or template structure (Vue 3) - api_indicators = [ - "Service Status", # Section header - "Service URL", # Server info from API - "Version", # Server version from management API - "Semantic Search", # Vector sync status - ] - - found_api_data = [ind for ind in api_indicators if ind in page_content] - assert len(found_api_data) >= 1, ( - f"Expected management API data on page. Found: {found_api_data}" - ) - - logger.info(f"✓ Management API access verified - found: {found_api_data}") - - async def test_admin_settings_page(self, authorized_nc_session): - """Test that admin settings page loads and displays server info. - - The admin page should show server status from the management API. - """ - page = authorized_nc_session["page"] - host = authorized_nc_session["host"] - - await page.goto(f"{host}/settings/admin/astrolabe") - await page.wait_for_load_state("networkidle") - - page_content = await page.content() - - # Admin page should show server status (Vue 3 AdminSettings.vue) - admin_indicators = [ - "Astrolabe", - "Service Status", - "Version", - "Semantic Search", - ] - - found_indicators = [ind for ind in admin_indicators if ind in page_content] - - # Admin page should at least show the Astrolabe header or Service Status - assert "Astrolabe" in page_content or "Service Status" in page_content, ( - "Admin settings page should show Astrolabe section" - ) - - logger.info(f"✓ Admin settings page verified - found: {found_indicators}") - - -class TestNcPhpAppDisconnect: - """Test suite for NC PHP app disconnect functionality. - - Note: These tests are run separately and may modify the authorization state. - They should run after the main OAuth tests. - """ - - @pytest.mark.skip(reason="Disconnect test modifies state - run manually if needed") - async def test_disconnect_flow(self, browser, nextcloud_credentials): - """Test that users can disconnect (revoke) their authorization. - - This test: - 1. Logs in fresh (separate from authorized_nc_session) - 2. Verifies user is authorized - 3. Clicks "Disconnect" button - 4. Verifies user is no longer authorized - - Skipped by default as it modifies authorization state. - """ - host = nextcloud_credentials["host"] - username = nextcloud_credentials["username"] - password = nextcloud_credentials["password"] - - context = await browser.new_context() - page = await context.new_page() - - try: - # Login - await page.goto(f"{host}/login") - await page.fill('input[name="user"]', username) - await page.fill('input[name="password"]', password) - await page.click('button[type="submit"]') - await page.wait_for_url(f"{host}/apps/dashboard/", timeout=10000) - - # Navigate to personal settings - await page.goto(f"{host}/settings/user/astrolabe") - await page.wait_for_load_state("networkidle") - - page_content = await page.content() - - # Check if user is authorized (Vue 3 personal.php shows Disconnect/Revoke when authorized) - if "Disconnect" not in page_content and "Revoke Access" not in page_content: - pytest.skip("User not authorized - cannot test disconnect") - - # Click disconnect button - disconnect_selectors = [ - 'button:has-text("Disconnect")', - 'form[action*="disconnect"] button', - "#mcp-disconnect-button", - ] - - for selector in disconnect_selectors: - try: - # Handle confirmation dialog - page.on("dialog", lambda dialog: dialog.accept()) - await page.click(selector, timeout=2000) - logger.info(f"✓ Clicked disconnect button (selector: {selector})") - break - except Exception: - continue - - # Wait for page reload - await page.wait_for_load_state("networkidle") - - # Verify we're back to "Enable Semantic Search" state (Vue 3 oauth-required.php) - page_content = await page.content() - assert "Enable Semantic Search" in page_content, ( - "Settings page should show 'Enable Semantic Search' after disconnect" - ) - - logger.info("✓ Disconnect flow test passed") - - finally: - await context.close() diff --git a/tests/server/oauth/test_oauth_core.py b/tests/server/oauth/test_oauth_core.py deleted file mode 100644 index 7364cc7c..00000000 --- a/tests/server/oauth/test_oauth_core.py +++ /dev/null @@ -1,262 +0,0 @@ -"""Core OAuth integration tests. - -Consolidated from: -- test_mcp_oauth.py: Basic OAuth connectivity -- test_mcp_oauth_jwt.py: JWT-specific operations -- test_jwt_tokens.py: JWT token structure validation - -Tests verify: -1. OAuth server connectivity and tool listing -2. Tool execution with OAuth tokens -3. JWT token structure and claims -4. Multiple operations with same token (persistence) -5. Error handling with OAuth -""" - -import base64 -import json -import logging - -import pytest - -logger = logging.getLogger(__name__) - -pytestmark = [pytest.mark.integration, pytest.mark.oauth] - - -def decode_jwt_without_verification(token: str) -> dict: - """Decode JWT token without signature verification (for inspection only). - - Returns: - Dict with header and payload - """ - parts = token.split(".") - if len(parts) != 3: - raise ValueError(f"Invalid JWT format: expected 3 parts, got {len(parts)}") - - # Decode header - header = json.loads( - base64.urlsafe_b64decode(parts[0] + "=" * (4 - len(parts[0]) % 4)) - ) - - # Decode payload - payload = json.loads( - base64.urlsafe_b64decode(parts[1] + "=" * (4 - len(parts[1]) % 4)) - ) - - return { - "header": header, - "payload": payload, - } - - -# ============================================================================ -# Basic OAuth Connectivity Tests -# ============================================================================ - - -async def test_mcp_oauth_server_connection(nc_mcp_oauth_client): - """Test connection to OAuth-enabled MCP server.""" - result = await nc_mcp_oauth_client.list_tools() - assert result is not None - assert len(result.tools) > 0 - - logger.info(f"OAuth MCP server has {len(result.tools)} tools available") - - -async def test_mcp_oauth_tool_execution(nc_mcp_oauth_client): - """Test executing a tool on the OAuth-enabled MCP server.""" - # Example: Execute the 'nc_notes_search_notes' tool - result = await nc_mcp_oauth_client.call_tool( - "nc_notes_search_notes", arguments={"query": ""} - ) - - assert result.isError is False, f"Tool execution failed: {result.content}" - assert result.content is not None - response_data = json.loads(result.content[0].text) - - # The search response should have a 'results' field containing the list - assert "results" in response_data - assert isinstance(response_data["results"], list) - - logger.info( - f"Successfully executed 'nc_notes_search_notes' tool on OAuth MCP server and got {len(response_data['results'])} notes." - ) - - -async def test_mcp_oauth_client_with_playwright(nc_mcp_oauth_client): - """Test that MCP OAuth client via Playwright can execute tools.""" - # Test: Execute the 'nc_notes_search_notes' tool - result = await nc_mcp_oauth_client.call_tool( - "nc_notes_search_notes", arguments={"query": ""} - ) - - assert result.isError is False, f"Tool execution failed: {result.content}" - assert result.content is not None - response_data = json.loads(result.content[0].text) - - # The search response should have a 'results' field containing the list - assert "results" in response_data - assert isinstance(response_data["results"], list) - - logger.info( - f"Successfully executed 'nc_notes_search_notes' tool on Playwright OAuth MCP server and got {len(response_data['results'])} notes." - ) - - -# ============================================================================ -# JWT-Specific Tests -# ============================================================================ - - -async def test_jwt_tool_list_operations(nc_mcp_oauth_jwt_client): - """Test that list_tools works with JWT authentication and returns expected tools. - - This test verifies that tools are properly filtered based on per-app scopes: - - notes:read/write → Notes app tools - - calendar:read/write → Calendar app tools - - files:read/write → WebDAV/Files app tools - - etc. - """ - result = await nc_mcp_oauth_jwt_client.list_tools() - - # Verify we have tools - assert len(result.tools) > 0 - - # Verify expected tools exist based on configured scopes - tool_names = [tool.name for tool in result.tools] - - # Notes tools (require notes:read and notes:write) - assert "nc_notes_get_note" in tool_names, "Missing nc_notes_get_note (notes:read)" - assert "nc_notes_create_note" in tool_names, ( - "Missing nc_notes_create_note (notes:write)" - ) - - # Calendar tools (require calendar:read and calendar:write) - assert "nc_calendar_list_calendars" in tool_names, ( - "Missing nc_calendar_list_calendars (calendar:read)" - ) - assert "nc_calendar_create_event" in tool_names, ( - "Missing nc_calendar_create_event (calendar:write)" - ) - - # Verify we have a reasonable number of tools for the configured scopes - # With notes + calendar scopes, expect ~20-30 tools - assert len(tool_names) >= 20, ( - f"Expected at least 20 tools with notes+calendar scopes, got {len(tool_names)}" - ) - - logger.info( - f"JWT OAuth server provides {len(result.tools)} tools with configured per-app scopes" - ) - - -async def test_jwt_multiple_operations(nc_mcp_oauth_jwt_client): - """Test multiple operations with same JWT token to verify token persistence. - - JWT tokens should work across multiple tool calls without re-authentication, - demonstrating that the token is properly cached and reused. - """ - # First operation: Search notes - result1 = await nc_mcp_oauth_jwt_client.call_tool( - "nc_notes_search_notes", arguments={"query": ""} - ) - assert result1.isError is False - - # Second operation: List calendars - result2 = await nc_mcp_oauth_jwt_client.call_tool( - "nc_calendar_list_calendars", arguments={} - ) - assert result2.isError is False - - # Third operation: List directory - result3 = await nc_mcp_oauth_jwt_client.call_tool( - "nc_webdav_list_directory", arguments={"path": "/"} - ) - assert result3.isError is False - - logger.info( - "Successfully executed 3 different operations with same JWT token (token persistence verified)" - ) - - -async def test_jwt_error_handling(nc_mcp_oauth_jwt_client): - """Test error handling with JWT authentication. - - Verifies that invalid operations return proper errors even with valid JWT tokens. - """ - # Try to get a non-existent note - result = await nc_mcp_oauth_jwt_client.call_tool( - "nc_notes_get_note", arguments={"note_id": 999999} - ) - - # Should get an error (note doesn't exist) - assert result.isError is True - logger.info("JWT OAuth server correctly handles errors for invalid operations") - - -# ============================================================================ -# JWT Token Structure Tests -# ============================================================================ - - -async def test_jwt_tokens_embed_scopes_in_payload(): - """Document that JWT tokens embed scopes in the payload (RFC 9068). - - This test documents expected JWT structure based on manual testing. - """ - expected_structure = { - "header": { - "typ": "at+JWT", # RFC 9068 access token type - "alg": "RS256", # Signature algorithm - }, - "payload_claims": { - "iss": "issuer URL", - "sub": "user ID", - "aud": "client ID", - "exp": "expiration timestamp", - "iat": "issued at timestamp", - "scope": "space-separated scope string (e.g., 'notes:read notes:write')", - "client_id": "client identifier", - "jti": "JWT ID", - }, - "scope_claim": { - "format": "space-separated string", - "example": "openid profile email notes:read notes:write", - "extraction": "payload['scope'].split()", - }, - } - - logger.info("JWT token structure (RFC 9068):") - logger.info(json.dumps(expected_structure, indent=2)) - - # This test documents expected behavior - assert True - - -async def test_opaque_token_vs_jwt_comparison(): - """Document differences between opaque tokens and JWT tokens. - - This test captures our findings about the two token types. - """ - findings = { - "jwt_advantages": [ - "Scopes embedded in payload - no introspection needed", - "Self-contained - can validate with JWKS", - "Standard approach (RFC 9068)", - ], - "jwt_disadvantages": [ - "10-15x larger than opaque tokens (~800-1200 chars vs 72)", - "Cannot be easily revoked (until expiration)", - ], - "token_sizes": { - "opaque": "72 characters", - "jwt": "~800-1200 characters", - }, - "recommendation": "Use JWT for MCP server (scopes available without introspection)", - } - - logger.info("JWT vs Opaque token comparison:") - logger.info(json.dumps(findings, indent=2)) - - assert True diff --git a/tests/server/oauth/test_oauth_deck_permissions.py b/tests/server/oauth/test_oauth_deck_permissions.py deleted file mode 100644 index ee444aaf..00000000 --- a/tests/server/oauth/test_oauth_deck_permissions.py +++ /dev/null @@ -1,350 +0,0 @@ -""" -Multi-user OAuth tests for Nextcloud Deck board permissions. - -Tests verify that the MCP server respects Nextcloud Deck board ACL permissions -when accessed via OAuth authentication with different users. -""" - -import json -import logging - -import pytest - -logger = logging.getLogger(__name__) - -pytestmark = [pytest.mark.integration, pytest.mark.oauth] - - -async def add_board_acl(nc_client, board_id: int, user: str, permission_type: int = 0): - """ - Helper to add ACL entry to a Deck board. - - Args: - nc_client: Admin NextcloudClient - board_id: Board ID - user: Username to grant access - permission_type: 0=view, 1=edit, 2=manage - - Returns: - ACL entry ID - """ - acl = await nc_client.deck.add_acl_rule( - board_id=board_id, - type=0, # 0 = user, 1 = group - participant=user, - permission_edit=permission_type >= 1, - permission_share=permission_type >= 2, - permission_manage=permission_type >= 2, - ) - logger.info(f"Added ACL for board {board_id}: {user} (type={permission_type})") - return acl.id - - -async def delete_board_acl(nc_client, board_id: int, acl_id: int): - """Helper to delete a board ACL entry.""" - await nc_client.deck.delete_acl_rule(board_id, acl_id) - logger.info(f"Deleted ACL {acl_id} from board {board_id}") - - -async def test_deck_board_view_permissions( - nc_client, alice_mcp_client, bob_mcp_client, diana_mcp_client -): - """ - Test that Deck boards respect view permissions. - - Scenario: - 1. Admin creates a board as alice - 2. Admin adds bob to board with view-only permissions - 3. Bob can view the board via MCP tools - 4. Diana cannot access the board (no ACL entry) - """ - # Create a board as alice - logger.info("Creating Deck board as alice...") - board = await nc_client.deck.create_board( - "Alice's Shared Board - View Test", "FF0000" - ) - board_id = board.id - - bob_acl_id = None - - try: - # Add bob to board with view-only permission - logger.info("Adding bob to board with view permission...") - bob_acl_id = await add_board_acl(nc_client, board_id, "bob", permission_type=0) - - # Test: Bob can view the board via MCP - logger.info("Bob attempting to list boards via MCP...") - result = await bob_mcp_client.call_tool("deck_get_boards", arguments={}) - - if not result.isError: - response_data = json.loads(result.content[0].text) - # Response is a ListBoardsResponse with a "boards" field - board_list = response_data.get("boards", []) - board_ids = [b["id"] for b in board_list] - logger.info(f"Bob can see {len(board_list)} boards: {board_ids}") - - # Bob should see the shared board - if board_id in board_ids: - logger.info(f"Bob can see shared board {board_id}") - else: - logger.warning(f"Bob cannot see shared board {board_id}") - else: - logger.warning(f"Bob could not list boards: {result.content}") - - # Test: Diana cannot see the board - logger.info("Diana attempting to list boards via MCP...") - result = await diana_mcp_client.call_tool("deck_get_boards", arguments={}) - - if not result.isError: - response_data = json.loads(result.content[0].text) - # Response is a ListBoardsResponse with a "boards" field - board_list = response_data.get("boards", []) - board_ids = [b["id"] for b in board_list] - logger.info(f"Diana can see {len(board_list)} boards") - - # Diana should NOT see the board - assert board_id not in board_ids, "Diana should not see board without ACL" - logger.info("Diana correctly cannot see board without ACL") - else: - logger.warning(f"Diana could not list boards: {result.content}") - - finally: - # Cleanup - if bob_acl_id: - await delete_board_acl(nc_client, board_id, bob_acl_id) - logger.info(f"Deleting board {board_id}") - await nc_client.deck.delete_board(board_id) - - -async def test_deck_board_edit_permissions( - nc_client, alice_mcp_client, charlie_mcp_client, bob_mcp_client -): - """ - Test that Deck boards respect edit permissions. - - Scenario: - 1. Admin creates a board as alice with a stack - 2. Admin adds charlie with edit permission - 3. Admin adds bob with view-only permission - 4. Charlie can create cards via MCP tools - 5. Bob cannot create cards - """ - # Create a board as alice - logger.info("Creating Deck board as alice...") - board = await nc_client.deck.create_board( - "Alice's Shared Board - Edit Test", "00FF00" - ) - board_id = board.id - - # Create a stack in the board - logger.info("Creating stack in board...") - stack = await nc_client.deck.create_stack(board_id, "Test Stack", 1) - stack_id = stack.id - - charlie_acl_id = None - bob_acl_id = None - - try: - # Add charlie with edit permission - logger.info("Adding charlie to board with edit permission...") - charlie_acl_id = await add_board_acl( - nc_client, board_id, "charlie", permission_type=1 - ) - - # Add bob with view-only permission - logger.info("Adding bob to board with view permission...") - bob_acl_id = await add_board_acl(nc_client, board_id, "bob", permission_type=0) - - # Test: Charlie can create a card - logger.info("Charlie attempting to create card via MCP...") - result = await charlie_mcp_client.call_tool( - "deck_create_card", - arguments={ - "board_id": board_id, - "stack_id": stack_id, - "title": "Charlie's Card", - "description": "Created by Charlie with edit permission", - }, - ) - - if not result.isError: - response_data = json.loads(result.content[0].text) - card_id = response_data.get("id") - logger.info(f"Charlie successfully created card {card_id}") - - # Cleanup the card - await nc_client.deck.delete_card(board_id, stack_id, card_id) - else: - logger.warning(f"Charlie could not create card: {result.content}") - - # Test: Bob attempts to create a card (should fail) - logger.info("Bob attempting to create card via MCP...") - result = await bob_mcp_client.call_tool( - "deck_create_card", - arguments={ - "board_id": board_id, - "stack_id": stack_id, - "title": "Bob's Card", - "description": "Bob trying to create a card", - }, - ) - - if result.isError: - logger.info("Bob correctly denied card creation (view-only)") - else: - logger.warning("Bob unexpectedly succeeded in creating card") - # Cleanup if bob somehow created a card - response_data = json.loads(result.content[0].text) - if "id" in response_data: - await nc_client.deck.delete_card( - board_id, stack_id, response_data["id"] - ) - - finally: - # Cleanup - if charlie_acl_id: - await delete_board_acl(nc_client, board_id, charlie_acl_id) - if bob_acl_id: - await delete_board_acl(nc_client, board_id, bob_acl_id) - logger.info(f"Deleting board {board_id}") - await nc_client.deck.delete_board(board_id) - - -async def test_deck_board_manage_permissions( - nc_client, alice_mcp_client, charlie_mcp_client -): - """ - Test that Deck boards respect manage permissions. - - Scenario: - 1. Admin creates a board as alice - 2. Admin adds charlie with manage permission - 3. Charlie can create stacks and modify board settings - """ - # Create a board as alice - logger.info("Creating Deck board as alice...") - board = await nc_client.deck.create_board( - "Alice's Shared Board - Manage Test", "0000FF" - ) - board_id = board.id - - charlie_acl_id = None - - try: - # Add charlie with manage permission - logger.info("Adding charlie to board with manage permission...") - charlie_acl_id = await add_board_acl( - nc_client, board_id, "charlie", permission_type=2 - ) - - # Test: Charlie can create a stack - logger.info("Charlie attempting to create stack via MCP...") - result = await charlie_mcp_client.call_tool( - "deck_create_stack", - arguments={"board_id": board_id, "title": "Charlie's Stack", "order": 1}, - ) - - if not result.isError: - response_data = json.loads(result.content[0].text) - stack_id = response_data.get("id") - logger.info(f"Charlie successfully created stack {stack_id}") - - # Cleanup the stack - await nc_client.deck.delete_stack(board_id, stack_id) - else: - logger.warning(f"Charlie could not create stack: {result.content}") - - # Test: Charlie can delete a stack (manage permission) - logger.info("Charlie attempting to delete stack via MCP...") - # First create a temporary stack to delete - temp_stack = await nc_client.deck.create_stack( - board_id, "Temp Stack for Deletion", 99 - ) - - result = await charlie_mcp_client.call_tool( - "deck_delete_stack", - arguments={"board_id": board_id, "stack_id": temp_stack.id}, - ) - - if not result.isError: - logger.info("Charlie successfully deleted stack") - else: - logger.warning(f"Charlie could not delete stack: {result.content}") - # Cleanup if deletion via MCP failed - try: - await nc_client.deck.delete_stack(board_id, temp_stack.id) - except Exception: - pass - - finally: - # Cleanup - if charlie_acl_id: - await delete_board_acl(nc_client, board_id, charlie_acl_id) - logger.info(f"Deleting board {board_id}") - await nc_client.deck.delete_board(board_id) - - -async def test_deck_user_isolation(nc_client, alice_mcp_client, bob_mcp_client): - """ - Test that users can only see their own boards when not shared. - - Scenario: - 1. Admin creates a board as alice (not shared) - 2. Admin creates a board as bob (not shared) - 3. Alice can only see her own board - 4. Bob can only see his own board - """ - # Create alice's board - logger.info("Creating alice's private board...") - alice_board = await nc_client.deck.create_board("Alice's Private Board", "FF00FF") - alice_board_id = alice_board.id - - # Create bob's board - logger.info("Creating bob's private board...") - bob_board = await nc_client.deck.create_board("Bob's Private Board", "00FFFF") - bob_board_id = bob_board.id - - try: - # Test: Alice lists boards - logger.info("Alice listing boards via MCP...") - result = await alice_mcp_client.call_tool("deck_get_boards", arguments={}) - - if not result.isError: - response_data = json.loads(result.content[0].text) - # Response is a ListBoardsResponse with a "boards" field - board_list = response_data.get("boards", []) - board_ids = [b["id"] for b in board_list] - logger.info(f"Alice can see boards: {board_ids}") - - # Alice should NOT see Bob's board - assert bob_board_id not in board_ids, ( - "Alice should not see Bob's private board" - ) - else: - logger.warning(f"Alice could not list boards: {result.content}") - - # Test: Bob lists boards - logger.info("Bob listing boards via MCP...") - result = await bob_mcp_client.call_tool("deck_get_boards", arguments={}) - - if not result.isError: - response_data = json.loads(result.content[0].text) - # Response is a ListBoardsResponse with a "boards" field - board_list = response_data.get("boards", []) - board_ids = [b["id"] for b in board_list] - logger.info(f"Bob can see boards: {board_ids}") - - # Bob should NOT see Alice's board - assert alice_board_id not in board_ids, ( - "Bob should not see Alice's private board" - ) - else: - logger.warning(f"Bob could not list boards: {result.content}") - - logger.info("User isolation test passed: users can only see their own boards") - - finally: - # Cleanup - logger.info("Cleaning up test boards...") - await nc_client.deck.delete_board(alice_board_id) - await nc_client.deck.delete_board(bob_board_id) diff --git a/tests/server/oauth/test_oauth_file_permissions.py b/tests/server/oauth/test_oauth_file_permissions.py deleted file mode 100644 index 1254ee1d..00000000 --- a/tests/server/oauth/test_oauth_file_permissions.py +++ /dev/null @@ -1,421 +0,0 @@ -""" -Multi-user OAuth tests for Nextcloud WebDAV file permissions. - -Tests verify that the MCP server respects Nextcloud file sharing permissions -when accessed via OAuth authentication with different users. - -All operations (file creation, sharing, access) are performed through MCP tools -to ensure the MCP server properly supports multi-user scenarios. -""" - -import json -import logging - -import pytest - -logger = logging.getLogger(__name__) - -pytestmark = [pytest.mark.integration, pytest.mark.oauth] - - -async def test_file_share_read_permissions( - alice_mcp_client, bob_mcp_client, diana_mcp_client -): - """ - Test that shared files respect read permissions. - - Scenario: - 1. Alice creates a file via MCP - 2. Alice shares the file with Bob (read-only) via MCP - 3. Bob can read the file via MCP tools - 4. Diana cannot access the file (no share) - """ - file_path = "/alice_shared_file_read.txt" - file_content = "This file is shared with Bob for reading only." - - # Alice creates a file - logger.info(f"Alice creating file: {file_path}") - result = await alice_mcp_client.call_tool( - "nc_webdav_write_file", - arguments={"path": file_path, "content": file_content}, - ) - assert not result.isError, f"Alice failed to create file: {result.content}" - - share_id = None - - try: - # Alice shares the file with bob (read-only, permissions=1) - logger.info("Alice sharing file with bob (read-only)...") - result = await alice_mcp_client.call_tool( - "nc_share_create", - arguments={ - "path": file_path, - "share_with": "bob", - "share_type": 0, - "permissions": 1, - }, - ) - assert not result.isError, f"Alice failed to create share: {result.content}" - share_data = json.loads(result.content[0].text) - share_id = share_data["id"] - logger.info(f"Created share {share_id}") - - # Test: Bob reads the file via MCP - logger.info("Bob attempting to read file via MCP...") - result = await bob_mcp_client.call_tool( - "nc_webdav_read_file", arguments={"path": file_path} - ) - - # Bob should be able to read the shared file - if not result.isError: - response_data = json.loads(result.content[0].text) - logger.info( - f"Bob successfully read file: {response_data.get('content', '')[:50]}..." - ) - assert "content" in response_data - assert file_content in response_data["content"] - else: - logger.warning(f"Bob could not read file: {result.content}") - # This might fail if the share path is different for bob - - # Test: Diana attempts to read the file - logger.info("Diana attempting to read file via MCP...") - result = await diana_mcp_client.call_tool( - "nc_webdav_read_file", arguments={"path": file_path} - ) - - # Diana should NOT be able to read (no share) - if result.isError: - logger.info("Diana correctly denied access to unshared file") - else: - logger.warning("Diana unexpectedly could read unshared file") - - finally: - # Cleanup - Alice deletes the share and file - if share_id: - logger.info(f"Alice deleting share {share_id}") - await alice_mcp_client.call_tool( - "nc_share_delete", arguments={"share_id": share_id} - ) - logger.info(f"Alice deleting file {file_path}") - await alice_mcp_client.call_tool( - "nc_webdav_delete_resource", arguments={"path": file_path} - ) - - -async def test_file_share_write_permissions( - alice_mcp_client, charlie_mcp_client, bob_mcp_client -): - """ - Test that shared files respect write permissions. - - Scenario: - 1. Alice creates a file via MCP - 2. Alice shares the file with Charlie (edit permission) via MCP - 3. Alice shares the file with Bob (read-only) via MCP - 4. Charlie can edit the file via MCP tools - 5. Bob cannot edit the file - """ - file_path = "/alice_shared_file_write.txt" - file_content = "This file is shared with Charlie for editing." - - logger.info(f"Alice creating file: {file_path}") - result = await alice_mcp_client.call_tool( - "nc_webdav_write_file", - arguments={"path": file_path, "content": file_content}, - ) - assert not result.isError, f"Alice failed to create file: {result.content}" - - charlie_share_id = None - bob_share_id = None - - try: - # Alice shares with Charlie (read+write, permissions=3) - logger.info("Alice sharing file with Charlie (edit permission)...") - result = await alice_mcp_client.call_tool( - "nc_share_create", - arguments={ - "path": file_path, - "share_with": "charlie", - "share_type": 0, - "permissions": 3, - }, - ) - assert not result.isError, ( - f"Alice failed to share with Charlie: {result.content}" - ) - charlie_share_data = json.loads(result.content[0].text) - charlie_share_id = charlie_share_data["id"] - logger.info(f"Created share {charlie_share_id} for Charlie") - - # Alice shares with Bob (read-only, permissions=1) - logger.info("Alice sharing file with Bob (read-only)...") - result = await alice_mcp_client.call_tool( - "nc_share_create", - arguments={ - "path": file_path, - "share_with": "bob", - "share_type": 0, - "permissions": 1, - }, - ) - assert not result.isError, f"Alice failed to share with Bob: {result.content}" - bob_share_data = json.loads(result.content[0].text) - bob_share_id = bob_share_data["id"] - logger.info(f"Created share {bob_share_id} for Bob") - - # Test: Charlie can write to the file - logger.info("Charlie attempting to write to file via MCP...") - updated_content = f"{file_content}\nCharlie added this line." - result = await charlie_mcp_client.call_tool( - "nc_webdav_write_file", - arguments={"path": file_path, "content": updated_content}, - ) - - if not result.isError: - logger.info("Charlie successfully wrote to file") - else: - logger.warning(f"Charlie could not write to file: {result.content}") - - # Test: Bob attempts to write (should fail) - logger.info("Bob attempting to write to file via MCP...") - result = await bob_mcp_client.call_tool( - "nc_webdav_write_file", - arguments={"path": file_path, "content": "Bob tries to overwrite this."}, - ) - - # Bob should be denied - if result.isError: - logger.info("Bob correctly denied write access") - else: - logger.warning("Bob unexpectedly succeeded in writing (permissions issue?)") - - finally: - # Cleanup - Alice deletes shares and file - if charlie_share_id: - logger.info(f"Alice deleting Charlie's share {charlie_share_id}") - await alice_mcp_client.call_tool( - "nc_share_delete", arguments={"share_id": charlie_share_id} - ) - if bob_share_id: - logger.info(f"Alice deleting Bob's share {bob_share_id}") - await alice_mcp_client.call_tool( - "nc_share_delete", arguments={"share_id": bob_share_id} - ) - logger.info(f"Alice deleting file {file_path}") - await alice_mcp_client.call_tool( - "nc_webdav_delete_resource", arguments={"path": file_path} - ) - - -async def test_file_list_permissions(alice_mcp_client, bob_mcp_client): - """ - Test that file listing respects share permissions. - - Scenario: - 1. Alice creates her private file via MCP - 2. Bob creates his private file via MCP - 3. Alice creates a file and shares it with Bob via MCP - 4. Alice can list her own files + shared files - 5. Bob can list his own files + shared files from Alice - """ - alice_file = "/alice_private_file.txt" - bob_file = "/bob_private_file.txt" - shared_file = "/alice_shared_with_bob.txt" - - # Alice creates her private file - logger.info(f"Alice creating private file: {alice_file}") - result = await alice_mcp_client.call_tool( - "nc_webdav_write_file", - arguments={"path": alice_file, "content": "Alice's private file"}, - ) - assert not result.isError, f"Alice failed to create file: {result.content}" - - # Bob creates his private file - logger.info(f"Bob creating private file: {bob_file}") - result = await bob_mcp_client.call_tool( - "nc_webdav_write_file", - arguments={"path": bob_file, "content": "Bob's private file"}, - ) - assert not result.isError, f"Bob failed to create file: {result.content}" - - # Alice creates a shared file - logger.info(f"Alice creating shared file: {shared_file}") - result = await alice_mcp_client.call_tool( - "nc_webdav_write_file", - arguments={"path": shared_file, "content": "Shared file content"}, - ) - assert not result.isError, f"Alice failed to create shared file: {result.content}" - - share_id = None - - try: - # Alice shares the file with Bob - logger.info("Alice sharing file with Bob...") - result = await alice_mcp_client.call_tool( - "nc_share_create", - arguments={ - "path": shared_file, - "share_with": "bob", - "share_type": 0, - "permissions": 1, - }, - ) - assert not result.isError, f"Alice failed to create share: {result.content}" - share_data = json.loads(result.content[0].text) - share_id = share_data["id"] - - # Test: Alice lists files in root - logger.info("Alice listing files via MCP...") - result = await alice_mcp_client.call_tool( - "nc_webdav_list_directory", arguments={"path": "/"} - ) - - if not result.isError: - response_data = json.loads(result.content[0].text) - # Extract files from DirectoryListing response - files = response_data.get("files", []) - file_names = [f["name"] for f in files] - logger.info(f"Alice can see files: {file_names}") - - # Alice should see her own files - # Note: Exact assertions depend on test isolation - else: - logger.warning(f"Alice could not list files: {result.content}") - - # Test: Bob lists files in root - logger.info("Bob listing files via MCP...") - result = await bob_mcp_client.call_tool( - "nc_webdav_list_directory", arguments={"path": "/"} - ) - - if not result.isError: - response_data = json.loads(result.content[0].text) - # Extract files from DirectoryListing response - files = response_data.get("files", []) - file_names = [f["name"] for f in files] - logger.info(f"Bob can see files: {file_names}") - - # Bob should see his own file, but not Alice's private file - # Bob may see shared files in his shared folder or via different path - else: - logger.warning(f"Bob could not list files: {result.content}") - - finally: - # Cleanup - if share_id: - logger.info(f"Alice deleting share {share_id}") - await alice_mcp_client.call_tool( - "nc_share_delete", arguments={"share_id": share_id} - ) - - logger.info("Cleaning up Alice's files...") - await alice_mcp_client.call_tool( - "nc_webdav_delete_resource", arguments={"path": alice_file} - ) - await alice_mcp_client.call_tool( - "nc_webdav_delete_resource", arguments={"path": shared_file} - ) - - logger.info("Cleaning up Bob's files...") - await bob_mcp_client.call_tool( - "nc_webdav_delete_resource", arguments={"path": bob_file} - ) - - -async def test_folder_share_permissions(alice_mcp_client, bob_mcp_client): - """ - Test that folder sharing works correctly. - - Scenario: - 1. Alice creates a folder via MCP - 2. Alice creates files in the folder via MCP - 3. Alice shares the folder with Bob via MCP - 4. Bob can access files in the shared folder via MCP - """ - folder_path = "/alice_shared_folder" - file_in_folder = f"{folder_path}/document.txt" - file_content = "This is a document in Alice's shared folder" - - # Alice creates folder - logger.info(f"Alice creating folder: {folder_path}") - result = await alice_mcp_client.call_tool( - "nc_webdav_create_directory", arguments={"path": folder_path} - ) - assert not result.isError, f"Alice failed to create folder: {result.content}" - - # Alice creates file in folder - logger.info(f"Alice creating file in folder: {file_in_folder}") - result = await alice_mcp_client.call_tool( - "nc_webdav_write_file", - arguments={"path": file_in_folder, "content": file_content}, - ) - assert not result.isError, f"Alice failed to create file: {result.content}" - - share_id = None - - try: - # Alice shares the folder with Bob - logger.info("Alice sharing folder with Bob...") - result = await alice_mcp_client.call_tool( - "nc_share_create", - arguments={ - "path": folder_path, - "share_with": "bob", - "share_type": 0, - "permissions": 1, - }, - ) - assert not result.isError, f"Alice failed to create share: {result.content}" - share_data = json.loads(result.content[0].text) - share_id = share_data["id"] - logger.info(f"Created folder share {share_id}") - - # Test: Bob lists the shared folder - logger.info("Bob attempting to list shared folder via MCP...") - result = await bob_mcp_client.call_tool( - "nc_webdav_list_directory", arguments={"path": folder_path} - ) - - if not result.isError: - response_data = json.loads(result.content[0].text) - # Extract files from DirectoryListing response - files = response_data.get("files", []) - logger.info(f"Bob can see {len(files)} files in shared folder") - - # Bob should see the file in the shared folder - file_names = [f["name"] for f in files] - assert "document.txt" in file_names, ( - "Bob should see the file in shared folder" - ) - else: - logger.warning(f"Bob could not list shared folder: {result.content}") - - # Test: Bob reads the file in the shared folder - logger.info("Bob attempting to read file in shared folder via MCP...") - result = await bob_mcp_client.call_tool( - "nc_webdav_read_file", arguments={"path": file_in_folder} - ) - - if not result.isError: - response_data = json.loads(result.content[0].text) - logger.info("Bob successfully read file in shared folder") - assert "content" in response_data - assert file_content in response_data["content"] - else: - logger.warning( - f"Bob could not read file in shared folder: {result.content}" - ) - - finally: - # Cleanup - Alice deletes the share and folder - if share_id: - logger.info(f"Alice deleting share {share_id}") - await alice_mcp_client.call_tool( - "nc_share_delete", arguments={"share_id": share_id} - ) - - logger.info("Alice cleaning up test folder...") - await alice_mcp_client.call_tool( - "nc_webdav_delete_resource", arguments={"path": folder_path} - ) diff --git a/tests/server/oauth/test_oauth_notes_permissions.py b/tests/server/oauth/test_oauth_notes_permissions.py deleted file mode 100644 index 524a5d2a..00000000 --- a/tests/server/oauth/test_oauth_notes_permissions.py +++ /dev/null @@ -1,256 +0,0 @@ -""" -Multi-user OAuth tests for Nextcloud Notes permissions. - -Tests verify that the MCP server respects Nextcloud Notes sharing permissions -when accessed via OAuth authentication with different users. -""" - -import json -import logging - -import pytest - -logger = logging.getLogger(__name__) - -pytestmark = [pytest.mark.integration, pytest.mark.oauth] - - -async def test_notes_share_read_permissions( - nc_client, alice_mcp_client, bob_mcp_client, diana_mcp_client -): - """ - Test that shared notes respect read permissions. - - Scenario: - 1. Admin creates a note as alice - 2. Admin shares the note with bob (read-only) - 3. Bob can read the note via MCP tools - 4. Diana cannot access the note (no share) - """ - # Create a note as alice (using admin client to set up data) - note_title = "Alice's Shared Note - Read Test" - note_content = "This note is shared with Bob for reading only." - note_category = "SharedNotes" - - logger.info("Creating note as alice...") - created_note = await nc_client.notes.create_note( - title=note_title, content=note_content, category=note_category - ) - note_id = created_note.get("id") - - try: - # TODO: Share the note with bob (read-only) - # Note: Nextcloud Notes API doesn't have direct sharing endpoints - # Sharing is typically done at the folder level via WebDAV - # For now, this test documents the expected behavior - - # Test: Bob searches for notes via MCP - logger.info("Bob searching for notes via MCP...") - result = await bob_mcp_client.call_tool( - "nc_notes_search_notes", arguments={"query": "Alice's Shared"} - ) - - assert result.isError is False, f"Bob's search failed: {result.content}" - response_data = json.loads(result.content[0].text) - - # Bob should see the shared note in search results - # (assuming proper share setup) - assert "results" in response_data - logger.info(f"Bob found {len(response_data['results'])} notes") - - # Test: Diana searches for the same note - logger.info("Diana searching for notes via MCP...") - result = await diana_mcp_client.call_tool( - "nc_notes_search_notes", arguments={"query": "Alice's Shared"} - ) - - assert result.isError is False - response_data = json.loads(result.content[0].text) - - # Diana should NOT see the note (no share) - assert "results" in response_data - shared_note_ids = [ - n["id"] for n in response_data["results"] if n["id"] == note_id - ] - assert len(shared_note_ids) == 0, "Diana should not see unshared note" - logger.info("Diana correctly cannot see unshared note") - - finally: - # Cleanup - logger.info(f"Cleaning up note {note_id}") - await nc_client.notes.delete_note(note_id) - - -async def test_notes_share_write_permissions( - nc_client, alice_mcp_client, charlie_mcp_client, bob_mcp_client -): - """ - Test that shared notes respect write permissions. - - Scenario: - 1. Admin creates a note as alice - 2. Admin shares the note with charlie (edit permission) - 3. Admin shares the note with bob (read-only) - 4. Charlie can edit the note via MCP tools - 5. Bob cannot edit the note - """ - # Create a note as alice - note_title = "Alice's Shared Note - Write Test" - note_content = "This note is shared with Charlie for editing." - note_category = "SharedNotes" - - logger.info("Creating note as alice...") - created_note = await nc_client.notes.create_note( - title=note_title, content=note_content, category=note_category - ) - note_id = created_note.get("id") - - try: - # TODO: Share the note with charlie (edit permission) and bob (read-only) - # Note: Nextcloud Notes sharing is folder-based - - # Test: Charlie can append content to the note - logger.info("Charlie attempting to append content via MCP...") - result = await charlie_mcp_client.call_tool( - "nc_notes_append_content", - arguments={ - "note_id": note_id, - "content": "\n\nCharlie added this content.", - }, - ) - - # If sharing is properly configured, Charlie should succeed - # Without proper sharing setup, this will fail - logger.info(f"Charlie's append result: isError={result.isError}") - if not result.isError: - logger.info("Charlie successfully appended content (shares configured)") - else: - logger.warning("Charlie could not append (shares not yet configured)") - - # Test: Bob attempts to append content (should fail) - logger.info("Bob attempting to append content via MCP...") - result = await bob_mcp_client.call_tool( - "nc_notes_append_content", - arguments={"note_id": note_id, "content": "\n\nBob tried to add this."}, - ) - - # Bob should fail (read-only access) - logger.info(f"Bob's append result: isError={result.isError}") - if result.isError: - logger.info("Bob correctly denied write access") - else: - logger.warning("Bob unexpectedly succeeded (permissions issue?)") - - finally: - # Cleanup - logger.info(f"Cleaning up note {note_id}") - await nc_client.notes.delete_note(note_id) - - -async def test_user_isolation_notes(nc_client, alice_mcp_client, bob_mcp_client): - """ - Test that users can only see their own notes when not shared. - - Scenario: - 1. Admin creates a note as alice (not shared) - 2. Admin creates a note as bob (not shared) - 3. Alice can only see her own note - 4. Bob can only see his own note - """ - # Create alice's note - logger.info("Creating alice's private note...") - alice_note = await nc_client.notes.create_note( - title="Alice's Private Note", - content="This is Alice's private content.", - category="AlicePrivate", - ) - alice_note_id = alice_note.get("id") - - # Create bob's note - logger.info("Creating bob's private note...") - bob_note = await nc_client.notes.create_note( - title="Bob's Private Note", - content="This is Bob's private content.", - category="BobPrivate", - ) - bob_note_id = bob_note.get("id") - - try: - # Test: Alice searches all notes - logger.info("Alice searching all notes via MCP...") - result = await alice_mcp_client.call_tool( - "nc_notes_search_notes", arguments={"query": ""} - ) - - assert result.isError is False - response_data = json.loads(result.content[0].text) - alice_notes = response_data.get("results", []) - alice_note_ids = [n["id"] for n in alice_notes] - - logger.info(f"Alice can see {len(alice_notes)} notes") - # Alice should NOT see Bob's note - assert bob_note_id not in alice_note_ids, ( - "Alice should not see Bob's private note" - ) - - # Test: Bob searches all notes - logger.info("Bob searching all notes via MCP...") - result = await bob_mcp_client.call_tool( - "nc_notes_search_notes", arguments={"query": ""} - ) - - assert result.isError is False - response_data = json.loads(result.content[0].text) - bob_notes = response_data.get("results", []) - bob_note_ids = [n["id"] for n in bob_notes] - - logger.info(f"Bob can see {len(bob_notes)} notes") - # Bob should NOT see Alice's note - assert alice_note_id not in bob_note_ids, ( - "Bob should not see Alice's private note" - ) - - logger.info("User isolation test passed: users can only see their own notes") - - finally: - # Cleanup - logger.info("Cleaning up test notes...") - await nc_client.notes.delete_note(alice_note_id) - await nc_client.notes.delete_note(bob_note_id) - - -async def test_oauth_mcp_clients_initialized( - alice_mcp_client, bob_mcp_client, charlie_mcp_client, diana_mcp_client -): - """ - Smoke test to verify all OAuth MCP clients are properly initialized. - """ - logger.info("Testing alice_mcp_client initialization...") - result = await alice_mcp_client.call_tool( - "nc_notes_search_notes", arguments={"query": ""} - ) - assert result.isError is False, f"Alice MCP client failed: {result.content}" - logger.info("Alice MCP client working") - - logger.info("Testing bob_mcp_client initialization...") - result = await bob_mcp_client.call_tool( - "nc_notes_search_notes", arguments={"query": ""} - ) - assert result.isError is False, f"Bob MCP client failed: {result.content}" - logger.info("Bob MCP client working") - - logger.info("Testing charlie_mcp_client initialization...") - result = await charlie_mcp_client.call_tool( - "nc_notes_search_notes", arguments={"query": ""} - ) - assert result.isError is False, f"Charlie MCP client failed: {result.content}" - logger.info("Charlie MCP client working") - - logger.info("Testing diana_mcp_client initialization...") - result = await diana_mcp_client.call_tool( - "nc_notes_search_notes", arguments={"query": ""} - ) - assert result.isError is False, f"Diana MCP client failed: {result.content}" - logger.info("Diana MCP client working") - - logger.info("All OAuth MCP clients successfully initialized!") diff --git a/tests/server/oauth/test_token_exchange.py b/tests/server/oauth/test_token_exchange.py deleted file mode 100644 index 8f1704e5..00000000 --- a/tests/server/oauth/test_token_exchange.py +++ /dev/null @@ -1,436 +0,0 @@ -"""Unit tests for RFC 8693 Token Exchange (ADR-004). - -Tests the critical token exchange pattern that separates: -- Session tokens (ephemeral, on-demand) -- Background tokens (stored refresh tokens) -""" - -import os -import tempfile -from unittest.mock import AsyncMock, MagicMock, patch - -import jwt -import pytest -from cryptography.fernet import Fernet - -from nextcloud_mcp_server.auth.storage import RefreshTokenStorage -from nextcloud_mcp_server.auth.token_broker import TokenBrokerService -from nextcloud_mcp_server.auth.token_exchange import TokenExchangeService - -pytestmark = pytest.mark.unit - - -@pytest.fixture -async def token_storage(): - """Create test token storage.""" - - # Generate valid Fernet key - encryption_key = Fernet.generate_key() - - # Create temporary database file - with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as tmp: - db_path = tmp.name - - storage = RefreshTokenStorage(db_path=db_path, encryption_key=encryption_key) - await storage.initialize() - - # Expose encryption key for tests that need to manually encrypt/decrypt - storage._test_encryption_key = encryption_key - - yield storage - - # Cleanup - if os.path.exists(db_path): - os.unlink(db_path) - - -@pytest.fixture -async def token_exchange_service(token_storage): - """Create test token exchange service.""" - service = TokenExchangeService( - oidc_discovery_url="http://test-idp/.well-known/openid-configuration", - client_id="test-client", - client_secret="test-secret", - nextcloud_host="http://test-nextcloud", - ) - service.storage = token_storage - yield service - await service.http_client.aclose() - - -@pytest.fixture -async def token_broker(token_storage): - """Create test token broker service.""" - broker = TokenBrokerService( - storage=token_storage, - oidc_discovery_url="http://test-idp/.well-known/openid-configuration", - nextcloud_host="http://test-nextcloud", - client_id="test-client", - client_secret="test-secret", - cache_ttl=300, - cache_early_refresh=30, - ) - yield broker - await broker.close() - - -def create_test_jwt( - user_id: str = "testuser", audience: str = "mcp-server", expires_in: int = 3600 -) -> str: - """Create a test JWT token.""" - import time - - payload = { - "sub": user_id, - "aud": audience, - "exp": int(time.time()) + expires_in, - "iat": int(time.time()), - "iss": "http://test-idp", - } - - # For testing, we don't sign the token (uses 'none' algorithm) - # In production, tokens would be properly signed - return jwt.encode(payload, "", algorithm="none") - - -class TestTokenExchange: - """Test RFC 8693 token exchange implementation.""" - - async def test_validate_flow1_token_success(self, token_exchange_service): - """Test validation of Flow 1 token with correct audience.""" - # Create token with correct audience - flow1_token = create_test_jwt(audience="mcp-server") - - # Should not raise an exception - await token_exchange_service._validate_flow1_token(flow1_token) - - async def test_validate_flow1_token_wrong_audience(self, token_exchange_service): - """Test validation fails with wrong audience.""" - # Create token with wrong audience - flow1_token = create_test_jwt(audience="nextcloud") - - with pytest.raises(ValueError, match="Invalid token audience"): - await token_exchange_service._validate_flow1_token(flow1_token) - - async def test_validate_flow1_token_expired(self, token_exchange_service): - """Test validation fails with expired token.""" - # Create expired token - flow1_token = create_test_jwt(audience="mcp-server", expires_in=-3600) - - with pytest.raises(ValueError, match="Token has expired"): - await token_exchange_service._validate_flow1_token(flow1_token) - - async def test_extract_user_id(self, token_exchange_service): - """Test extraction of user ID from token.""" - flow1_token = create_test_jwt(user_id="alice") - - user_id = token_exchange_service._extract_user_id(flow1_token) - assert user_id == "alice" - - async def test_check_provisioning_not_provisioned(self, token_exchange_service): - """Test provisioning check when user not provisioned.""" - result = await token_exchange_service._check_provisioning("unknown_user") - assert result is False - - async def test_check_provisioning_is_provisioned( - self, token_exchange_service, token_storage - ): - """Test provisioning check when user is provisioned.""" - # Store a refresh token for user - await token_storage.store_refresh_token( - user_id="alice", refresh_token="encrypted_refresh_token", flow_type="flow2" - ) - - result = await token_exchange_service._check_provisioning("alice") - assert result is True - - async def test_exchange_token_not_provisioned(self, token_exchange_service): - """Test token exchange fails when user not provisioned.""" - flow1_token = create_test_jwt(user_id="unprovisioneduser") - - with pytest.raises(RuntimeError, match="Nextcloud access not provisioned"): - await token_exchange_service.exchange_token_for_delegation( - flow1_token=flow1_token, - requested_scopes=["notes:read"], - requested_audience="nextcloud", - ) - - async def test_exchange_token_with_fallback( - self, token_exchange_service, token_storage - ): - """Test token exchange with refresh grant fallback.""" - # Store a refresh token for user - await token_storage.store_refresh_token( - user_id="alice", refresh_token="test_refresh_token", flow_type="flow2" - ) - - # Create Flow 1 token - flow1_token = create_test_jwt(user_id="alice", audience="mcp-server") - - # Mock HTTP client for token endpoint - mock_response = MagicMock() - mock_response.status_code = 200 - mock_response.json.return_value = { - "access_token": "delegated_token_12345", - "token_type": "Bearer", - "expires_in": 300, # 5 minutes - } - - with patch.object( - token_exchange_service.http_client, "post", return_value=mock_response - ): - # Mock discovery endpoint - with patch.object( - token_exchange_service, - "_discover_endpoints", - return_value={"token_endpoint": "http://test-idp/token"}, - ): - # Perform exchange - ( - token, - expires_in, - ) = await token_exchange_service.exchange_token_for_delegation( - flow1_token=flow1_token, - requested_scopes=["notes:read"], - requested_audience="nextcloud", - ) - - assert token == "delegated_token_12345" - assert expires_in == 300 - - -class TestTokenBroker: - """Test Token Broker session/background separation.""" - - async def test_get_session_token(self, token_broker, token_storage): - """Test getting ephemeral session token via exchange.""" - # Store refresh token for user - await token_storage.store_refresh_token( - user_id="alice", refresh_token="test_refresh_token", flow_type="flow2" - ) - - # Create Flow 1 token - flow1_token = create_test_jwt(user_id="alice", audience="mcp-server") - - # Mock token exchange - with patch( - "nextcloud_mcp_server.auth.token_broker.exchange_token_for_delegation", - return_value=("ephemeral_token_xyz", 300), - ): - token = await token_broker.get_session_token( - flow1_token=flow1_token, - required_scopes=["notes:read"], - requested_audience="nextcloud", - ) - - assert token == "ephemeral_token_xyz" - - # Verify token is NOT cached (ephemeral) - cached = await token_broker.cache.get("alice") - assert cached is None # Should not be in cache - - async def test_get_background_token(self, token_broker, token_storage): - """Test getting background token with stored refresh.""" - # Store encrypted refresh token for user - from cryptography.fernet import Fernet - - # Use the same encryption key as token_storage/token_broker - fernet = Fernet(token_storage._test_encryption_key) - encrypted_token = fernet.encrypt(b"background_refresh_token").decode() - - await token_storage.store_refresh_token( - user_id="alice", refresh_token=encrypted_token, flow_type="flow2" - ) - - # Mock OIDC config and token response - mock_response = MagicMock() - mock_response.status_code = 200 - mock_response.json.return_value = { - "access_token": "background_token_abc", - "token_type": "Bearer", - "expires_in": 3600, # 1 hour - } - - with patch.object( - token_broker, - "_get_oidc_config", - return_value={"token_endpoint": "http://test/token"}, - ): - with patch.object(token_broker, "_get_http_client") as mock_client: - mock_client.return_value.post = AsyncMock(return_value=mock_response) - - # Mock audience validation - with patch.object( - token_broker, "_validate_token_audience", return_value=None - ): - token = await token_broker.get_background_token( - user_id="alice", required_scopes=["notes:sync", "files:sync"] - ) - - assert token == "background_token_abc" - - # Verify token IS cached (background tokens can be cached) - cache_key = "alice:background:files:sync,notes:sync" - cached = await token_broker.cache.get(cache_key) - assert cached == "background_token_abc" - - async def test_session_background_separation(self, token_broker, token_storage): - """Test that session and background tokens are kept separate.""" - # Store refresh token - from cryptography.fernet import Fernet - - # Use the same encryption key as token_storage/token_broker - fernet = Fernet(token_storage._test_encryption_key) - encrypted_token = fernet.encrypt(b"master_refresh_token").decode() - - await token_storage.store_refresh_token( - user_id="alice", refresh_token=encrypted_token, flow_type="flow2" - ) - - flow1_token = create_test_jwt(user_id="alice", audience="mcp-server") - - # Mock different tokens for session vs background - session_token = "ephemeral_session_123" - background_token = "cached_background_456" - - # Get session token - with patch( - "nextcloud_mcp_server.auth.token_broker.exchange_token_for_delegation", - return_value=(session_token, 300), - ): - session_result = await token_broker.get_session_token( - flow1_token=flow1_token, required_scopes=["notes:read"] - ) - assert session_result == session_token - - # Get background token - mock_response = MagicMock() - mock_response.status_code = 200 - mock_response.json.return_value = { - "access_token": background_token, - "expires_in": 3600, - } - - with patch.object( - token_broker, - "_get_oidc_config", - return_value={"token_endpoint": "http://test/token"}, - ): - with patch.object(token_broker, "_get_http_client") as mock_client: - mock_client.return_value.post = AsyncMock(return_value=mock_response) - with patch.object( - token_broker, "_validate_token_audience", return_value=None - ): - background_result = await token_broker.get_background_token( - user_id="alice", required_scopes=["notes:sync"] - ) - assert background_result == background_token - - # Verify they are different tokens - assert session_result != background_result - - # Verify session token not cached - assert await token_broker.cache.get("alice") is None - - # Verify background token IS cached - cache_key = "alice:background:notes:sync" - assert await token_broker.cache.get(cache_key) == background_token - - -class TestScopeDownscoping: - """Test that tokens request only necessary scopes.""" - - async def test_session_token_minimal_scopes( - self, token_exchange_service, token_storage - ): - """Test session tokens request minimal scopes.""" - # Store refresh token - await token_storage.store_refresh_token( - user_id="alice", refresh_token="test_refresh_token", flow_type="flow2" - ) - - flow1_token = create_test_jwt(user_id="alice", audience="mcp-server") - - # Track what scopes are requested - requested_scopes = None - - async def mock_post(url, data, headers=None): - nonlocal requested_scopes - requested_scopes = data.get("scope", "").split() - - mock_response = MagicMock() - mock_response.status_code = 200 - mock_response.json.return_value = { - "access_token": "scoped_token", - "expires_in": 300, - } - return mock_response - - with patch.object( - token_exchange_service.http_client, "post", side_effect=mock_post - ): - with patch.object( - token_exchange_service, - "_discover_endpoints", - return_value={"token_endpoint": "http://test/token"}, - ): - await token_exchange_service.exchange_token_for_delegation( - flow1_token=flow1_token, - requested_scopes=["notes:read"], # Only read scope - requested_audience="nextcloud", - ) - - # Verify only requested scope was included - assert "notes:read" in requested_scopes - assert "notes:write" not in requested_scopes - assert "calendar:write" not in requested_scopes - - async def test_background_token_different_scopes(self, token_broker, token_storage): - """Test background tokens can request different scopes than session.""" - from cryptography.fernet import Fernet - - # Use the same encryption key as token_storage/token_broker - fernet = Fernet(token_storage._test_encryption_key) - encrypted_token = fernet.encrypt(b"refresh_token").decode() - - await token_storage.store_refresh_token( - user_id="alice", refresh_token=encrypted_token, flow_type="flow2" - ) - - # Track requested scopes - requested_scopes = None - - async def mock_post(url, data, headers=None): - nonlocal requested_scopes - requested_scopes = data.get("scope", "").split() - - mock_response = MagicMock() - mock_response.status_code = 200 - mock_response.json.return_value = { - "access_token": "background_sync_token", - "expires_in": 3600, - } - return mock_response - - with patch.object( - token_broker, - "_get_oidc_config", - return_value={"token_endpoint": "http://test/token"}, - ): - with patch.object(token_broker, "_get_http_client") as mock_client: - mock_client.return_value.post = mock_post - with patch.object( - token_broker, "_validate_token_audience", return_value=None - ): - await token_broker.get_background_token( - user_id="alice", - required_scopes=["notes:sync", "files:sync", "calendar:sync"], - ) - - # Verify sync scopes were requested - assert "notes:sync" in requested_scopes - assert "files:sync" in requested_scopes - assert "calendar:sync" in requested_scopes - # Basic OIDC scopes should also be included - assert "openid" in requested_scopes - assert "profile" in requested_scopes diff --git a/tests/smoke/test_smoke.py b/tests/smoke/test_smoke.py index 62f28533..3a118cb2 100644 --- a/tests/smoke/test_smoke.py +++ b/tests/smoke/test_smoke.py @@ -102,19 +102,3 @@ async def test_webdav_basic_smoke(nc_mcp_client): data = json.loads(result.content[0].text) assert "files" in data assert isinstance(data["files"], list) - - -@pytest.mark.oauth -async def test_oauth_connectivity_smoke(nc_mcp_oauth_client): - """Smoke test: Verify OAuth authentication works.""" - # List tools with OAuth - result = await nc_mcp_oauth_client.list_tools() - assert result is not None - assert len(result.tools) > 0 - - # Execute a simple tool - search_result = await nc_mcp_oauth_client.call_tool( - "nc_notes_search_notes", - arguments={"query": ""}, - ) - assert search_result.isError is False