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

feat: Split read/write scopes into app:read/write scopes

Browse Source
This commit is contained in:
Chris Coutinho
2025-10-24 04:38:49 +02:00
parent d55e5708c7
commit d452684535
45 changed files with 1630 additions and 952 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/client/cookbook/test_cookbook_api.py
+6 -6
View File
@@ -1,7 +1,7 @@
import asyncio
import logging
import uuid
import anyio
import pytest
from httpx import HTTPStatusError
@@ -120,7 +120,7 @@ async def test_cookbook_update_recipe(nc_client: NextcloudClient):
assert updated_id == recipe_id
# Verify the update
await asyncio.sleep(1) # Allow propagation
await anyio.sleep(1) # Allow propagation
updated_recipe = await nc_client.cookbook.get_recipe(recipe_id)
assert updated_recipe["description"] == "Updated description"
assert len(updated_recipe["recipeIngredient"]) == 2
@@ -227,7 +227,7 @@ async def test_cookbook_search_recipes(nc_client: NextcloudClient):
try:
# Allow time for indexing
await asyncio.sleep(2)
await anyio.sleep(2)
# Search for the recipe
logger.info(f"Searching for recipes with keyword: {unique_keyword}")
@@ -279,7 +279,7 @@ async def test_cookbook_get_recipes_in_category(nc_client: NextcloudClient):
try:
# Allow time for indexing
await asyncio.sleep(2)
await anyio.sleep(2)
# Get recipes in this category
logger.info(f"Getting recipes in category: {unique_category}")
@@ -335,11 +335,11 @@ async def test_cookbook_get_recipes_with_keywords(nc_client: NextcloudClient):
try:
# Allow extra time for indexing
await asyncio.sleep(3)
await anyio.sleep(3)
# Trigger a reindex to ensure the recipe is indexed
await nc_client.cookbook.reindex()
await asyncio.sleep(2)
await anyio.sleep(2)
# Get recipes with this keyword
logger.info(f"Getting recipes with keyword: {unique_keyword}")
tests/client/notes/test_notes_api.py
+6 -6
View File
@@ -1,7 +1,7 @@
import asyncio
import logging
import uuid # Keep uuid if needed for generating unique data within tests
import anyio
import pytest
from httpx import HTTPStatusError
@@ -66,7 +66,7 @@ async def test_notes_api_update(nc_client: NextcloudClient, temporary_note: dict
assert updated_note["etag"] != original_etag # Etag must change
# Optional: Verify update by reading again
await asyncio.sleep(1) # Allow potential propagation delay
await anyio.sleep(1) # Allow potential propagation delay
read_updated_note = await nc_client.notes.get_note(note_id=note_id)
assert read_updated_note["title"] == update_title
assert read_updated_note["content"] == update_content
@@ -96,7 +96,7 @@ async def test_notes_api_update_conflict(
new_etag = first_updated_note["etag"]
assert new_etag != original_etag
logger.info(f"Note ID: {note_id} updated, new etag: {new_etag}")
await asyncio.sleep(1)
await anyio.sleep(1)
# Now attempt update with the *original* etag
logger.info(
@@ -155,7 +155,7 @@ async def test_notes_api_append_content_to_existing_note(
assert updated_note["content"] == expected_content
# Verify by reading the note again
await asyncio.sleep(1) # Allow potential propagation delay
await anyio.sleep(1) # Allow potential propagation delay
read_note = await nc_client.notes.get_note(note_id=note_id)
assert read_note["content"] == expected_content
logger.info(f"Successfully appended content to note ID: {note_id}")
@@ -189,7 +189,7 @@ async def test_notes_api_append_content_to_empty_note(nc_client: NextcloudClient
assert updated_note["content"] == append_text
# Verify by reading the note again
await asyncio.sleep(1)
await anyio.sleep(1)
read_note = await nc_client.notes.get_note(note_id=note_id)
assert read_note["content"] == append_text
logger.info(f"Successfully appended content to empty note ID: {note_id}")
@@ -237,7 +237,7 @@ async def test_notes_api_append_content_multiple_times(
assert updated_note["content"] == expected_content_after_second
# Verify by reading the note again
await asyncio.sleep(1)
await anyio.sleep(1)
read_note = await nc_client.notes.get_note(note_id=note_id)
assert read_note["content"] == expected_content_after_second
logger.info(f"Successfully performed multiple appends to note ID: {note_id}")
tests/client/tables/test_tables_api.py
+4 -4
View File
@@ -1,8 +1,8 @@
import asyncio
import logging
import uuid
from typing import Any, Dict
import anyio
import pytest
from httpx import HTTPStatusError
@@ -264,7 +264,7 @@ async def test_tables_create_row(
assert created_row["tableId"] == table_id
# Verify the row was created by reading it back
await asyncio.sleep(1) # Allow potential propagation delay
await anyio.sleep(1) # Allow potential propagation delay
rows = await nc_client.tables.get_table_rows(table_id)
created_row_id = created_row["id"]
@@ -338,7 +338,7 @@ async def test_tables_update_row(
assert updated_row["id"] == row_id
# Verify the row was updated by reading it back
await asyncio.sleep(1) # Allow potential propagation delay
await anyio.sleep(1) # Allow potential propagation delay
table_id = sample_table_info["table_id"]
rows = await nc_client.tables.get_table_rows(table_id)
@@ -401,7 +401,7 @@ async def test_tables_delete_row(
# The delete response might vary, but it should be successful
# Verify the row was deleted by trying to find it
await asyncio.sleep(1) # Allow potential propagation delay
await anyio.sleep(1) # Allow potential propagation delay
rows = await nc_client.tables.get_table_rows(table_id)
# Ensure the deleted row is not in the results
tests/client/test_sharing_api.py
-3
View File
@@ -9,7 +9,6 @@ logger = logging.getLogger(__name__)
pytestmark = pytest.mark.integration
@pytest.mark.anyio
async def test_create_and_delete_share(nc_client):
"""Test creating and deleting a file share."""
# Create a test user to share with
@@ -68,7 +67,6 @@ async def test_create_and_delete_share(nc_client):
pass
@pytest.mark.anyio
async def test_update_share_permissions(nc_client):
"""Test updating share permissions."""
# Create a test user to share with
@@ -120,7 +118,6 @@ async def test_update_share_permissions(nc_client):
pass
@pytest.mark.anyio
async def test_list_shares(nc_client):
"""Test listing all shares."""
# Create a test user to share with
tests/conftest.py
+129 -76
View File
@@ -1,9 +1,9 @@
import asyncio
import logging
import os
import uuid
from typing import Any, AsyncGenerator
import anyio
import httpx
import pytest
from httpx import HTTPStatusError
@@ -14,6 +14,48 @@ from nextcloud_mcp_server.client import NextcloudClient
logger = logging.getLogger(__name__)
# Default scopes for OAuth testing - all app-specific read/write scopes
DEFAULT_FULL_SCOPES = (
"openid profile email "
"notes:read notes:write "
"calendar:read calendar:write "
"todo:read todo: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"
)
# Read-only scopes (all read scopes across apps) - should match DEFAULT_FULL_SCOPES read portion
DEFAULT_READ_SCOPES = (
"openid profile email "
"notes:read "
"calendar:read "
"todo:read "
"contacts:read "
"cookbook:read "
"deck:read "
"tables:read "
"files:read "
"sharing:read"
)
# Write-only scopes (all write scopes across apps) - should match DEFAULT_FULL_SCOPES write portion
DEFAULT_WRITE_SCOPES = (
"openid profile email "
"notes:write "
"calendar:write "
"todo:write "
"contacts:write "
"cookbook:write "
"deck:write "
"tables:write "
"files:write "
"sharing:write"
)
@pytest.fixture(scope="session")
def anyio_backend():
@@ -56,7 +98,7 @@ async def wait_for_nextcloud(
logger.info(
f"Nextcloud not ready yet, waiting {delay}s... (attempt {attempt}/{max_attempts})"
)
await asyncio.sleep(delay)
await anyio.sleep(delay)
logger.error(
f"Nextcloud server at {host} did not become ready after {max_attempts} attempts"
@@ -191,18 +233,18 @@ async def nc_mcp_oauth_jwt_client(
) -> AsyncGenerator[ClientSession, Any]:
"""
Fixture to create an MCP client session for JWT OAuth integration tests.
Connects to the JWT OAuth-enabled MCP server on port 8002 with OAuth authentication.
Connects to the OAuth-enabled MCP server on port 8001 with JWT token authentication.
This server uses JWT tokens (RFC 9068) instead of opaque tokens, enabling:
- Token introspection via JWT signature verification
Uses JWT tokens (RFC 9068) which provide:
- Token validation via JWT signature verification (JWKS)
- Scope information embedded in token claims
- Offline token validation without userinfo endpoint
- Faster validation without userinfo endpoint call
Uses headless browser automation suitable for CI/CD.
Uses anyio pytest plugin for proper async fixture handling.
"""
async for session in create_mcp_client_session(
url="http://localhost:8002/mcp",
url="http://localhost:8001/mcp",
token=playwright_oauth_token_jwt,
client_name="OAuth JWT MCP (Playwright)",
):
@@ -215,17 +257,17 @@ async def nc_mcp_oauth_client_read_only(
playwright_oauth_token_read_only: str,
) -> AsyncGenerator[ClientSession, Any]:
"""
Fixture to create an MCP client session with only nc:read scope.
Connects to the JWT OAuth-enabled MCP server on port 8002.
Fixture to create an MCP client session with only read scopes.
Connects to the OAuth-enabled MCP server on port 8001.
This client should only see read tools and should get 403 errors
when attempting to call write tools.
Uses JWT MCP server because JWT tokens embed scope information in claims,
enabling proper scope-based filtering.
Uses JWT tokens because they embed scope information in claims,
enabling proper scope-based tool filtering.
"""
async for session in create_mcp_client_session(
url="http://localhost:8002/mcp",
url="http://localhost:8001/mcp",
token=playwright_oauth_token_read_only,
client_name="OAuth JWT MCP Read-Only (Playwright)",
):
@@ -238,17 +280,17 @@ async def nc_mcp_oauth_client_write_only(
playwright_oauth_token_write_only: str,
) -> AsyncGenerator[ClientSession, Any]:
"""
Fixture to create an MCP client session with only nc:write scope.
Connects to the JWT OAuth-enabled MCP server on port 8002.
Fixture to create an MCP client session with only write scopes.
Connects to the OAuth-enabled MCP server on port 8001.
This client should only see write tools and should get 403 errors
when attempting to call read tools.
Uses JWT MCP server because JWT tokens embed scope information in claims,
enabling proper scope-based filtering.
Uses JWT tokens because they embed scope information in claims,
enabling proper scope-based tool filtering.
"""
async for session in create_mcp_client_session(
url="http://localhost:8002/mcp",
url="http://localhost:8001/mcp",
token=playwright_oauth_token_write_only,
client_name="OAuth JWT MCP Write-Only (Playwright)",
):
@@ -261,16 +303,16 @@ async def nc_mcp_oauth_client_full_access(
playwright_oauth_token_full_access: str,
) -> AsyncGenerator[ClientSession, Any]:
"""
Fixture to create an MCP client session with both nc:read and nc:write scopes.
Connects to the JWT OAuth-enabled MCP server on port 8002.
Fixture to create an MCP client session with both read and write scopes.
Connects to the OAuth-enabled MCP server on port 8001.
This client should see all tools and be able to call all operations.
Uses JWT MCP server because JWT tokens embed scope information in claims,
enabling proper scope-based filtering.
Uses JWT tokens because they embed scope information in claims,
enabling proper scope-based tool filtering.
"""
async for session in create_mcp_client_session(
url="http://localhost:8002/mcp",
url="http://localhost:8001/mcp",
token=playwright_oauth_token_full_access,
client_name="OAuth JWT MCP Full Access (Playwright)",
):
@@ -284,18 +326,18 @@ async def nc_mcp_oauth_client_no_custom_scopes(
) -> AsyncGenerator[ClientSession, Any]:
"""
Fixture to create an MCP client session with NO custom scopes.
Connects to the JWT OAuth-enabled MCP server on port 8002.
Connects to the OAuth-enabled MCP server on port 8001.
This client has only OIDC default scopes (openid, profile, email) without
application-specific scopes (nc:read, nc:write).
application-specific scopes (notes:read, notes:write, etc.).
Expected behavior: Should see 0 tools (all tools require custom scopes).
Uses JWT MCP server because JWT tokens embed scope information in claims,
enabling proper scope-based filtering.
Uses JWT tokens because they embed scope information in claims,
enabling proper scope-based tool filtering.
"""
async for session in create_mcp_client_session(
url="http://localhost:8002/mcp",
url="http://localhost:8001/mcp",
token=playwright_oauth_token_no_custom_scopes,
client_name="OAuth JWT MCP No Custom Scopes (Playwright)",
):
@@ -682,10 +724,9 @@ async def shared_calendar_2(
try:
# Wait for first calendar to fully initialize to avoid Nextcloud rate limiting
# When creating multiple calendars rapidly, Nextcloud may not register them all
import asyncio
logger.info("Waiting before creating second calendar to avoid rate limiting...")
await asyncio.sleep(3) # Increased from 2 to 3 seconds
await anyio.sleep(3) # Increased from 2 to 3 seconds
# Create a test calendar
logger.info(f"Creating second shared test calendar: {calendar_name}")
@@ -703,9 +744,8 @@ async def shared_calendar_2(
# Verify calendar was created by listing calendars
# Add small delay to allow calendar to propagate in the system
import asyncio
await asyncio.sleep(1.0) # Allow time for calendar to propagate
await anyio.sleep(1.0) # Allow time for calendar to propagate
calendars = await nc_client.calendar.list_calendars()
calendar_names = [cal["name"] for cal in calendars]
@@ -714,7 +754,7 @@ async def shared_calendar_2(
f"Calendar {calendar_name} not found immediately after creation. Available: {calendar_names}"
)
# Try one more time after a longer delay
await asyncio.sleep(3) # Additional wait for calendar synchronization
await anyio.sleep(3) # Additional wait for calendar synchronization
calendars = await nc_client.calendar.list_calendars()
calendar_names = [cal["name"] for cal in calendars]
if calendar_name not in calendar_names:
@@ -946,7 +986,7 @@ async def shared_oauth_client_credentials(anyio_backend, oauth_callback_server):
client_id, client_secret = await _create_oauth_client_with_scopes(
callback_url=callback_url,
client_name="Pytest - Shared Test Client (Opaque)",
allowed_scopes="openid profile email nc:read nc:write",
allowed_scopes=DEFAULT_FULL_SCOPES,
token_type="Bearer", # Opaque tokens for port 8001
cache_file=".nextcloud_oauth_shared_test_client.json",
)
@@ -968,10 +1008,11 @@ async def shared_oauth_client_credentials(anyio_backend, oauth_callback_server):
@pytest.fixture(scope="session")
async def shared_jwt_oauth_client_credentials(anyio_backend, oauth_callback_server):
"""
Fixture to obtain shared JWT OAuth client credentials for JWT MCP server.
Fixture to obtain shared JWT OAuth client credentials for testing JWT token behavior.
Creates a JWT OAuth client with full scopes (nc:read and nc:write) for use with
the JWT MCP server (port 8002) that validates JWT tokens locally.
Creates a JWT OAuth client with full scopes (all app read/write scopes). The client
is configured with token_type="JWT" to request JWT-formatted access tokens from the
OIDC server (instead of opaque tokens).
Returns:
Tuple of (client_id, client_secret, callback_url, token_endpoint, authorization_endpoint)
@@ -1001,12 +1042,12 @@ async def shared_jwt_oauth_client_credentials(anyio_backend, oauth_callback_serv
"OIDC discovery missing required endpoints (token_endpoint or authorization_endpoint)"
)
# Create JWT client with full scopes (nc:read and nc:write)
# Create JWT client with full scopes (all app read/write scopes)
# Cache to file to avoid creating new client on every test run
client_id, client_secret = await _create_oauth_client_with_scopes(
callback_url=callback_url,
client_name="Pytest - Shared JWT Test Client",
allowed_scopes="openid profile email nc:read nc:write",
allowed_scopes=DEFAULT_FULL_SCOPES,
token_type="JWT", # Explicitly set JWT token type
cache_file=".nextcloud_oauth_shared_jwt_test_client.json",
)
@@ -1140,7 +1181,7 @@ async def _create_oauth_client_with_scopes(
@pytest.fixture(scope="session")
async def read_only_oauth_client_credentials(anyio_backend, oauth_callback_server):
"""
Fixture for OAuth client with only nc:read scope.
Fixture for OAuth client with only read scopes.
Returns:
Tuple of (client_id, client_secret, callback_url, token_endpoint, authorization_endpoint)
@@ -1164,7 +1205,7 @@ async def read_only_oauth_client_credentials(anyio_backend, oauth_callback_serve
client_id, client_secret = await _create_oauth_client_with_scopes(
callback_url=callback_url,
client_name="Test Client Read Only",
allowed_scopes="openid profile email nc:read",
allowed_scopes=DEFAULT_READ_SCOPES,
token_type="JWT", # JWT tokens for scope validation
)
@@ -1180,7 +1221,7 @@ async def read_only_oauth_client_credentials(anyio_backend, oauth_callback_serve
@pytest.fixture(scope="session")
async def write_only_oauth_client_credentials(anyio_backend, oauth_callback_server):
"""
Fixture for OAuth client with only nc:write scope.
Fixture for OAuth client with only write scopes.
Returns:
Tuple of (client_id, client_secret, callback_url, token_endpoint, authorization_endpoint)
@@ -1204,7 +1245,7 @@ async def write_only_oauth_client_credentials(anyio_backend, oauth_callback_serv
client_id, client_secret = await _create_oauth_client_with_scopes(
callback_url=callback_url,
client_name="Test Client Write Only",
allowed_scopes="openid profile email nc:write",
allowed_scopes=DEFAULT_WRITE_SCOPES,
token_type="JWT", # JWT tokens for scope validation
)
@@ -1220,7 +1261,7 @@ async def write_only_oauth_client_credentials(anyio_backend, oauth_callback_serv
@pytest.fixture(scope="session")
async def full_access_oauth_client_credentials(anyio_backend, oauth_callback_server):
"""
Fixture for OAuth client with both nc:read and nc:write scopes.
Fixture for OAuth client with both read and write scopes.
Returns:
Tuple of (client_id, client_secret, callback_url, token_endpoint, authorization_endpoint)
@@ -1244,7 +1285,7 @@ async def full_access_oauth_client_credentials(anyio_backend, oauth_callback_ser
client_id, client_secret = await _create_oauth_client_with_scopes(
callback_url=callback_url,
client_name="Test Client Full Access",
allowed_scopes="openid profile email nc:read nc:write",
allowed_scopes=DEFAULT_FULL_SCOPES,
token_type="JWT", # JWT tokens for scope validation
)
@@ -1265,7 +1306,7 @@ async def no_custom_scopes_oauth_client_credentials(
Fixture for OAuth client with NO custom scopes (only OIDC defaults).
Tests the security behavior when a user grants only the default OIDC scopes
(openid, profile, email) but declines custom application scopes (nc:read, nc:write).
(openid, profile, email) but declines custom application scopes (notes:read, notes:write, etc.).
Returns:
Tuple of (client_id, client_secret, callback_url, token_endpoint, authorization_endpoint)
@@ -1289,7 +1330,7 @@ async def no_custom_scopes_oauth_client_credentials(
client_id, client_secret = await _create_oauth_client_with_scopes(
callback_url=callback_url,
client_name="Test Client No Custom Scopes",
allowed_scopes="openid profile email", # No nc:read or nc:write
allowed_scopes="openid profile email", # No app-specific scopes (no app access)
token_type="JWT", # JWT tokens for scope validation
)
@@ -1363,7 +1404,7 @@ async def playwright_oauth_token(
f"client_id={client_id}&"
f"redirect_uri={quote(callback_url, safe='')}&"
f"state={state}&"
f"scope=openid%20profile%20email%20nc:read%20nc:write"
f"scope=openid%20profile%20email%20notes:read%20notes:write%20calendar:read%20calendar:write%20contacts:read%20contacts:write%20cookbook:read%20cookbook:write%20deck:read%20deck:write%20tables:read%20tables:write%20files:read%20files:write%20sharing:read%20sharing:write"
)
# Async browser automation using pytest-playwright's browser fixture
@@ -1420,7 +1461,7 @@ async def playwright_oauth_token(
raise TimeoutError(
f"Timeout waiting for OAuth callback (state={state[:16]}...)"
)
await asyncio.sleep(0.5)
await anyio.sleep(0.5)
auth_code = auth_states[state]
logger.info(f"Successfully received authorization code: {auth_code[:20]}...")
@@ -1460,7 +1501,7 @@ async def playwright_oauth_token_jwt(
"""
Fixture to obtain a JWT OAuth access token for the JWT MCP server.
Uses a JWT OAuth client with full scopes (nc:read and nc:write) to ensure
Uses a JWT OAuth client with full scopes (all app read/write scopes) to ensure
the access token includes proper scope claims that the JWT MCP server can validate.
Returns:
@@ -1470,7 +1511,7 @@ async def playwright_oauth_token_jwt(
browser,
shared_jwt_oauth_client_credentials,
oauth_callback_server,
scopes="openid profile email nc:read nc:write",
scopes=DEFAULT_FULL_SCOPES,
)
@@ -1591,7 +1632,7 @@ async def _get_oauth_token_with_scopes(
browser: Playwright browser instance
shared_oauth_client_credentials: Tuple of OAuth client credentials
oauth_callback_server: OAuth callback server fixture
scopes: Space-separated list of scopes (e.g., "openid profile email nc:read")
scopes: Space-separated list of scopes (e.g., "openid profile email notes:read")
Returns:
OAuth access token string with requested scopes
@@ -1688,7 +1729,7 @@ async def _get_oauth_token_with_scopes(
auth_code = auth_states[state]
logger.info("Auth code received from callback server")
break
await asyncio.sleep(0.1)
await anyio.sleep(0.1)
else:
raise TimeoutError(
f"Auth code not received within {timeout}s. State: {state[:16]}..."
@@ -1727,18 +1768,18 @@ async def playwright_oauth_token_read_only(
anyio_backend, browser, read_only_oauth_client_credentials, oauth_callback_server
) -> str:
"""
Fixture to obtain an OAuth access token with only nc:read scope.
Fixture to obtain an OAuth access token with only read scopes.
This token will only be able to perform read operations and should
have write tools filtered out from the tool list.
Uses a dedicated OAuth client with allowed_scopes="openid profile email nc:read"
Uses a dedicated OAuth client with allowed_scopes=DEFAULT_READ_SCOPES
"""
return await _get_oauth_token_with_scopes(
browser,
read_only_oauth_client_credentials,
oauth_callback_server,
scopes="openid profile email nc:read",
scopes=DEFAULT_READ_SCOPES,
)
@@ -1747,18 +1788,18 @@ async def playwright_oauth_token_write_only(
anyio_backend, browser, write_only_oauth_client_credentials, oauth_callback_server
) -> str:
"""
Fixture to obtain an OAuth access token with only nc:write scope.
Fixture to obtain an OAuth access token with only write scopes.
This token will only be able to perform write operations and should
have read tools filtered out from the tool list.
Uses a dedicated OAuth client with allowed_scopes="openid profile email nc:write"
Uses a dedicated OAuth client with allowed_scopes=DEFAULT_WRITE_SCOPES
"""
return await _get_oauth_token_with_scopes(
browser,
write_only_oauth_client_credentials,
oauth_callback_server,
scopes="openid profile email nc:write",
scopes=DEFAULT_WRITE_SCOPES,
)
@@ -1767,17 +1808,17 @@ async def playwright_oauth_token_full_access(
anyio_backend, browser, full_access_oauth_client_credentials, oauth_callback_server
) -> str:
"""
Fixture to obtain an OAuth access token with both nc:read and nc:write scopes.
Fixture to obtain an OAuth access token with both read and write scopes.
This token will be able to perform all operations.
Uses a dedicated JWT OAuth client with allowed_scopes="openid profile email nc:read nc:write"
Uses a dedicated JWT OAuth client with allowed_scopes=DEFAULT_FULL_SCOPES
"""
return await _get_oauth_token_with_scopes(
browser,
full_access_oauth_client_credentials,
oauth_callback_server,
scopes="openid profile email nc:read nc:write",
scopes=DEFAULT_FULL_SCOPES,
)
@@ -1795,7 +1836,7 @@ async def playwright_oauth_token_no_custom_scopes(
(openid, profile, email) but declines application-specific scopes.
Expected: JWT token will contain only default scopes, and all MCP tools
should be filtered out since they all require nc:read or nc:write.
should be filtered out since they all require app-specific scopes.
Uses a dedicated JWT OAuth client with allowed_scopes="openid profile email"
"""
@@ -1967,7 +2008,7 @@ async def _get_oauth_token_for_user(
f"client_id={client_id}&"
f"redirect_uri={quote(callback_url, safe='')}&"
f"state={state}&"
f"scope=openid%20profile%20email%20nc:read%20nc:write"
f"scope=openid%20profile%20email%20notes:read%20notes:write%20calendar:read%20calendar:write%20contacts:read%20contacts:write%20cookbook:read%20cookbook:write%20deck:read%20deck:write%20tables:read%20tables:write%20files:read%20files:write%20sharing:read%20sharing:write"
)
logger.info(f"Performing browser OAuth flow for {username}...")
@@ -2013,7 +2054,7 @@ async def _get_oauth_token_for_user(
raise TimeoutError(
f"Timeout waiting for OAuth callback for {username} (state={state[:16]}...)"
)
await asyncio.sleep(0.5)
await anyio.sleep(0.5)
auth_code = auth_states[state]
logger.info(f"Got auth code for {username}: {auth_code[:20]}...")
@@ -2064,7 +2105,6 @@ async def all_oauth_tokens(
Now uses the real callback server with state parameters for reliable
concurrent token acquisition without race conditions.
"""
import asyncio
import time
# Get auth_states dict from callback server
@@ -2077,7 +2117,7 @@ async def all_oauth_tokens(
async def get_token_with_delay(username: str, config: dict, delay: float):
"""Get token for a user after a small delay to stagger requests."""
if delay > 0:
await asyncio.sleep(delay)
await anyio.sleep(delay)
return await _get_oauth_token_for_user(
browser,
shared_oauth_client_credentials,
@@ -2087,17 +2127,30 @@ async def all_oauth_tokens(
)
# Create tasks for all users with staggered starts (0.5s apart)
tasks = {
username: get_token_with_delay(username, config, idx * 0.5)
for idx, (username, config) in enumerate(test_users_setup.items())
}
# Run all token fetches concurrently
results = await asyncio.gather(*tasks.values(), return_exceptions=True)
# Build result dict, handling any errors
user_list = list(test_users_setup.items())
tokens = {}
for username, result in zip(tasks.keys(), results):
# Run all token fetches concurrently using anyio task groups
async with anyio.create_task_group() as tg:
# Create a dict to store results as they complete
results = {}
def create_task_wrapper(username: str, config: dict, idx: int):
async def task():
try:
token = await get_token_with_delay(username, config, idx * 0.5)
results[username] = token
except Exception as e:
results[username] = e
return task
for idx, (username, config) in enumerate(user_list):
tg.start_soon(create_task_wrapper(username, config, idx))
# Build token dict, handling any errors
for username in results:
result = results[username]
if isinstance(result, Exception):
logger.error(f"Failed to get OAuth token for {username}: {result}")
raise result
tests/load/oauth_pool.py
+2 -2
View File
@@ -4,11 +4,11 @@ OAuth User Pool Management for Load Testing.
Manages multiple OAuth-authenticated users for realistic multi-user load testing scenarios.
"""
import asyncio
import logging
from dataclasses import dataclass
from typing import Any
import anyio
import httpx
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
@@ -394,7 +394,7 @@ class OAuthUserPool:
raise TimeoutError(
f"Timeout waiting for OAuth callback for {username}"
)
await asyncio.sleep(0.5)
await anyio.sleep(0.5)
auth_code = auth_states[state]
logger.info(f"Received auth code for {username}")
tests/load/oauth_workloads.py
+8 -3
View File
@@ -5,7 +5,6 @@ Defines coordinated workflows that span multiple users, simulating realistic
collaborative scenarios like note sharing, file collaboration, and permission management.
"""
import asyncio
import json
import logging
import random
@@ -15,6 +14,8 @@ from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any, Awaitable, Callable
import anyio
from tests.load.oauth_pool import UserSessionWrapper
logger = logging.getLogger(__name__)
@@ -299,7 +300,9 @@ class CollaborativeEditWorkflow(Workflow):
)
)
await asyncio.gather(*read_tasks)
async with anyio.create_task_group() as tg:
for task in read_tasks:
tg.start_soon(task)
# Step 3: Append content concurrently by all collaborators
append_tasks = []
@@ -318,7 +321,9 @@ class CollaborativeEditWorkflow(Workflow):
)
)
await asyncio.gather(*append_tasks)
async with anyio.create_task_group() as tg:
for task in append_tasks:
tg.start_soon(task)
# Step 4: Owner verifies final state
await self._execute_step(
tests/server/oauth/__init__.py
+1
View File
@@ -0,0 +1 @@
"""OAuth-specific integration tests."""
tests/server/oauth/test_dcr_token_type.py
+395
View File
@@ -0,0 +1,395 @@
"""
Tests for Dynamic Client Registration (DCR) token_type parameter.
These tests verify that the Nextcloud OIDC server properly honors the token_type
parameter during client registration, issuing the correct type of access tokens:
- token_type="JWT" → JWT-formatted tokens (RFC 9068)
- token_type="Bearer" → Opaque tokens (standard OAuth2)
This is critical for ensuring:
1. Client choice is respected by the OIDC server
2. JWT tokens embed scope information in claims
3. Opaque tokens require introspection for scope information
"""
import base64
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 register_client
from ...conftest import _handle_oauth_consent_screen
logger = logging.getLogger(__name__)
pytestmark = [pytest.mark.integration, pytest.mark.oauth]
def is_jwt_format(token: str) -> bool:
"""
Check if a token is in JWT format (three base64-encoded parts separated by dots).
Args:
token: The access token to check
Returns:
True if token appears to be JWT format, False otherwise
"""
parts = token.split(".")
if len(parts) != 3:
return False
# Try to decode the header and payload to verify it's valid base64
try:
# Add padding if needed
header_part = parts[0] + "=" * (4 - len(parts[0]) % 4)
payload_part = parts[1] + "=" * (4 - len(parts[1]) % 4)
# Decode
base64.urlsafe_b64decode(header_part)
base64.urlsafe_b64decode(payload_part)
return True
except Exception:
return False
def decode_jwt_payload(token: str) -> dict:
"""
Decode the payload of a JWT token without verification.
Args:
token: The JWT token
Returns:
Dict containing the decoded payload
Raises:
ValueError: If token is not valid JWT format
"""
parts = token.split(".")
if len(parts) != 3:
raise ValueError(f"Invalid JWT format: expected 3 parts, got {len(parts)}")
# Decode payload (second part)
payload_part = parts[1] + "=" * (4 - len(parts[1]) % 4)
payload_bytes = base64.urlsafe_b64decode(payload_part)
return json.loads(payload_bytes)
async def get_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",
) -> str:
"""
Helper to obtain OAuth access token using existing client credentials.
Args:
browser: Playwright browser instance
client_id: OAuth client ID
client_secret: OAuth client secret
token_endpoint: Token endpoint URL
authorization_endpoint: 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
Returns:
Access token string
"""
nextcloud_host = os.getenv("NEXTCLOUD_HOST")
username = os.getenv("NEXTCLOUD_USERNAME")
password = os.getenv("NEXTCLOUD_PASSWORD")
if not all([nextcloud_host, username, password]):
pytest.skip(
"OAuth requires NEXTCLOUD_HOST, NEXTCLOUD_USERNAME, and NEXTCLOUD_PASSWORD"
)
# 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}"
)
# 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
# Login if needed
if "/login" in current_url or "/index.php/login" in current_url:
logger.info("Logging in for DCR test...")
await page.wait_for_selector('input[name="user"]', timeout=10000)
await page.fill('input[name="user"]', username)
await page.fill('input[name="password"]', password)
await page.click('button[type="submit"]')
await page.wait_for_load_state("networkidle", timeout=60000)
# Handle consent screen if present
try:
await _handle_oauth_consent_screen(page, username)
except Exception as e:
logger.debug(f"No consent screen or already authorized: {e}")
# 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")
return access_token
@pytest.mark.integration
async def test_dcr_respects_jwt_token_type(
anyio_backend,
browser,
oauth_callback_server,
):
"""
Test that DCR honors token_type=JWT and issues JWT-formatted tokens.
This verifies:
1. Client registration with token_type="JWT" succeeds
2. Tokens obtained via this client are JWT format (base64.base64.signature)
3. JWT payload contains expected claims (sub, iss, scope, etc.)
"""
nextcloud_host = os.getenv("NEXTCLOUD_HOST")
if not nextcloud_host:
pytest.skip("Test requires NEXTCLOUD_HOST")
auth_states, callback_url = oauth_callback_server
# Discover OIDC endpoints
async with httpx.AsyncClient(timeout=30.0) as client:
discovery_url = f"{nextcloud_host}/.well-known/openid-configuration"
discovery_response = await client.get(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")
# Register client with token_type="JWT"
logger.info("Registering OAuth client with token_type=JWT...")
client_info = await register_client(
nextcloud_url=nextcloud_host,
registration_endpoint=registration_endpoint,
client_name="DCR Test - JWT Token Type",
redirect_uris=[callback_url],
scopes="openid profile email notes:read notes:write",
token_type="JWT",
)
logger.info(f"Registered JWT client: {client_info.client_id[:16]}...")
# Obtain token via OAuth flow
access_token = await get_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,
)
# Verify token is JWT format
assert is_jwt_format(access_token), (
f"Expected JWT format token (3 parts separated by dots), "
f"but got token with {len(access_token.split('.'))} parts"
)
# Decode and verify JWT payload
payload = decode_jwt_payload(access_token)
# Verify standard JWT claims
assert "sub" in payload, "JWT payload missing 'sub' claim (subject/user ID)"
assert "iss" in payload, "JWT payload missing 'iss' claim (issuer)"
assert "exp" in payload, "JWT payload missing 'exp' claim (expiration)"
assert "iat" in payload, "JWT payload missing 'iat' claim (issued at)"
# Verify scope claim exists (critical for MCP tool filtering)
assert "scope" in payload, "JWT payload missing 'scope' claim"
scopes = payload["scope"].split()
assert "notes:read" in scopes, "JWT scope claim missing notes:read"
assert "notes:write" in scopes, "JWT scope claim missing notes:write"
logger.info(
f"✅ DCR with token_type=JWT works correctly! "
f"Token is JWT format with scope claim: {payload['scope']}"
)
@pytest.mark.integration
async def test_dcr_respects_bearer_token_type(
anyio_backend,
browser,
oauth_callback_server,
):
"""
Test that DCR honors token_type=Bearer and issues opaque tokens.
This verifies:
1. Client registration with token_type="Bearer" succeeds
2. Tokens obtained via this client are opaque (NOT JWT format)
3. Opaque tokens are simple strings, not base64-encoded structures
"""
nextcloud_host = os.getenv("NEXTCLOUD_HOST")
if not nextcloud_host:
pytest.skip("Test requires NEXTCLOUD_HOST")
auth_states, callback_url = oauth_callback_server
# Discover OIDC endpoints
async with httpx.AsyncClient(timeout=30.0) as client:
discovery_url = f"{nextcloud_host}/.well-known/openid-configuration"
discovery_response = await client.get(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")
# Register client with token_type="Bearer" (opaque tokens)
logger.info("Registering OAuth client with token_type=Bearer...")
client_info = await register_client(
nextcloud_url=nextcloud_host,
registration_endpoint=registration_endpoint,
client_name="DCR Test - Bearer Token Type",
redirect_uris=[callback_url],
scopes="openid profile email notes:read notes:write",
token_type="Bearer",
)
logger.info(f"Registered Bearer client: {client_info.client_id[:16]}...")
# Obtain token via OAuth flow
access_token = await get_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,
)
# Verify token is NOT JWT format
assert not is_jwt_format(access_token), (
f"Expected opaque token (not JWT format), "
f"but got token that looks like JWT: {access_token[:50]}..."
)
# Opaque tokens should be simple strings (not parseable as JWT)
try:
decode_jwt_payload(access_token)
pytest.fail("Opaque token should not be decodable as JWT")
except ValueError:
# Expected - opaque tokens are not JWT format
pass
logger.info(
f"✅ DCR with token_type=Bearer works correctly! "
f"Token is opaque (not JWT format): {access_token[:30]}..."
)
@pytest.mark.integration
async def test_jwt_tokens_embed_scopes_in_payload():
"""
Test that JWT tokens contain scope information in the payload.
This is critical for MCP server's dynamic tool filtering, which extracts
scopes from JWT token claims without making additional API calls.
Note: Uses existing shared JWT OAuth client fixture.
"""
from ...conftest import (
DEFAULT_FULL_SCOPES,
)
nextcloud_host = os.getenv("NEXTCLOUD_HOST")
if not nextcloud_host:
pytest.skip("Test requires NEXTCLOUD_HOST")
# This test leverages the existing JWT client creation helper
# to verify that JWT tokens contain scope claims
# The test verifies that when we create a JWT client with specific scopes,
# and obtain a token, the token's payload contains those scopes
# This is already tested implicitly by the scope authorization tests,
# but we document the behavior explicitly here for reference
logger.info(
"✅ JWT token scope embedding verified. "
f"Expected scopes in JWT payload: {DEFAULT_FULL_SCOPES}"
)
# This test primarily serves as documentation
# Actual verification happens in test_dcr_respects_jwt_token_type
assert True
tests/server/test_introspection_authorization.py → tests/server/oauth/test_introspection_authorization.py
+4 -4
View File
@@ -9,7 +9,6 @@ authorization rules:
4. Other clients cannot introspect tokens they don't own or aren't the audience for
"""
import asyncio
import logging
import os
import secrets
@@ -19,11 +18,12 @@ import time
from typing import AsyncGenerator
from urllib.parse import quote
import anyio
import httpx
import pytest
# Import from the root tests/ conftest.py using relative import
from ..conftest import _handle_oauth_consent_screen
from ...conftest import _handle_oauth_consent_screen
logger = logging.getLogger(__name__)
@@ -216,7 +216,7 @@ async def _obtain_token_for_client(
logger.info(f"After login: {current_url}")
# Wait a bit for page to fully render after login
await asyncio.sleep(2)
await anyio.sleep(2)
current_url = page.url
logger.info(f"After waiting, current URL: {current_url}")
@@ -251,7 +251,7 @@ async def _obtain_token_for_client(
raise TimeoutError(
f"Timeout waiting for OAuth callback (state={state[:16]}...)"
)
await asyncio.sleep(0.5)
await anyio.sleep(0.5)
auth_code = auth_states[state]
logger.info(f"Successfully received auth code: {auth_code[:20]}...")
tests/server/oauth/test_oauth_core.py
+262
View File
@@ -0,0 +1,262 @@
"""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/test_oauth_deck_permissions.py → tests/server/oauth/test_oauth_deck_permissions.py
-4
View File
@@ -46,7 +46,6 @@ async def delete_board_acl(nc_client, board_id: int, acl_id: int):
logger.info(f"Deleted ACL {acl_id} from board {board_id}")
@pytest.mark.anyio
async def test_deck_board_view_permissions(
nc_client, alice_mcp_client, bob_mcp_client, diana_mcp_client
):
@@ -119,7 +118,6 @@ async def test_deck_board_view_permissions(
await nc_client.deck.delete_board(board_id)
@pytest.mark.anyio
async def test_deck_board_edit_permissions(
nc_client, alice_mcp_client, charlie_mcp_client, bob_mcp_client
):
@@ -214,7 +212,6 @@ async def test_deck_board_edit_permissions(
await nc_client.deck.delete_board(board_id)
@pytest.mark.anyio
async def test_deck_board_manage_permissions(
nc_client, alice_mcp_client, charlie_mcp_client
):
@@ -289,7 +286,6 @@ async def test_deck_board_manage_permissions(
await nc_client.deck.delete_board(board_id)
@pytest.mark.anyio
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.
tests/server/test_oauth_file_permissions.py → tests/server/oauth/test_oauth_file_permissions.py
-4
View File
@@ -18,7 +18,6 @@ logger = logging.getLogger(__name__)
pytestmark = [pytest.mark.integration, pytest.mark.oauth]
@pytest.mark.anyio
async def test_file_share_read_permissions(
alice_mcp_client, bob_mcp_client, diana_mcp_client
):
@@ -104,7 +103,6 @@ async def test_file_share_read_permissions(
)
@pytest.mark.anyio
async def test_file_share_write_permissions(
alice_mcp_client, charlie_mcp_client, bob_mcp_client
):
@@ -210,7 +208,6 @@ async def test_file_share_write_permissions(
)
@pytest.mark.anyio
async def test_file_list_permissions(alice_mcp_client, bob_mcp_client):
"""
Test that file listing respects share permissions.
@@ -326,7 +323,6 @@ async def test_file_list_permissions(alice_mcp_client, bob_mcp_client):
)
@pytest.mark.anyio
async def test_folder_share_permissions(alice_mcp_client, bob_mcp_client):
"""
Test that folder sharing works correctly.
tests/server/test_oauth_notes_permissions.py → tests/server/oauth/test_oauth_notes_permissions.py
-4
View File
@@ -15,7 +15,6 @@ logger = logging.getLogger(__name__)
pytestmark = [pytest.mark.integration, pytest.mark.oauth]
@pytest.mark.anyio
async def test_notes_share_read_permissions(
nc_client, alice_mcp_client, bob_mcp_client, diana_mcp_client
):
@@ -82,7 +81,6 @@ async def test_notes_share_read_permissions(
await nc_client.notes.delete_note(note_id)
@pytest.mark.anyio
async def test_notes_share_write_permissions(
nc_client, alice_mcp_client, charlie_mcp_client, bob_mcp_client
):
@@ -149,7 +147,6 @@ async def test_notes_share_write_permissions(
await nc_client.notes.delete_note(note_id)
@pytest.mark.anyio
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.
@@ -222,7 +219,6 @@ async def test_user_isolation_notes(nc_client, alice_mcp_client, bob_mcp_client)
await nc_client.notes.delete_note(bob_note_id)
@pytest.mark.anyio
async def test_oauth_mcp_clients_initialized(
alice_mcp_client, bob_mcp_client, charlie_mcp_client, diana_mcp_client
):
tests/server/test_scope_authorization.py → tests/server/oauth/test_scope_authorization.py
+77 -76
View File
@@ -1,11 +1,14 @@
"""Integration tests for OAuth scope-based authorization and dynamic tool filtering.
These tests verify:
1. Dynamic tool filtering based on user's token scopes
1. Dynamic tool filtering based on user's token scopes (using JWT tokens)
2. Scope enforcement (403 responses for insufficient scopes)
3. Protected Resource Metadata (PRM) endpoint
3. Protected Resource Metadata (PRM) endpoint (RFC 9728)
4. WWW-Authenticate challenge headers
5. BasicAuth bypass (all tools visible)
Note: Tests use JWT OAuth tokens because scopes are embedded in the token payload,
enabling efficient scope-based tool filtering without additional API calls.
"""
import pytest
@@ -25,8 +28,8 @@ async def test_prm_endpoint():
prm_data = response.json()
assert prm_data["resource"] == "http://localhost:8001/mcp"
assert "nc:read" in prm_data["scopes_supported"]
assert "nc:write" in prm_data["scopes_supported"]
assert "notes:read" in prm_data["scopes_supported"]
assert "notes:write" in prm_data["scopes_supported"]
assert "http://localhost:8080" in prm_data["authorization_servers"]
assert "header" in prm_data["bearer_methods_supported"]
assert "RS256" in prm_data["resource_signing_alg_values_supported"]
@@ -56,12 +59,12 @@ async def test_basicauth_shows_all_tools(nc_mcp_client):
@pytest.mark.integration
async def test_read_only_token_filters_write_tools(nc_mcp_oauth_client_read_only):
"""Test that a token with only nc:read scope filters out write tools."""
"""Test that a token with only read scopes filters out write tools."""
import logging
logger = logging.getLogger(__name__)
# Connect with token that has only "nc:read" scope
# Connect with token that has only "notes:read" scope
result = await nc_mcp_oauth_client_read_only.list_tools()
assert result is not None
assert len(result.tools) > 0
@@ -69,29 +72,27 @@ async def test_read_only_token_filters_write_tools(nc_mcp_oauth_client_read_only
tool_names = [tool.name for tool in result.tools]
logger.info(f"Read-only token sees {len(tool_names)} tools")
# Verify read tools are present
# Verify read tools are present (only for apps with :read scopes)
# Read-only token has: notes:read, calendar:read, contacts:read,
# cookbook:read, deck:read, tables:read, files:read, sharing:read
expected_read_tools = [
"nc_notes_get_note",
"nc_notes_search_notes",
"nc_calendar_list_calendars",
"nc_calendar_get_event",
"nc_webdav_list_directory",
"nc_webdav_read_file",
"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
# Verify write tools are NOT present (filtered out)
write_tools_should_be_filtered = [
"nc_notes_create_note",
"nc_notes_update_note",
"nc_notes_delete_note",
"nc_calendar_create_event",
"nc_calendar_update_event",
"nc_calendar_delete_event",
"nc_webdav_write_file",
"nc_webdav_create_directory",
"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:
@@ -107,12 +108,12 @@ async def test_read_only_token_filters_write_tools(nc_mcp_oauth_client_read_only
@pytest.mark.integration
async def test_write_only_token_filters_read_tools(nc_mcp_oauth_client_write_only):
"""Test that a token with only nc:write scope filters out read tools."""
"""Test that a token with only write scopes filters out read tools."""
import logging
logger = logging.getLogger(__name__)
# Connect with token that has only "nc:write" scope
# Connect with token that has only "notes:write" scope
result = await nc_mcp_oauth_client_write_only.list_tools()
assert result is not None
assert len(result.tools) > 0
@@ -121,28 +122,26 @@ async def test_write_only_token_filters_read_tools(nc_mcp_oauth_client_write_onl
logger.info(f"Write-only token sees {len(tool_names)} tools")
# Verify write tools are present
# Write-only token has: notes:write, calendar:write, contacts:write,
# cookbook:write, deck:write, tables:write, files:write, sharing:write
expected_write_tools = [
"nc_notes_create_note",
"nc_notes_update_note",
"nc_notes_delete_note",
"nc_calendar_create_event",
"nc_calendar_update_event",
"nc_calendar_delete_event",
"nc_webdav_write_file",
"nc_webdav_create_directory",
"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 tools are NOT present (write-only scope)
# Verify read-only tools are NOT present (write-only scope)
read_tools_should_be_filtered = [
"nc_notes_get_note",
"nc_notes_search_notes",
"nc_calendar_list_calendars",
"nc_calendar_get_event",
"nc_webdav_list_directory",
"nc_webdav_read_file",
"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:
@@ -158,31 +157,31 @@ async def test_write_only_token_filters_read_tools(nc_mcp_oauth_client_write_onl
@pytest.mark.integration
async def test_full_access_token_shows_all_tools(nc_mcp_oauth_client_full_access):
"""Test that a token with both nc:read and nc:write scopes can see all tools."""
"""Test that a token with both read and write scopes scopes can see all tools."""
import logging
logger = logging.getLogger(__name__)
# Connect with token that has both "nc:read" and "nc:write" scopes
# Connect with token that has both "notes:read" and "notes:write" scopes
result = await nc_mcp_oauth_client_full_access.list_tools()
assert result is not None
assert len(result.tools) > 0
tool_names = [tool.name for tool in result.tools]
logger.info(f"Full access token sees {len(tool_names)} tools")
logger.info(f"Tools: {sorted(tool_names)}")
# Verify both read and write tools are present
# Full access has all *read and *write scopes
expected_read_tools = [
"nc_notes_get_note",
"nc_notes_search_notes",
"nc_calendar_list_calendars",
"nc_webdav_read_file",
"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",
"nc_calendar_create_event",
"nc_webdav_write_file",
"nc_notes_create_note", # notes:write
"nc_calendar_create_event", # calendar:write
]
for tool in expected_read_tools:
@@ -215,17 +214,17 @@ async def test_scope_helper_functions():
pass
# Add scope metadata
mock_read_tool._required_scopes = ["nc:read"] # type: ignore
mock_write_tool._required_scopes = ["nc:write"] # type: ignore
mock_read_tool._required_scopes = ["notes:read"] # type: ignore
mock_write_tool._required_scopes = ["notes:write"] # type: ignore
# Test get_required_scopes
assert get_required_scopes(mock_read_tool) == ["nc:read"]
assert get_required_scopes(mock_write_tool) == ["nc:write"]
assert get_required_scopes(mock_read_tool) == ["notes:read"]
assert get_required_scopes(mock_write_tool) == ["notes:write"]
assert get_required_scopes(mock_no_scope_tool) == []
# Test has_required_scopes
read_only_scopes = {"nc:read"}
full_scopes = {"nc:read", "nc:write"}
read_only_scopes = {"notes:read"}
full_scopes = {"notes:read", "notes:write"}
no_scopes = set()
# User with only read scope
@@ -249,13 +248,13 @@ async def test_scope_decorator_stores_metadata():
"""Test that @require_scopes decorator properly stores metadata."""
from nextcloud_mcp_server.auth import require_scopes
@require_scopes("nc:read", "nc:write")
@require_scopes("notes:read", "notes:write")
async def test_function():
pass
# Check that metadata was stored
assert hasattr(test_function, "_required_scopes")
assert test_function._required_scopes == ["nc:read", "nc:write"]
assert test_function._required_scopes == ["notes:read", "notes:write"]
@pytest.mark.integration
@@ -299,36 +298,38 @@ async def test_tools_have_scope_decorators(nc_mcp_client):
assert tool in tool_names, f"Expected write tool {tool} not found"
@pytest.mark.skip(reason="Script no longer exists - decorators are already in place")
@pytest.mark.integration
async def test_scope_classification():
"""Test that our scope classification correctly identifies read vs write operations."""
from scripts.add_scope_decorators_simple import classify_function
# Test read operations
assert classify_function("nc_notes_get_note") == "nc:read"
assert classify_function("nc_notes_search_notes") == "nc:read"
assert classify_function("nc_calendar_list_events") == "nc:read"
assert classify_function("nc_webdav_read_file") == "nc:read"
assert classify_function("nc_calendar_find_availability") == "nc:read"
assert classify_function("nc_calendar_get_upcoming_events") == "nc:read"
assert classify_function("nc_notes_get_note") == "notes:read"
assert classify_function("nc_notes_search_notes") == "notes:read"
assert classify_function("nc_calendar_list_events") == "calendar:read"
assert classify_function("nc_webdav_read_file") == "files:read"
assert classify_function("nc_calendar_find_availability") == "calendar:read"
assert classify_function("nc_calendar_get_upcoming_events") == "notes:read"
# Test write operations
assert classify_function("nc_notes_create_note") == "nc:write"
assert classify_function("nc_notes_update_note") == "nc:write"
assert classify_function("nc_notes_delete_note") == "nc:write"
assert classify_function("nc_notes_append_content") == "nc:write"
assert classify_function("nc_calendar_create_event") == "nc:write"
assert classify_function("nc_calendar_update_event") == "nc:write"
assert classify_function("nc_calendar_manage_calendar") == "nc:write"
assert classify_function("nc_webdav_write_file") == "nc:write"
assert classify_function("nc_webdav_move_resource") == "nc:write"
assert classify_function("nc_contacts_create_contact") == "nc:write"
assert classify_function("nc_cookbook_import_recipe") == "nc:write"
assert classify_function("nc_tables_insert_row") == "nc:write"
assert classify_function("deck_archive_card") == "nc:write"
assert classify_function("deck_assign_label_to_card") == "nc:write"
assert classify_function("nc_notes_create_note") == "notes:write"
assert classify_function("nc_notes_update_note") == "notes:write"
assert classify_function("nc_notes_delete_note") == "notes:write"
assert classify_function("nc_notes_append_content") == "notes:write"
assert classify_function("nc_calendar_create_event") == "calendar:write"
assert classify_function("nc_calendar_update_event") == "notes:write"
assert classify_function("nc_calendar_manage_calendar") == "notes:write"
assert classify_function("nc_webdav_write_file") == "files:write"
assert classify_function("nc_webdav_move_resource") == "notes:write"
assert classify_function("nc_contacts_create_contact") == "notes:write"
assert classify_function("nc_cookbook_import_recipe") == "notes:write"
assert classify_function("nc_tables_insert_row") == "notes:write"
assert classify_function("deck_archive_card") == "notes:write"
assert classify_function("deck_assign_label_to_card") == "notes:write"
@pytest.mark.skip(reason="Script no longer exists - decorators are already in place")
@pytest.mark.integration
async def test_all_tools_classified():
"""Verify that all tools can be properly classified as read or write."""
tests/server/test_cookbook_mcp.py
+6 -6
View File
@@ -1,8 +1,8 @@
import asyncio
import json
import logging
import uuid
import anyio
import pytest
from mcp import ClientSession
@@ -136,7 +136,7 @@ async def test_mcp_cookbook_update_recipe(
)
# 4. Verify update via direct NextcloudClient
await asyncio.sleep(1) # Allow propagation
await anyio.sleep(1) # Allow propagation
updated_recipe = await nc_client.cookbook.get_recipe(created_recipe_id)
assert updated_recipe["description"] == "Updated via MCP"
assert len(updated_recipe["recipeIngredient"]) == 2
@@ -282,7 +282,7 @@ async def test_mcp_cookbook_search_recipes(
created_recipe_id = await nc_client.cookbook.create_recipe(recipe_data)
# 2. Allow time for indexing
await asyncio.sleep(2)
await anyio.sleep(2)
# 3. Search for the recipe via MCP
logger.info(f"Searching for recipes via MCP with keyword: {unique_keyword}")
@@ -358,7 +358,7 @@ async def test_mcp_cookbook_categories_workflow(
created_recipe_id = await nc_client.cookbook.create_recipe(recipe_data)
# 2. Allow time for indexing
await asyncio.sleep(2)
await anyio.sleep(2)
# 3. List categories via MCP
logger.info("Listing categories via MCP")
@@ -433,9 +433,9 @@ async def test_mcp_cookbook_keywords_workflow(
created_recipe_id = await nc_client.cookbook.create_recipe(recipe_data)
# 2. Allow extra time for indexing and trigger reindex
await asyncio.sleep(3)
await anyio.sleep(3)
await nc_client.cookbook.reindex()
await asyncio.sleep(2)
await anyio.sleep(2)
# 3. List keywords via MCP
logger.info("Listing keywords via MCP")
tests/server/test_jwt_tokens.py
-208
View File
@@ -1,208 +0,0 @@
"""
Test JWT token structure and scope support.
This test obtains a JWT token via OAuth and examines its structure.
"""
import base64
import json
import pytest
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,
}
@pytest.mark.integration
async def test_jwt_token_structure_with_custom_client():
"""
Test that we can create a JWT-enabled OAuth client and examine the token structure.
This test manually configures a JWT client and obtains a token.
"""
import os
import httpx
# This test requires manual setup of a JWT client
# Skip if not configured
client_id = os.getenv("NEXTCLOUD_JWT_CLIENT_ID")
if not client_id:
pytest.skip("NEXTCLOUD_JWT_CLIENT_ID not set - skipping JWT token test")
_client_secret = os.getenv("NEXTCLOUD_JWT_CLIENT_SECRET")
nextcloud_host = os.getenv("NEXTCLOUD_HOST", "http://localhost:8080")
# Fetch discovery
async with httpx.AsyncClient() as client:
discovery_response = await client.get(
f"{nextcloud_host}/.well-known/openid-configuration"
)
discovery_response.raise_for_status()
discovery = discovery_response.json()
_token_endpoint = discovery["token_endpoint"]
# For this test, we'll use client credentials grant if supported
# Otherwise, skip this test
pytest.skip(
"JWT token test requires OAuth flow - use manual testing script instead"
)
@pytest.mark.integration
async def test_opaque_token_vs_jwt_comparison():
"""
Compare opaque tokens vs JWT tokens to understand the differences.
This is a documentation test that explains the findings.
"""
# This test documents our findings about JWT vs opaque tokens
# Based on manual testing with the test script
findings = {
"oidc_app_capabilities": {
"supports_jwt_tokens": True,
"supports_opaque_tokens": True,
"configuration_method": "per-client via token_type field",
"jwt_standard": "RFC 9068 (OAuth 2.0 Access Token JWT Profile)",
},
"dynamic_registration": {
"sets_allowed_scopes": False,
"note": "Dynamic registration does NOT populate allowed_scopes from the scope parameter in registration request",
"workaround": "Must use occ oidc:create with --allowed_scopes flag or manually update via web UI/API",
},
"jwt_token_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 (THIS IS THE KEY!)",
"client_id": "client identifier",
"jti": "JWT ID",
# Optional based on scopes:
"roles": "if roles scope present",
"groups": "if groups scope present",
"email": "if email scope present",
"name": "if profile scope present",
},
"scope_claim": {
"format": "space-separated string",
"example": "openid profile email nc:read nc:write",
"extraction": "payload['scope'].split()",
},
},
"scope_validation": {
"oidc_app": {
"validates": True,
"method": "Intersects requested scopes with allowed_scopes per client",
"location": "LoginRedirectorController.php:251-267",
},
"user_oidc_app": {
"validates_scopes": False,
"validates": ["token expiration", "issuer", "audience (optional)"],
"limitation": "Does NOT extract or validate scopes from JWT",
},
},
"token_size": {
"opaque": "72 characters",
"jwt": "~800-1200 characters (depends on claims)",
"overhead": "JWT is 10-15x larger than opaque tokens",
},
"recommendation": {
"for_mcp_server": "Use JWT tokens with self-validation",
"reasoning": [
"Can extract scopes directly from token payload",
"No additional API call needed",
"Standard approach (RFC 9068)",
"Works with existing oidc app",
],
"alternative": "Implement introspection endpoint in oidc app (future work)",
},
}
# Print findings for documentation
print("\n" + "=" * 80)
print("JWT Token vs Opaque Token Findings")
print("=" * 80)
print(json.dumps(findings, indent=2))
print("=" * 80 + "\n")
# This test always passes - it's for documentation
assert True, "Findings documented"
@pytest.mark.integration
async def test_scope_presence_in_jwt():
"""
Verify that custom scopes (nc:read, nc:write) are present in JWT tokens.
NOTE: This test documents the expected behavior based on manual testing.
Actual implementation will be tested in integration tests after JWT validation is implemented.
"""
expected_behavior = {
"client_configuration": {
"allowed_scopes": "openid profile email nc:read nc:write",
"token_type": "jwt",
},
"authorization_request": {
"scope": "openid profile email nc:read nc:write",
},
"token_response": {
"access_token": "JWT with scope claim",
},
"jwt_payload": {
"scope": "openid profile email nc:read nc:write", # All requested scopes present if in allowed_scopes
},
"scope_filtering": {
"description": "oidc app filters requested scopes against allowed_scopes",
"example": {
"requested": "openid profile nc:read nc:write nc:admin",
"allowed": "openid profile email nc:read nc:write",
"granted": "openid profile nc:read nc:write", # nc:admin filtered out, email not requested
},
},
}
print("\n" + "=" * 80)
print("Expected JWT Scope Behavior")
print("=" * 80)
print(json.dumps(expected_behavior, indent=2))
print("=" * 80 + "\n")
assert True, "Expected behavior documented"
if __name__ == "__main__":
# Run with: uv run pytest tests/server/test_jwt_tokens.py -v
pytest.main([__file__, "-v", "-s"])
tests/server/test_mcp_oauth.py
-60
View File
@@ -1,60 +0,0 @@
import json
import logging
import pytest
logger = logging.getLogger(__name__)
pytestmark = [pytest.mark.integration, pytest.mark.oauth]
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."""
import json
# 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."
)
tests/server/test_mcp_oauth_jwt.py
-246
View File
@@ -1,246 +0,0 @@
"""Integration tests for JWT OAuth authentication.
These tests verify:
1. JWT token authentication works correctly
2. JWT token verification via JWKS
3. Scope information is properly extracted from JWT claims
4. Dynamic tool filtering works with JWT tokens
5. All MCP operations work with JWT authentication
"""
import json
import logging
import pytest
logger = logging.getLogger(__name__)
pytestmark = [pytest.mark.integration, pytest.mark.oauth]
async def test_jwt_mcp_server_connection(nc_mcp_oauth_jwt_client):
"""Test connection to JWT OAuth-enabled MCP server."""
result = await nc_mcp_oauth_jwt_client.list_tools()
assert result is not None
assert len(result.tools) > 0
logger.info(f"JWT OAuth MCP server has {len(result.tools)} tools available")
async def test_jwt_token_authentication(nc_mcp_oauth_jwt_client):
"""Test that JWT token authentication works."""
# Execute a simple read operation
result = await nc_mcp_oauth_jwt_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)
assert "results" in response_data
assert isinstance(response_data["results"], list)
logger.info(
f"Successfully authenticated with JWT token and executed tool, got {len(response_data['results'])} notes."
)
async def test_jwt_tool_list_operations(nc_mcp_oauth_jwt_client):
"""Test that list_tools works with JWT authentication."""
result = await nc_mcp_oauth_jwt_client.list_tools()
# Verify we have tools
assert len(result.tools) > 0
# Verify some expected tools exist
tool_names = [tool.name for tool in result.tools]
assert "nc_notes_get_note" in tool_names
assert "nc_notes_create_note" in tool_names
assert "nc_calendar_list_calendars" in tool_names
assert "nc_webdav_list_directory" in tool_names
logger.info(f"JWT server provides {len(result.tools)} tools")
async def test_jwt_read_operation(nc_mcp_oauth_jwt_client):
"""Test read operation with JWT authentication."""
# List calendars (read operation)
result = await nc_mcp_oauth_jwt_client.call_tool(
"nc_calendar_list_calendars", arguments={}
)
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)
assert "calendars" in response_data
assert isinstance(response_data["calendars"], list)
logger.info(
f"Successfully executed read operation with JWT, got {len(response_data['calendars'])} calendars."
)
async def test_jwt_write_operation(nc_mcp_oauth_jwt_client):
"""Test write operation with JWT authentication."""
import uuid
# Create a note (write operation)
note_title = f"JWT Test Note {uuid.uuid4().hex[:8]}"
note_content = "This note was created during JWT authentication testing"
result = await nc_mcp_oauth_jwt_client.call_tool(
"nc_notes_create_note",
arguments={
"title": note_title,
"content": note_content,
"category": "Testing",
},
)
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)
# Verify note was created
assert "id" in response_data
assert response_data["title"] == note_title
note_id = response_data["id"]
logger.info(f"Successfully created note {note_id} with JWT authentication")
# Clean up: Delete the note
delete_result = await nc_mcp_oauth_jwt_client.call_tool(
"nc_notes_delete_note", arguments={"note_id": note_id}
)
assert delete_result.isError is False, f"Cleanup failed: {delete_result.content}"
logger.info(f"Cleaned up test note {note_id}")
async def test_jwt_multiple_operations(nc_mcp_oauth_jwt_client):
"""Test multiple operations with same JWT token to verify token persistence."""
# 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 multiple operations with JWT token")
async def test_jwt_vs_opaque_token_compatibility(
nc_mcp_oauth_client, nc_mcp_oauth_jwt_client
):
"""Verify that both opaque and JWT tokens provide same functionality."""
# Execute same operation on both servers
opaque_result = await nc_mcp_oauth_client.call_tool(
"nc_notes_search_notes", arguments={"query": ""}
)
jwt_result = await nc_mcp_oauth_jwt_client.call_tool(
"nc_notes_search_notes", arguments={"query": ""}
)
# Both should succeed
assert opaque_result.isError is False
assert jwt_result.isError is False
# Both should have results
opaque_data = json.loads(opaque_result.content[0].text)
jwt_data = json.loads(jwt_result.content[0].text)
assert "results" in opaque_data
assert "results" in jwt_data
# Results should be the same (same user, same notes)
assert len(opaque_data["results"]) == len(jwt_data["results"])
logger.info(
"Verified opaque and JWT tokens provide identical functionality: "
f"{len(opaque_data['results'])} notes accessible from both servers"
)
async def test_jwt_error_handling(nc_mcp_oauth_jwt_client):
"""Test error handling with JWT authentication."""
# 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 server correctly handles errors for invalid operations")
async def test_jwt_scope_enforcement(nc_mcp_oauth_jwt_client):
"""Test that JWT server properly enforces scopes."""
# This test assumes the JWT token has both nc:read and nc:write scopes
# Both read and write operations should succeed
# Read operation
read_result = await nc_mcp_oauth_jwt_client.call_tool(
"nc_notes_search_notes", arguments={"query": ""}
)
assert read_result.isError is False
# Write operation
import uuid
note_title = f"Scope Test {uuid.uuid4().hex[:8]}"
write_result = await nc_mcp_oauth_jwt_client.call_tool(
"nc_notes_create_note",
arguments={
"title": note_title,
"content": "Testing scope enforcement",
"category": "Testing",
},
)
assert write_result.isError is False
# Clean up
note_id = json.loads(write_result.content[0].text)["id"]
await nc_mcp_oauth_jwt_client.call_tool(
"nc_notes_delete_note", arguments={"note_id": note_id}
)
logger.info("JWT server properly allows operations based on token scopes")
async def test_jwt_automation_worked(nc_mcp_oauth_jwt_client):
"""Test that verifies the automated JWT client creation worked correctly.
This test confirms that:
1. JWT client was auto-created during container initialization
2. MCP server loaded credentials from auto-generated file
3. JWT authentication flow works end-to-end
4. Server uses JWT tokens (not opaque tokens)
"""
# If we can connect and execute tools, the automation worked
result = await nc_mcp_oauth_jwt_client.list_tools()
assert result is not None
assert len(result.tools) > 0
# Execute a tool to verify full OAuth flow
tool_result = await nc_mcp_oauth_jwt_client.call_tool(
"nc_notes_search_notes", arguments={"query": ""}
)
assert tool_result.isError is False
logger.info(
"✅ JWT client automation successful! "
"Auto-generated credentials working correctly."
)
tests/server/test_users_api.py
-8
View File
@@ -1,9 +1,6 @@
import pytest
from nextcloud_mcp_server.client import NextcloudClient
@pytest.mark.anyio
async def test_create_and_delete_user(nc_client: NextcloudClient, test_user):
"""Test creating a user and verifying deletion (cleanup by fixture)."""
user_config = test_user
@@ -29,7 +26,6 @@ async def test_create_and_delete_user(nc_client: NextcloudClient, test_user):
# Note: Fixture cleanup will also try to delete but handle 404 gracefully
@pytest.mark.anyio
async def test_update_user_field(nc_client: NextcloudClient, test_user):
"""Test updating user fields."""
user_config = test_user
@@ -44,7 +40,6 @@ async def test_update_user_field(nc_client: NextcloudClient, test_user):
# Fixture will handle cleanup
@pytest.mark.anyio
async def test_user_groups(nc_client: NextcloudClient, test_user_in_group):
"""Test adding and removing users from groups."""
user_config, groupid = test_user_in_group
@@ -61,7 +56,6 @@ async def test_user_groups(nc_client: NextcloudClient, test_user_in_group):
# Fixtures will handle cleanup
@pytest.mark.anyio
async def test_user_subadmins(nc_client: NextcloudClient, test_user, test_group):
"""Test promoting and demoting subadmins."""
user_config = test_user
@@ -82,7 +76,6 @@ async def test_user_subadmins(nc_client: NextcloudClient, test_user, test_group)
# Fixtures will handle cleanup
@pytest.mark.anyio
async def test_disable_enable_user(nc_client: NextcloudClient, test_user):
"""Test disabling and enabling users."""
user_config = test_user
@@ -102,7 +95,6 @@ async def test_disable_enable_user(nc_client: NextcloudClient, test_user):
# Fixture will handle cleanup
@pytest.mark.anyio
async def test_get_editable_user_fields(nc_client: NextcloudClient):
editable_fields = await nc_client.users.get_editable_user_fields()
assert "displayname" in editable_fields
tests/smoke/__init__.py
+1
View File
@@ -0,0 +1 @@
"""Critical path smoke tests for quick validation."""
tests/smoke/test_smoke.py
+120
View File
@@ -0,0 +1,120 @@
"""Smoke tests - critical path tests for quick validation.
These tests verify the most essential functionality:
- MCP server connectivity
- Basic CRUD operations for core apps
- OAuth authentication
- Tool schema validation
Run with: uv run pytest -m smoke -v
Expected runtime: ~30-60 seconds
"""
import json
import pytest
pytestmark = [pytest.mark.integration, pytest.mark.smoke]
async def test_mcp_connectivity_smoke(nc_mcp_client):
"""Smoke test: Verify MCP server is reachable and lists tools."""
tools = await nc_mcp_client.list_tools()
# Should have a reasonable number of tools
assert len(tools.tools) > 30, f"Expected >30 tools, got {len(tools.tools)}"
# Check for core tool categories
tool_names = [tool.name for tool in tools.tools]
assert any("notes" in name for name in tool_names), "Missing notes tools"
assert any("calendar" in name for name in tool_names), "Missing calendar tools"
assert any("webdav" in name for name in tool_names), "Missing webdav tools"
async def test_notes_crud_smoke(nc_mcp_client, nc_client):
"""Smoke test: Verify basic Notes CRUD operations work."""
# Create
create_result = await nc_mcp_client.call_tool(
"nc_notes_create_note",
arguments={
"title": "Smoke Test Note",
"content": "Testing basic CRUD",
"category": "test",
},
)
assert create_result.isError is False
data = json.loads(create_result.content[0].text)
note_id = data["id"]
try:
# Read
get_result = await nc_mcp_client.call_tool(
"nc_notes_get_note",
arguments={"note_id": note_id},
)
assert get_result.isError is False
# Update
update_result = await nc_mcp_client.call_tool(
"nc_notes_update_note",
arguments={
"note_id": note_id,
"title": "Updated Smoke Test",
"content": "Updated content",
"category": "test",
"etag": data["etag"],
},
)
assert update_result.isError is False
finally:
# Delete
delete_result = await nc_mcp_client.call_tool(
"nc_notes_delete_note",
arguments={"note_id": note_id},
)
assert delete_result.isError is False
async def test_calendar_basic_smoke(nc_mcp_client):
"""Smoke test: Verify calendar operations work."""
# List calendars
result = await nc_mcp_client.call_tool(
"nc_calendar_list_calendars",
arguments={},
)
assert result.isError is False
data = json.loads(result.content[0].text)
assert "calendars" in data
assert len(data["calendars"]) > 0
async def test_webdav_basic_smoke(nc_mcp_client):
"""Smoke test: Verify WebDAV file operations work."""
# List root directory
result = await nc_mcp_client.call_tool(
"nc_webdav_list_directory",
arguments={"path": "/"},
)
assert result.isError is False
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
tests/test_cli.py
+2 -1
View File
@@ -254,7 +254,8 @@ def test_default_values(runner, clean_env, monkeypatch):
# Verify default values
assert (
captured_env["NEXTCLOUD_OIDC_SCOPES"] == "openid profile email nc:read nc:write"
captured_env["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"
)
assert captured_env["NEXTCLOUD_OIDC_TOKEN_TYPE"] == "bearer"
assert captured_env["NEXTCLOUD_MCP_SERVER_URL"] == "http://localhost:8000"
tests/unit/__init__.py
+1
View File
@@ -0,0 +1 @@
"""Unit tests with mocked dependencies for fast feedback."""
tests/unit/test_response_models.py
+123
View File
@@ -0,0 +1,123 @@
"""Unit tests for Pydantic response models."""
import pytest
from nextcloud_mcp_server.models.notes import (
CreateNoteResponse,
Note,
NoteSearchResult,
SearchNotesResponse,
)
@pytest.mark.unit
def test_note_model_creation():
"""Test creating a Note model with required fields."""
note = Note(
id=123,
title="Test Note",
content="# Test Content",
modified=1700000000,
etag="abc123",
)
assert note.id == 123
assert note.title == "Test Note"
assert note.content == "# Test Content"
assert note.category == "" # default value
assert note.favorite is False # default value
assert note.etag == "abc123"
@pytest.mark.unit
def test_note_modified_datetime_property():
"""Test that Note.modified_datetime converts Unix timestamp correctly."""
note = Note(
id=1,
title="Test",
content="Content",
modified=1700000000,
etag="etag",
)
dt = note.modified_datetime
assert dt.year == 2023 # Nov 14, 2023
assert dt.month == 11
@pytest.mark.unit
def test_create_note_response_serialization():
"""Test CreateNoteResponse can serialize to JSON."""
response = CreateNoteResponse(
id=42,
title="New Note",
category="Work",
etag="xyz789",
)
# Test serialization
data = response.model_dump()
assert data["id"] == 42
assert data["title"] == "New Note"
assert data["category"] == "Work"
assert data["etag"] == "xyz789"
@pytest.mark.unit
def test_search_notes_response_wraps_results():
"""Test SearchNotesResponse wraps list of results correctly.
This is critical - FastMCP mangles raw List[Dict] responses,
so we must wrap them in a response model.
"""
results = [
NoteSearchResult(id=1, title="First Note", category="Work"),
NoteSearchResult(id=2, title="Second Note", category="Personal"),
]
response = SearchNotesResponse(
results=results,
query="test query",
total_found=2,
)
# Verify the response structure
assert len(response.results) == 2
assert response.results[0].id == 1
assert response.results[1].title == "Second Note"
assert response.query == "test query"
assert response.total_found == 2
# Verify it serializes correctly
data = response.model_dump()
assert "results" in data
assert isinstance(data["results"], list)
assert len(data["results"]) == 2
assert data["results"][0]["id"] == 1
@pytest.mark.unit
def test_note_search_result_with_score():
"""Test NoteSearchResult with optional score field."""
result = NoteSearchResult(
id=99,
title="Relevant Note",
category="Archive",
score=0.95,
)
assert result.id == 99
assert result.score == 0.95
@pytest.mark.unit
def test_note_search_result_without_score():
"""Test NoteSearchResult without optional score field."""
result = NoteSearchResult(
id=99,
title="Relevant Note",
category="Archive",
)
assert result.id == 99
assert result.score is None
tests/unit/test_scope_decorator.py
+65
View File
@@ -0,0 +1,65 @@
"""Unit tests for scope decorator metadata and classification logic."""
import pytest
from nextcloud_mcp_server.auth.scope_authorization import (
InsufficientScopeError,
require_scopes,
)
@pytest.mark.unit
def test_scope_decorator_stores_metadata():
"""Test that @require_scopes decorator stores scope requirements as function metadata."""
@require_scopes("notes:read", "notes:write")
async def example_function():
pass
# Verify metadata is stored
assert hasattr(example_function, "_required_scopes")
assert example_function._required_scopes == ["notes:read", "notes:write"]
@pytest.mark.unit
def test_scope_decorator_with_single_scope():
"""Test decorator with a single scope requirement."""
@require_scopes("calendar:read")
async def example_function():
pass
assert example_function._required_scopes == ["calendar:read"]
@pytest.mark.unit
def test_scope_decorator_with_no_scopes():
"""Test decorator with no scope requirements."""
@require_scopes()
async def example_function():
pass
assert example_function._required_scopes == []
@pytest.mark.unit
def test_insufficient_scope_error():
"""Test InsufficientScopeError exception structure."""
missing = ["notes:write", "calendar:write"]
error = InsufficientScopeError(missing)
assert error.missing_scopes == missing
assert "notes:write" in str(error)
assert "calendar:write" in str(error)
@pytest.mark.unit
def test_insufficient_scope_error_with_custom_message():
"""Test InsufficientScopeError with custom message."""
missing = ["files:write"]
custom_msg = "You need more permissions"
error = InsufficientScopeError(missing, custom_msg)
assert error.missing_scopes == missing
assert str(error) == custom_msg