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

feat(server): Experimental support for OAuth2/OIDC authentication

Browse Source
This commit is contained in:
Chris Coutinho
2025-10-14 01:22:15 +02:00
parent fafede2282
commit 4d7e4b9a4b
23 changed files with 2767 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/conftest.py
+235 -1
View File
@@ -1,8 +1,10 @@
import asyncio
import logging
import os
import uuid
from typing import Any, AsyncGenerator
import httpx
import pytest
from httpx import HTTPStatusError
from mcp import ClientSession
@@ -13,19 +15,71 @@ from nextcloud_mcp_server.client import NextcloudClient
logger = logging.getLogger(__name__)
async def wait_for_nextcloud(
host: str, max_attempts: int = 30, delay: float = 2.0
) -> bool:
"""
Wait for Nextcloud server to be ready by checking the status endpoint.
Args:
host: Nextcloud host URL
max_attempts: Maximum number of connection attempts
delay: Delay between attempts in seconds
Returns:
True if server is ready, False otherwise
"""
logger.info(f"Waiting for Nextcloud server at {host} to be ready...")
async with httpx.AsyncClient(timeout=5.0) as client:
for attempt in range(1, max_attempts + 1):
try:
# Try to hit the status endpoint
response = await client.get(f"{host}/status.php")
if response.status_code == 200:
data = response.json()
if data.get("installed"):
logger.info(
f"Nextcloud server is ready (version: {data.get('versionstring', 'unknown')})"
)
return True
except (httpx.RequestError, httpx.TimeoutException) as e:
logger.debug(f"Attempt {attempt}/{max_attempts}: {e}")
if attempt < max_attempts:
logger.info(
f"Nextcloud not ready yet, waiting {delay}s... (attempt {attempt}/{max_attempts})"
)
await asyncio.sleep(delay)
logger.error(
f"Nextcloud server at {host} did not become ready after {max_attempts} attempts"
)
return False
@pytest.fixture(scope="session")
async def nc_client() -> AsyncGenerator[NextcloudClient, Any]:
"""
Fixture to create a NextcloudClient instance for integration tests.
Uses environment variables for configuration.
Waits for Nextcloud to be ready before proceeding.
"""
assert os.getenv("NEXTCLOUD_HOST"), "NEXTCLOUD_HOST env var not set"
assert os.getenv("NEXTCLOUD_USERNAME"), "NEXTCLOUD_USERNAME env var not set"
assert os.getenv("NEXTCLOUD_PASSWORD"), "NEXTCLOUD_PASSWORD env var not set"
host = os.getenv("NEXTCLOUD_HOST")
# Wait for Nextcloud to be ready
if not await wait_for_nextcloud(host):
pytest.fail(f"Nextcloud server at {host} is not ready")
logger.info("Creating session-scoped NextcloudClient from environment variables.")
client = NextcloudClient.from_env()
# Optional: Perform a quick check like getting capabilities to ensure connection works
# Perform a quick check to ensure connection works
try:
await client.capabilities()
logger.info(
@@ -396,3 +450,183 @@ async def temporary_board_with_card(
)
except Exception as e:
logger.error(f"Unexpected error deleting temporary card {card.id}: {e}")
async def get_oauth_token(nextcloud_url: str, username: str, password: str) -> str:
"""
Get an OAuth access token from Nextcloud OIDC using Resource Owner Password flow.
This is a helper function for testing only - it bypasses the normal OAuth flow
to directly obtain a token for automated testing.
Args:
nextcloud_url: Nextcloud base URL
username: Nextcloud username
password: Nextcloud password
Returns:
Access token string
Raises:
Exception: If token acquisition fails
"""
from nextcloud_mcp_server.auth.client_registration import load_or_register_client
logger.info(f"Getting OAuth token for testing from {nextcloud_url}")
# Perform OIDC discovery
async with httpx.AsyncClient() as http_client:
discovery_url = f"{nextcloud_url}/.well-known/openid-configuration"
logger.debug(f"Fetching OIDC discovery from: {discovery_url}")
discovery_response = await http_client.get(discovery_url)
if discovery_response.status_code != 200:
raise Exception(f"OIDC discovery failed: {discovery_response.status_code}")
oidc_config = discovery_response.json()
token_endpoint = oidc_config.get("token_endpoint")
registration_endpoint = oidc_config.get("registration_endpoint")
if not token_endpoint or not registration_endpoint:
raise Exception("OIDC discovery missing required endpoints")
logger.debug(f"Token endpoint: {token_endpoint}")
logger.debug(f"Registration endpoint: {registration_endpoint}")
# Get or register an OAuth client
client_info = await load_or_register_client(
nextcloud_url=nextcloud_url,
registration_endpoint=registration_endpoint,
storage_path=".nextcloud_oauth_test_client.json",
redirect_uris=["http://localhost:8000/oauth/callback"],
)
# Use client credentials to get a token via password grant
# Note: This requires the OIDC app to support Resource Owner Password flow
token_response = await http_client.post(
token_endpoint,
data={
"grant_type": "password",
"client_id": client_info.client_id,
"client_secret": client_info.client_secret,
"username": username,
"password": password,
"scope": "openid profile email",
},
)
if token_response.status_code != 200:
logger.error(f"Failed to get OAuth token: {token_response.text}")
raise Exception(f"Token request failed: {token_response.status_code}")
token_data = token_response.json()
access_token = token_data.get("access_token")
if not access_token:
raise Exception("No access_token in response")
logger.info("Successfully obtained OAuth access token for testing")
return access_token
@pytest.fixture(scope="session")
async def oauth_token() -> str:
"""
Fixture to obtain an OAuth access token for integration tests.
This uses the Resource Owner Password flow to get a token without
requiring interactive browser authentication.
"""
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 token fixture requires NEXTCLOUD_HOST, USERNAME, and PASSWORD"
)
# Wait for Nextcloud to be ready
if not await wait_for_nextcloud(nextcloud_host):
pytest.fail(f"Nextcloud server at {nextcloud_host} is not ready")
try:
token = await get_oauth_token(nextcloud_host, username, password)
return token
except Exception as e:
logger.error(f"Failed to obtain OAuth token: {e}")
pytest.skip(f"Could not obtain OAuth token for testing: {e}")
@pytest.fixture(scope="session")
async def nc_oauth_client(oauth_token: str) -> AsyncGenerator[NextcloudClient, Any]:
"""
Fixture to create a NextcloudClient instance using OAuth authentication.
Uses the oauth_token fixture to get an access token.
"""
nextcloud_host = os.getenv("NEXTCLOUD_HOST")
username = os.getenv("NEXTCLOUD_USERNAME")
if not all([nextcloud_host, username]):
pytest.skip("OAuth client fixture requires NEXTCLOUD_HOST and USERNAME")
logger.info(f"Creating OAuth NextcloudClient for user: {username}")
client = NextcloudClient.from_token(
base_url=nextcloud_host,
token=oauth_token,
username=username,
)
# Verify the OAuth client works
try:
await client.capabilities()
logger.info("OAuth NextcloudClient initialized and capabilities checked.")
yield client
except Exception as e:
logger.error(f"Failed to initialize OAuth NextcloudClient: {e}")
pytest.fail(f"Failed to connect to Nextcloud with OAuth token: {e}")
finally:
await client.close()
@pytest.fixture(scope="session")
async def nc_mcp_oauth_client() -> AsyncGenerator[ClientSession, Any]:
"""
Fixture to create an MCP client session for OAuth integration tests.
Connects to the OAuth-enabled MCP server on port 8001.
"""
logger.info("Creating Streamable HTTP client for OAuth MCP server")
streamable_context = streamablehttp_client("http://127.0.0.1:8001/mcp")
session_context = None
try:
read_stream, write_stream, _ = await streamable_context.__aenter__()
session_context = ClientSession(read_stream, write_stream)
session = await session_context.__aenter__()
await session.initialize()
logger.info("OAuth MCP client session initialized successfully")
yield session
finally:
# Clean up in reverse order, ignoring task scope issues
if session_context is not None:
try:
await session_context.__aexit__(None, None, None)
except RuntimeError as e:
if "cancel scope" in str(e):
logger.debug(f"Ignoring cancel scope teardown issue: {e}")
else:
logger.warning(f"Error closing OAuth session: {e}")
except Exception as e:
logger.warning(f"Error closing OAuth session: {e}")
try:
await streamable_context.__aexit__(None, None, None)
except RuntimeError as e:
if "cancel scope" in str(e):
logger.debug(f"Ignoring cancel scope teardown issue: {e}")
else:
logger.warning(f"Error closing OAuth streamable HTTP client: {e}")
except Exception as e:
logger.warning(f"Error closing OAuth streamable HTTP client: {e}")
tests/integration/test_oauth.py
+126
View File
@@ -0,0 +1,126 @@
"""Integration tests for OAuth authentication."""
import logging
import pytest
from nextcloud_mcp_server.client import NextcloudClient
logger = logging.getLogger(__name__)
pytestmark = pytest.mark.integration
class TestOAuthClient:
"""Test OAuth-authenticated NextcloudClient."""
async def test_oauth_client_capabilities(self, nc_oauth_client: NextcloudClient):
"""Test that OAuth client can fetch capabilities."""
capabilities = await nc_oauth_client.capabilities()
assert capabilities is not None
assert "version" in capabilities
logger.info(
f"OAuth client successfully fetched capabilities: {capabilities.get('version')}"
)
async def test_oauth_client_notes_list(self, nc_oauth_client: NextcloudClient):
"""Test that OAuth client can list notes."""
notes = await nc_oauth_client.notes.get_notes()
assert isinstance(notes, list)
logger.info(f"OAuth client successfully listed {len(notes)} notes")
async def test_oauth_client_create_note(self, 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
class TestOAuthTokenValidation:
"""Test OAuth token validation and bearer auth."""
async def test_token_in_request_headers(
self, nc_oauth_client: NextcloudClient, oauth_token: str
):
"""Verify that bearer token is being used in requests."""
# The client should be using BearerAuth
assert nc_oauth_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(self):
"""Test that an invalid token results in authentication failure."""
import os
from nextcloud_mcp_server.auth import BearerAuth
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 the client should fail with 401
from httpx import HTTPStatusError
with pytest.raises(HTTPStatusError) as exc_info:
await invalid_client.capabilities()
assert exc_info.value.response.status_code == 401
await invalid_client.close()
logger.info("Invalid OAuth token correctly rejected")
class TestOAuthMCPIntegration:
"""Test OAuth integration with MCP server."""
@pytest.mark.skip(
reason="OAuth MCP server integration requires full OAuth flow implementation"
)
async def test_mcp_oauth_server_connection(self, nc_mcp_oauth_client):
"""Test connection to OAuth-enabled MCP server."""
# This test is currently skipped because the OAuth MCP server
# requires the full OAuth authorization flow to be implemented
# in the MCP SDK and app.py
# Once implemented, this test should:
# 1. Connect to the OAuth MCP server
# 2. Verify tools are available
# 3. Call a tool and verify it works with OAuth auth
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")