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

refactor: remove oauth profile, migrate MCP/OAuth tests to login-flow

Browse Source 
Remove the oauth Docker Compose profile (mcp-oauth service, port 8001)
which used OAuth bearer tokens for direct NC API access, requiring
upstream OIDC patches. All NC access should use app passwords via
Login Flow v2 or BasicAuth.

Changes:
- Remove mcp-oauth service from docker-compose.yml
- Remove oauth mode from CI test matrix
- Delete oauth pass-through tests (core, permissions, token exchange)
- Delete oauth-specific tests (elicitation, NC PHP app, astrolabe)
- Migrate MCP/OAuth integration tests to login-flow profile:
  - DCR lifecycle, deletion, token type tests
  - Scope authorization (tool filtering) tests
  - Token introspection tests
- Fix flaky consent screen automation: replace JS btn.click() with
  Playwright native click + retry (handles Vue.js event binding race)
- Add scope-filtered OAuth client fixtures to login-flow conftest
- Keep keycloak profile for external IdP testing

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2026-04-01 16:05:14 +02:00
co-authored by Claude Opus 4.6
parent 270ef82527
commit aeddc28ca6
28 changed files with 191 additions and 4264 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/test.yml
+2 -11
View File
@@ -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 .
docker-compose.yml
-52
View File
@@ -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:
pyproject.toml
-1
View File
@@ -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)",
tests/client/test_oauth.py
-103
View File
@@ -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")
tests/client/test_oauth_playwright.py
-32
View File
@@ -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")
tests/conftest.py
+29 -24
View File
@@ -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
tests/server/auth/test_userinfo_routes.py
+1 -6
View File
@@ -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
tests/server/login_flow/conftest.py
+129
View File
@@ -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
tests/server/oauth/test_dcr_deletion_methods.py → tests/server/login_flow/test_dcr_deletion_methods.py
+1 -1
View File
@@ -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
tests/server/oauth/test_dcr_lifecycle.py → tests/server/login_flow/test_dcr_lifecycle.py
+1 -1
View File
@@ -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(
tests/server/oauth/test_dcr_new_implementation.py → tests/server/login_flow/test_dcr_new_implementation.py
+1 -1
View File
@@ -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
tests/server/oauth/test_dcr_token_type.py → tests/server/login_flow/test_dcr_token_type.py
+1 -1
View File
@@ -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:
tests/server/oauth/test_introspection_authorization.py → tests/server/login_flow/test_introspection_authorization.py
+1 -1
View File
@@ -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")
tests/server/oauth/test_scope_authorization.py → tests/server/login_flow/test_scope_authorization.py
+25 -25
View File
@@ -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
tests/server/oauth/__init__.py
-1
View File
@@ -1 +0,0 @@
"""OAuth-specific integration tests."""
tests/server/oauth/test_astrolabe_multi_server_integration.py
-109
View File
@@ -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")
tests/server/oauth/test_elicitation_integration.py
-206
View File
@@ -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")
tests/server/oauth/test_keycloak_dcr.py
-630
View File
@@ -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
tests/server/oauth/test_keycloak_external_idp.py
-566
View File
@@ -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)"
)
tests/server/oauth/test_login_elicitation.py
-246
View File
@@ -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")
tests/server/oauth/test_nc_php_app_debug.py
-92
View File
@@ -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()
tests/server/oauth/test_nc_php_app_oauth.py
-414
View File
@@ -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()
tests/server/oauth/test_oauth_core.py
-262
View File
@@ -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
tests/server/oauth/test_oauth_deck_permissions.py
-350
View File
@@ -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)
tests/server/oauth/test_oauth_file_permissions.py
-421
View File
@@ -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}
)
tests/server/oauth/test_oauth_notes_permissions.py
-256
View File
@@ -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!")
tests/server/oauth/test_token_exchange.py
-436
View File
@@ -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
tests/smoke/test_smoke.py
-16
View File
@@ -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