diff --git a/Dockerfile b/Dockerfile index 045ef68..f651b87 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,7 +1,9 @@ FROM ghcr.io/astral-sh/uv:0.9.7-python3.11-alpine@sha256:0006b77df7ebf46e68959fdc8d3af9d19f1adfae8c2e7e77907ad257e5d05be4 -# Install git (required for caldav dependency from git) -RUN apk add --no-cache git +# Install dependencies +# 1. git (required for caldav dependency from git) +# 2. sqlite for development with token db +RUN apk add --no-cache git sqlite WORKDIR /app diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index 462bbab..9fedef5 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -15,6 +15,7 @@ from mcp.server.auth.settings import AuthSettings from mcp.server.fastmcp import Context, FastMCP from pydantic import AnyHttpUrl from starlette.applications import Starlette +from starlette.middleware.authentication import AuthenticationMiddleware from starlette.middleware.cors import CORSMiddleware from starlette.responses import JSONResponse from starlette.routing import Mount, Route @@ -295,31 +296,19 @@ async def load_oauth_client_credentials( if registration_endpoint: logger.info("Dynamic client registration available") mcp_server_url = os.getenv("NEXTCLOUD_MCP_SERVER_URL", "http://localhost:8000") - redirect_uris = [f"{mcp_server_url}/oauth/callback"] + redirect_uris = [ + f"{mcp_server_url}/oauth/callback", # MCP OAuth flow + f"{mcp_server_url}/oauth/login-callback", # Browser OAuth flow for /user/page + ] - # Get scopes from environment or use defaults - # Note: Client registration happens BEFORE tools are registered, so we can't - # dynamically discover scopes here. These scopes define the "maximum allowed" - # scopes for this OAuth client. The actual per-tool scope enforcement happens - # via @require_scopes decorators, and the PRM endpoint advertises the actual - # supported scopes dynamically. + # MCP server DCR: Only request basic OIDC scopes for the server's own authentication + # Note: Nextcloud app scopes (notes:read, calendar:write, etc.) are for MCP *clients* + # that request access tokens. The MCP server itself only needs to authenticate + # as a client application, not request any Nextcloud resource access. # - # IMPORTANT: Keep this list in sync with all @require_scopes decorators - # when adding new apps, or set NEXTCLOUD_OIDC_SCOPES environment variable - # to override. - default_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" - ) - scopes = os.getenv("NEXTCLOUD_OIDC_SCOPES", default_scopes) + # The PRM endpoint will advertise the full list of supported scopes dynamically + # by discovering all @require_scopes decorators on registered tools. + dcr_scopes = "openid profile email" # Add offline_access scope if refresh tokens are enabled enable_offline_access = os.getenv("ENABLE_OFFLINE_ACCESS", "false").lower() in ( @@ -327,11 +316,11 @@ async def load_oauth_client_credentials( "1", "yes", ) - if enable_offline_access and "offline_access" not in scopes: - scopes = f"{scopes} offline_access" + if enable_offline_access: + dcr_scopes = f"{dcr_scopes} offline_access" logger.info("✓ offline_access scope enabled for refresh tokens") - logger.info(f"Requesting OAuth scopes: {scopes}") + logger.info(f"MCP server DCR scopes: {dcr_scopes}") # Get token type from environment (Bearer or jwt) # Note: Must be lowercase "jwt" to match OIDC app's check @@ -354,7 +343,7 @@ async def load_oauth_client_credentials( storage=storage, client_name=f"Nextcloud MCP Server ({token_type})", redirect_uris=redirect_uris, - scopes=scopes, + scopes=dcr_scopes, # Use DCR-specific scopes (basic OIDC only) token_type=token_type, ) @@ -892,6 +881,7 @@ def get_app(transport: str = "sse", enabled_apps: list[str] | None = None): app.state.oauth_context = { "storage": refresh_token_storage, "oauth_client": oauth_client, + "token_verifier": token_verifier, # For querying IdP userinfo endpoint "config": { "mcp_server_url": mcp_server_url, "discovery_url": discovery_url, @@ -1045,9 +1035,55 @@ def get_app(transport: str = "sse", enabled_apps: list[str] | None = None): "OAuth login routes enabled: /oauth/authorize, /oauth/callback, /oauth/token" ) + # Add browser OAuth login routes (OAuth mode only) + if oauth_enabled: + from nextcloud_mcp_server.auth.browser_oauth_routes import ( + oauth_login, + oauth_login_callback, + oauth_logout, + ) + + routes.append( + Route("/oauth/login", oauth_login, methods=["GET"], name="oauth_login") + ) + routes.append( + Route( + "/oauth/login-callback", + oauth_login_callback, + methods=["GET"], + name="oauth_login_callback", + ) + ) + routes.append( + Route("/oauth/logout", oauth_logout, methods=["GET"], name="oauth_logout") + ) + logger.info( + "Browser OAuth routes enabled: /oauth/login, /oauth/login-callback, /oauth/logout" + ) + + # Add user info routes (available in both BasicAuth and OAuth modes) + from nextcloud_mcp_server.auth.userinfo_routes import ( + user_info_html, + user_info_json, + ) + + routes.append(Route("/user", user_info_json, methods=["GET"])) + routes.append(Route("/user/page", user_info_html, methods=["GET"])) + logger.info("User info routes enabled: /user (JSON), /user/page (HTML)") + routes.append(Mount("/", app=mcp_app)) app = Starlette(routes=routes, lifespan=starlette_lifespan) + # Add authentication middleware for browser-based routes + from nextcloud_mcp_server.auth.session_backend import SessionAuthBackend + + # SessionAuthBackend will look up oauth_context from app.state at runtime + app.add_middleware( + AuthenticationMiddleware, + backend=SessionAuthBackend(oauth_enabled=oauth_enabled), + ) + logger.info("Authentication middleware enabled for browser routes") + # Add CORS middleware to allow browser-based clients like MCP Inspector app.add_middleware( CORSMiddleware, diff --git a/nextcloud_mcp_server/auth/browser_oauth_routes.py b/nextcloud_mcp_server/auth/browser_oauth_routes.py new file mode 100644 index 0000000..1f932f8 --- /dev/null +++ b/nextcloud_mcp_server/auth/browser_oauth_routes.py @@ -0,0 +1,358 @@ +"""Browser-based OAuth login routes for admin UI. + +Separate from MCP OAuth flow - these routes establish browser sessions +for accessing admin UI endpoints like /user/page. +""" + +import logging +import os +import secrets +from urllib.parse import urlencode + +import httpx +import jwt +from starlette.requests import Request +from starlette.responses import HTMLResponse, JSONResponse, RedirectResponse + +logger = logging.getLogger(__name__) + + +async def oauth_login(request: Request) -> RedirectResponse | JSONResponse: + """Browser OAuth login endpoint - redirects to IdP for authentication. + + This is separate from the MCP OAuth flow (/oauth/authorize). + Creates a browser session with refresh token for admin UI access. + + Query parameters: + next: Optional URL to redirect to after login (default: /user/page) + + Returns: + 302 redirect to IdP authorization endpoint + """ + oauth_ctx = request.app.state.oauth_context + if not oauth_ctx: + # BasicAuth mode - no login needed, redirect to user page + return RedirectResponse("/user/page", status_code=302) + + storage = oauth_ctx["storage"] + oauth_client = oauth_ctx["oauth_client"] + oauth_config = oauth_ctx["config"] + + # Debug: Log oauth_config contents + logger.info(f"oauth_login called - oauth_config keys: {oauth_config.keys()}") + logger.info(f"oauth_login called - client_id: {oauth_config.get('client_id')}") + logger.info(f"oauth_login called - oauth_client: {oauth_client is not None}") + + # Generate state for CSRF protection + state = secrets.token_urlsafe(32) + + # Build OAuth authorization URL + mcp_server_url = oauth_config["mcp_server_url"] + callback_uri = f"{mcp_server_url}/oauth/login-callback" + + # Request only basic OIDC scopes for browser session + # Note: Nextcloud app scopes (notes:read, etc.) are for MCP client access tokens, + # not for the MCP server's own browser authentication + scopes = "openid profile email offline_access" + + code_challenge = "" + code_verifier = "" + + if oauth_client: + # External IdP mode (Keycloak) + # Keycloak requires PKCE, so generate code_verifier and code_challenge + if not oauth_client.authorization_endpoint: + await oauth_client.discover() + + # Generate PKCE values + code_verifier, code_challenge = oauth_client.generate_pkce_challenge() + + # Store code_verifier temporarily (using state as key) + # We'll retrieve it in the callback using the state parameter + await storage.store_oauth_session( + session_id=state, # Use state as session ID + client_id="browser-ui", + client_redirect_uri="/user/page", + state=state, + code_challenge=code_challenge, + code_challenge_method="S256", + mcp_authorization_code=code_verifier, # Store code_verifier here temporarily + flow_type="browser", + ttl_seconds=600, # 10 minutes + ) + + idp_params = { + "client_id": oauth_client.client_id, + "redirect_uri": callback_uri, + "response_type": "code", + "scope": scopes, + "state": state, + "code_challenge": code_challenge, + "code_challenge_method": "S256", + "prompt": "consent", # Ensure refresh token + } + + auth_url = f"{oauth_client.authorization_endpoint}?{urlencode(idp_params)}" + logger.info(f"Redirecting to external IdP login: {auth_url.split('?')[0]}") + else: + # Integrated mode (Nextcloud OIDC) + discovery_url = oauth_config.get("discovery_url") + if not discovery_url: + return JSONResponse( + { + "error": "server_error", + "error_description": "OAuth discovery URL not configured", + }, + status_code=500, + ) + + # Fetch authorization endpoint + async with httpx.AsyncClient() as http_client: + response = await http_client.get(discovery_url) + response.raise_for_status() + discovery = response.json() + authorization_endpoint = discovery["authorization_endpoint"] + + # Replace internal Docker hostname with public URL + public_issuer = os.getenv("NEXTCLOUD_PUBLIC_ISSUER_URL") + if public_issuer: + from urllib.parse import urlparse as parse_url + + internal_parsed = parse_url(oauth_config["nextcloud_host"]) + auth_parsed = parse_url(authorization_endpoint) + + if auth_parsed.hostname == internal_parsed.hostname: + public_parsed = parse_url(public_issuer) + authorization_endpoint = ( + f"{public_parsed.scheme}://{public_parsed.netloc}{auth_parsed.path}" + ) + + idp_params = { + "client_id": oauth_config["client_id"], + "redirect_uri": callback_uri, + "response_type": "code", + "scope": scopes, + "state": state, + "prompt": "consent", # Ensure refresh token + } + + # Debug: Log full parameters + logger.info(f"Building Nextcloud OIDC auth URL with params: {idp_params}") + + auth_url = f"{authorization_endpoint}?{urlencode(idp_params)}" + logger.info(f"Redirecting to Nextcloud OIDC login: {auth_url}") + + return RedirectResponse(auth_url, status_code=302) + + +async def oauth_login_callback(request: Request) -> RedirectResponse | HTMLResponse: + """Browser OAuth callback - IdP redirects here after authentication. + + Exchanges authorization code for tokens, stores refresh token, + sets session cookie, and redirects to original destination. + + Query parameters: + code: Authorization code from IdP + state: State parameter + error: Error code (if authorization failed) + + Returns: + 302 redirect to next URL with session cookie + """ + # Check for errors + error = request.query_params.get("error") + if error: + error_description = request.query_params.get( + "error_description", "Authorization failed" + ) + logger.error(f"OAuth login error: {error} - {error_description}") + login_url = str(request.url_for("oauth_login")) + return HTMLResponse( + f""" + + + Login Failed + +

Login Failed

+

Error: {error}

+

{error_description}

+

Try again

+ + + """, + status_code=400, + ) + + # Extract code and state + code = request.query_params.get("code") + state = request.query_params.get("state") + + if not code or not state: + return HTMLResponse( + """ + + + Invalid Request + +

Invalid Request

+

Missing code or state parameter

+ + + """, + status_code=400, + ) + + # Get OAuth context + oauth_ctx = request.app.state.oauth_context + storage = oauth_ctx["storage"] + oauth_client = oauth_ctx["oauth_client"] + oauth_config = oauth_ctx["config"] + + # Retrieve code_verifier from session storage (if using PKCE) + code_verifier = "" + if oauth_client: + # For Keycloak (external IdP), we stored the code_verifier in the session + oauth_session = await storage.get_oauth_session(state) + if oauth_session: + # code_verifier was stored in mcp_authorization_code field + code_verifier = oauth_session.get("mcp_authorization_code", "") + # Clean up the temporary session + # Note: We don't have delete_oauth_session method, but it will expire after TTL + + # Exchange authorization code for tokens + mcp_server_url = oauth_config["mcp_server_url"] + callback_uri = f"{mcp_server_url}/oauth/login-callback" + + try: + if oauth_client: + # External IdP mode (Keycloak) + # Use PKCE if we have a code_verifier + if not oauth_client.token_endpoint: + await oauth_client.discover() + + token_params = { + "grant_type": "authorization_code", + "code": code, + "redirect_uri": callback_uri, + "client_id": oauth_client.client_id, + "client_secret": oauth_client.client_secret, + } + + # Add code_verifier if we have one (PKCE) + if code_verifier: + token_params["code_verifier"] = code_verifier + + async with httpx.AsyncClient() as http_client: + response = await http_client.post( + oauth_client.token_endpoint, + data=token_params, + ) + response.raise_for_status() + token_data = response.json() + else: + # Integrated mode (Nextcloud OIDC) + discovery_url = oauth_config.get("discovery_url") + async with httpx.AsyncClient() as http_client: + response = await http_client.get(discovery_url) + response.raise_for_status() + discovery = response.json() + token_endpoint = discovery["token_endpoint"] + + async with httpx.AsyncClient() as http_client: + response = await http_client.post( + token_endpoint, + data={ + "grant_type": "authorization_code", + "code": code, + "redirect_uri": callback_uri, + "client_id": oauth_config["client_id"], + "client_secret": oauth_config["client_secret"], + }, + ) + response.raise_for_status() + token_data = response.json() + + except Exception as e: + logger.error(f"Token exchange failed: {e}") + return HTMLResponse( + f""" + + + Login Failed + +

Login Failed

+

Failed to exchange authorization code for tokens

+

Error: {e}

+ + + """, + status_code=500, + ) + + refresh_token = token_data.get("refresh_token") + id_token = token_data.get("id_token") + + logger.info(f"Token exchange response keys: {token_data.keys()}") + logger.info(f"Refresh token present: {refresh_token is not None}") + logger.info(f"ID token present: {id_token is not None}") + + # Decode ID token to get user info + try: + userinfo = jwt.decode(id_token, options={"verify_signature": False}) + user_id = userinfo.get("sub") + username = userinfo.get("preferred_username") or userinfo.get("email") + logger.info(f"Browser login successful: {username} (sub={user_id})") + except Exception as e: + logger.warning(f"Failed to decode ID token: {e}") + user_id = f"user-{secrets.token_hex(8)}" + username = "unknown" + + # Store refresh token + if refresh_token: + logger.info(f"Storing refresh token for user_id: {user_id}") + await storage.store_refresh_token( + user_id=user_id, + refresh_token=refresh_token, + expires_at=None, + flow_type="browser", # Browser-based login flow + ) + logger.info(f"✓ Refresh token stored successfully for user_id: {user_id}") + else: + logger.warning("No refresh token in token response - cannot store session") + + # Create response and set session cookie + response = RedirectResponse("/user/page", status_code=302) + response.set_cookie( + key="mcp_session", + value=user_id, + max_age=86400 * 30, # 30 days + httponly=True, + secure=False, # Set to True in production with HTTPS + samesite="lax", + ) + + logger.info(f"Session cookie set for user: {username}") + return response + + +async def oauth_logout(request: Request) -> RedirectResponse: + """Browser OAuth logout - clears session cookie. + + Query parameters: + next: Optional URL to redirect to after logout (default: /oauth/login) + + Returns: + 302 redirect with cleared session cookie + """ + next_url = request.query_params.get("next", "/oauth/login") + + # TODO: Optionally revoke refresh token from storage + # session_id = request.cookies.get("mcp_session") + # if session_id: + # await storage.delete_refresh_token(session_id) + + response = RedirectResponse(next_url, status_code=302) + response.delete_cookie("mcp_session") + + logger.info("User logged out, session cookie cleared") + return response diff --git a/nextcloud_mcp_server/auth/session_backend.py b/nextcloud_mcp_server/auth/session_backend.py new file mode 100644 index 0000000..f702ee0 --- /dev/null +++ b/nextcloud_mcp_server/auth/session_backend.py @@ -0,0 +1,92 @@ +"""Session-based authentication backend for Starlette routes. + +Provides browser-based authentication for admin UI routes, separate from +MCP's OAuth authentication flow. +""" + +import logging +import os + +from starlette.authentication import ( + AuthCredentials, + AuthenticationBackend, + SimpleUser, +) +from starlette.requests import HTTPConnection + +logger = logging.getLogger(__name__) + + +class SessionAuthBackend(AuthenticationBackend): + """Authentication backend using signed session cookies. + + For BasicAuth mode: Always authenticates as the configured user. + For OAuth mode: Checks for valid session cookie with stored refresh token. + """ + + def __init__(self, oauth_enabled: bool = False): + """Initialize session authentication backend. + + Args: + oauth_enabled: Whether OAuth mode is enabled + """ + self.oauth_enabled = oauth_enabled + + async def authenticate( + self, conn: HTTPConnection + ) -> tuple[AuthCredentials, SimpleUser] | None: + """Authenticate the request based on session cookie or BasicAuth mode. + + Args: + conn: HTTP connection + + Returns: + Tuple of (credentials, user) if authenticated, None otherwise + """ + # BasicAuth mode: Always authenticated as the configured user + if not self.oauth_enabled: + username = os.getenv("NEXTCLOUD_USERNAME", "admin") + return AuthCredentials(["authenticated", "admin"]), SimpleUser(username) + + # OAuth mode: Check for session cookie + session_id = conn.cookies.get("mcp_session") + logger.info( + f"Session authentication check - cookie present: {session_id is not None}, path: {conn.url.path}" + ) + if not session_id: + logger.info("No session cookie found - redirecting to login") + return None + + logger.info(f"Found session cookie: {session_id[:16]}...") + + # Get OAuth context from app state + oauth_context = getattr(conn.app.state, "oauth_context", None) + if not oauth_context: + logger.warning("OAuth context not available in app state") + return None + + # Validate session + storage = oauth_context.get("storage") + if not storage: + logger.warning("OAuth storage not available") + return None + + try: + # Check if user has refresh token (indicates logged-in session) + logger.info(f"Looking up refresh token for session: {session_id[:16]}...") + token_data = await storage.get_refresh_token(session_id) + if not token_data: + logger.warning( + f"No refresh token found for session {session_id[:16]}..." + ) + return None + + # Session is valid - use session_id (which is user_id from ID token) as username + username = session_id + logger.info(f"✓ Session authenticated successfully: {username[:16]}...") + + return AuthCredentials(["authenticated"]), SimpleUser(username) + + except Exception as e: + logger.warning(f"Session validation error: {e}") + return None diff --git a/nextcloud_mcp_server/auth/userinfo_routes.py b/nextcloud_mcp_server/auth/userinfo_routes.py new file mode 100644 index 0000000..5582606 --- /dev/null +++ b/nextcloud_mcp_server/auth/userinfo_routes.py @@ -0,0 +1,442 @@ +"""User info routes for the MCP server admin UI. + +Provides browser-based endpoints to view information about the currently +authenticated user. Uses session-based authentication with OAuth flow. + +For BasicAuth mode: Shows configured user info (no login needed). +For OAuth mode: Requires browser-based OAuth login to establish session. +""" + +import logging +import os +from typing import Any + +import httpx +from starlette.authentication import requires +from starlette.requests import Request +from starlette.responses import HTMLResponse, JSONResponse + +logger = logging.getLogger(__name__) + + +async def _query_idp_userinfo( + access_token_str: str, userinfo_uri: str +) -> dict[str, Any] | None: + """Query the IdP's userinfo endpoint. + + Args: + access_token_str: The access token string + userinfo_uri: The userinfo endpoint URI + + Returns: + User info dictionary from IdP, or None if query fails + """ + try: + async with httpx.AsyncClient(timeout=10.0) as client: + response = await client.get( + userinfo_uri, + headers={"Authorization": f"Bearer {access_token_str}"}, + ) + response.raise_for_status() + return response.json() + except Exception as e: + logger.warning(f"Failed to query IdP userinfo endpoint: {e}") + return None + + +async def _get_user_info(request: Request) -> dict[str, Any]: + """Get user information for the currently authenticated user. + + Args: + request: Starlette request object (must be authenticated) + + Returns: + Dictionary containing user information + """ + username = request.user.display_name + oauth_ctx = getattr(request.app.state, "oauth_context", None) + + # BasicAuth mode + if not oauth_ctx: + return { + "username": username, + "auth_mode": "basic", + "nextcloud_host": os.getenv("NEXTCLOUD_HOST", "unknown"), + } + + # OAuth mode - get user's refresh token and current access token + storage = oauth_ctx.get("storage") + session_id = request.cookies.get("mcp_session") + + if not storage or not session_id: + return { + "error": "Session not found", + "username": username, + "auth_mode": "oauth", + } + + try: + # Get refresh token data + token_data = await storage.get_refresh_token(session_id) + if not token_data: + return { + "error": "No refresh token found", + "username": username, + "auth_mode": "oauth", + } + + refresh_token = token_data.get("refresh_token") + + # Exchange refresh token for fresh access token + oauth_client = oauth_ctx.get("oauth_client") + oauth_config = oauth_ctx.get("config") + + if oauth_client: + # External IdP mode (Keycloak) + # Create fresh HTTP client to avoid event loop issues + if not oauth_client.token_endpoint: + await oauth_client.discover() + + async with httpx.AsyncClient(timeout=30.0) as http_client: + response = await http_client.post( + oauth_client.token_endpoint, + data={ + "grant_type": "refresh_token", + "refresh_token": refresh_token, + }, + auth=(oauth_client.client_id, oauth_client.client_secret), + ) + response.raise_for_status() + token_response = response.json() + access_token = token_response["access_token"] + else: + # Integrated mode (Nextcloud OIDC) + # Note: This is server-side code, so we use internal Docker hostnames + # (not public URLs) for server-to-server communication + discovery_url = oauth_config.get("discovery_url") + logger.info(f"Querying discovery URL: {discovery_url}") + + async with httpx.AsyncClient() as http_client: + response = await http_client.get(discovery_url) + response.raise_for_status() + discovery = response.json() + token_endpoint = discovery["token_endpoint"] + logger.info( + f"Using token endpoint for server-side refresh: {token_endpoint}" + ) + + async with httpx.AsyncClient() as http_client: + response = await http_client.post( + token_endpoint, + data={ + "grant_type": "refresh_token", + "refresh_token": refresh_token, + "client_id": oauth_config["client_id"], + "client_secret": oauth_config["client_secret"], + }, + ) + response.raise_for_status() + token_response = response.json() + access_token = token_response["access_token"] + + # Build basic user context + user_context = { + "username": username, # From request.user.display_name + "auth_mode": "oauth", + "session_id": session_id[:16] + "...", # Truncated for security + } + + # Query IdP userinfo for enhanced profile + token_verifier = oauth_ctx.get("token_verifier") + if token_verifier and hasattr(token_verifier, "userinfo_uri"): + idp_profile = await _query_idp_userinfo( + access_token, token_verifier.userinfo_uri + ) + if idp_profile: + user_context["idp_profile"] = idp_profile + else: + user_context["idp_profile_error"] = ( + "Failed to retrieve profile from IdP" + ) + + return user_context + + except Exception as e: + import traceback + + logger.error(f"Error retrieving user info: {e}") + logger.error(f"Traceback: {traceback.format_exc()}") + return { + "error": f"Failed to retrieve user info: {e}", + "username": username, + "auth_mode": "oauth", + } + + +@requires("authenticated", redirect="oauth_login") +async def user_info_json(request: Request) -> JSONResponse: + """User info endpoint - returns JSON with current user information. + + Requires authentication via session cookie (redirects to oauth_login route if not authenticated). + + Args: + request: Starlette request object + + Returns: + JSON response with user information + """ + user_info = await _get_user_info(request) + return JSONResponse(user_info) + + +@requires("authenticated", redirect="oauth_login") +async def user_info_html(request: Request) -> HTMLResponse: + """User info page - returns HTML with current user information. + + Requires authentication via session cookie (redirects to oauth_login route if not authenticated). + + Args: + request: Starlette request object + + Returns: + HTML response with formatted user information + """ + user_context = await _get_user_info(request) + + # Check for error + if "error" in user_context and user_context["error"] != "": + # Get login URL dynamically + oauth_ctx = getattr(request.app.state, "oauth_context", None) + login_url = str(request.url_for("oauth_login")) if oauth_ctx else "/oauth/login" + + error_html = f""" + + + + + + Error - Nextcloud MCP Server + + + +
+

Error Retrieving User Info

+
+ Error: {user_context["error"]} +
+

Login again

+
+ + + """ + return HTMLResponse(content=error_html) + + # Build HTML response + auth_mode = user_context.get("auth_mode", "unknown") + username = user_context.get("username", "unknown") + + # Get logout URL dynamically for OAuth mode + logout_url = "" + if auth_mode == "oauth": + oauth_ctx = getattr(request.app.state, "oauth_context", None) + logout_url = ( + str(request.url_for("oauth_logout")) if oauth_ctx else "/oauth/logout" + ) + + # Build host info HTML (BasicAuth only) + host_info_html = "" + if auth_mode == "basic": + nextcloud_host = user_context.get("nextcloud_host", "unknown") + host_info_html = f""" +

Connection

+ + + + + +
Nextcloud Host{nextcloud_host}
+ """ + + # Build session info HTML (OAuth only) + session_info_html = "" + if auth_mode == "oauth" and "session_id" in user_context: + session_id = user_context.get("session_id", "unknown") + session_info_html = f""" +

Session Information

+ + + + + +
Session ID{session_id}
+ """ + + # Build IdP profile HTML + idp_profile_html = "" + if "idp_profile" in user_context: + idp_profile = user_context["idp_profile"] + idp_profile_html = "

Identity Provider Profile

" + for key, value in idp_profile.items(): + # Handle list values + if isinstance(value, list): + value_str = ", ".join(str(v) for v in value) + else: + value_str = str(value) + idp_profile_html += f""" + + + + + """ + idp_profile_html += "
{key}{value_str}
" + elif "idp_profile_error" in user_context: + idp_profile_html = f""" +

Identity Provider Profile

+
{user_context["idp_profile_error"]}
+ """ + + html_content = f""" + + + + + + User Info - Nextcloud MCP Server + + + +
+

Nextcloud MCP Server - User Info

+ +

Authentication

+ + + + + + + + + +
Username{username}
Authentication Mode{auth_mode}
+ + {host_info_html} + {session_info_html} + {idp_profile_html} + + {f'
Logout
' if auth_mode == "oauth" else ""} +
+ + + """ + + return HTMLResponse(content=html_content) diff --git a/tests/server/auth/__init__.py b/tests/server/auth/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/server/auth/test_userinfo_routes.py b/tests/server/auth/test_userinfo_routes.py new file mode 100644 index 0000000..5554bd0 --- /dev/null +++ b/tests/server/auth/test_userinfo_routes.py @@ -0,0 +1,333 @@ +"""Unit tests for user info routes.""" + +from unittest.mock import AsyncMock, Mock + +import pytest + +from nextcloud_mcp_server.auth.userinfo_routes import ( + _get_user_context, + _query_idp_userinfo, + user_info_html, + user_info_json, +) + +pytestmark = pytest.mark.unit + + +@pytest.mark.asyncio +async def test_query_idp_userinfo_success(mocker): + """Test successful IdP userinfo query.""" + mock_response = Mock() + mock_response.json.return_value = { + "sub": "alice", + "email": "alice@example.com", + "name": "Alice Smith", + } + mock_response.raise_for_status = Mock() + + mock_client = AsyncMock() + mock_client.get.return_value = mock_response + + mocker.patch("httpx.AsyncClient", return_value=mock_client) + + result = await _query_idp_userinfo("test_token", "https://example.com/userinfo") + + assert result == { + "sub": "alice", + "email": "alice@example.com", + "name": "Alice Smith", + } + mock_client.get.assert_called_once_with( + "https://example.com/userinfo", + headers={"Authorization": "Bearer test_token"}, + ) + + +@pytest.mark.asyncio +async def test_query_idp_userinfo_failure(mocker): + """Test IdP userinfo query failure handling.""" + mock_client = AsyncMock() + mock_client.get.side_effect = Exception("Network error") + + mocker.patch("httpx.AsyncClient", return_value=mock_client) + + result = await _query_idp_userinfo("test_token", "https://example.com/userinfo") + + assert result is None + + +@pytest.mark.asyncio +async def test_get_user_context_basic_auth(monkeypatch): + """Test get_user_context in BasicAuth mode.""" + monkeypatch.setenv("NEXTCLOUD_USERNAME", "testuser") + monkeypatch.setenv("NEXTCLOUD_HOST", "https://cloud.example.com") + + mock_request = Mock() + oauth_ctx = None # BasicAuth mode + + result = await _get_user_context(mock_request, oauth_ctx) + + assert result["username"] == "testuser" + assert result["auth_mode"] == "basic" + assert result["nextcloud_host"] == "https://cloud.example.com" + + +@pytest.mark.asyncio +async def test_get_user_context_oauth_no_token(): + """Test get_user_context in OAuth mode without token.""" + mock_request = Mock() + mock_request.user = Mock(spec=[]) # No access_token attribute + oauth_ctx = {"token_verifier": Mock()} + + result = await _get_user_context(mock_request, oauth_ctx) + + assert "error" in result + assert result["error"] == "Not authenticated" + assert result["auth_mode"] == "oauth" + + +@pytest.mark.asyncio +async def test_get_user_context_oauth_with_token_no_idp_query(mocker): + """Test get_user_context in OAuth mode with token but no IdP query.""" + mock_access_token = Mock() + mock_access_token.resource = "alice" + mock_access_token.client_id = "mcp_client_123" + mock_access_token.scopes = ["notes:read", "calendar:write"] + mock_access_token.expires_at = 1730678400 + mock_access_token.token = "test_token" + + mock_request = Mock() + mock_request.user = Mock() + mock_request.user.access_token = mock_access_token + + # OAuth context without token_verifier + oauth_ctx = {} + + result = await _get_user_context(mock_request, oauth_ctx) + + assert result["username"] == "alice" + assert result["auth_mode"] == "oauth" + assert result["client_id"] == "mcp_client_123" + assert result["scopes"] == ["notes:read", "calendar:write"] + assert result["token_expires_at"] == 1730678400 + assert "idp_profile" not in result + + +@pytest.mark.asyncio +async def test_get_user_context_oauth_with_idp_query_success(mocker): + """Test get_user_context in OAuth mode with successful IdP query.""" + mock_access_token = Mock() + mock_access_token.resource = "alice" + mock_access_token.client_id = "mcp_client_123" + mock_access_token.scopes = ["notes:read"] + mock_access_token.expires_at = 1730678400 + mock_access_token.token = "test_token" + + mock_request = Mock() + mock_request.user = Mock() + mock_request.user.access_token = mock_access_token + + mock_token_verifier = Mock() + mock_token_verifier.userinfo_uri = "https://example.com/userinfo" + oauth_ctx = {"token_verifier": mock_token_verifier} + + # Mock IdP response + idp_profile = { + "sub": "alice", + "email": "alice@example.com", + "name": "Alice Smith", + } + mocker.patch( + "nextcloud_mcp_server.auth.userinfo_routes._query_idp_userinfo", + return_value=idp_profile, + ) + + result = await _get_user_context(mock_request, oauth_ctx) + + assert result["username"] == "alice" + assert result["auth_mode"] == "oauth" + assert result["idp_profile"] == idp_profile + + +@pytest.mark.asyncio +async def test_get_user_context_oauth_with_idp_query_failure(mocker): + """Test get_user_context in OAuth mode with failed IdP query.""" + mock_access_token = Mock() + mock_access_token.resource = "alice" + mock_access_token.client_id = "mcp_client_123" + mock_access_token.scopes = ["notes:read"] + mock_access_token.expires_at = 1730678400 + mock_access_token.token = "test_token" + + mock_request = Mock() + mock_request.user = Mock() + mock_request.user.access_token = mock_access_token + + mock_token_verifier = Mock() + mock_token_verifier.userinfo_uri = "https://example.com/userinfo" + oauth_ctx = {"token_verifier": mock_token_verifier} + + # Mock IdP failure + mocker.patch( + "nextcloud_mcp_server.auth.userinfo_routes._query_idp_userinfo", + return_value=None, + ) + + result = await _get_user_context(mock_request, oauth_ctx) + + assert result["username"] == "alice" + assert result["auth_mode"] == "oauth" + assert "idp_profile_error" in result + assert result["idp_profile_error"] == "Failed to retrieve profile from IdP" + + +@pytest.mark.asyncio +async def test_user_info_json_basic_auth(mocker, monkeypatch): + """Test user_info_json endpoint in BasicAuth mode.""" + monkeypatch.setenv("NEXTCLOUD_USERNAME", "admin") + monkeypatch.setenv("NEXTCLOUD_HOST", "https://cloud.example.com") + + mock_request = Mock() + mock_request.app = Mock() + mock_request.app.state = Mock() + mock_request.app.state.oauth_context = None + + response = await user_info_json(mock_request) + + assert response.status_code == 200 + body = response.body.decode() + assert "admin" in body + assert "basic" in body + + +@pytest.mark.asyncio +async def test_user_info_json_oauth_unauthenticated(mocker): + """Test user_info_json endpoint in OAuth mode without authentication.""" + mock_request = Mock() + mock_request.app = Mock() + mock_request.app.state = Mock() + mock_request.app.state.oauth_context = {"token_verifier": Mock()} + mock_request.user = Mock(spec=[]) # No access_token + + response = await user_info_json(mock_request) + + assert response.status_code == 401 + body = response.body.decode() + assert "error" in body + + +@pytest.mark.asyncio +async def test_user_info_json_oauth_authenticated(mocker): + """Test user_info_json endpoint in OAuth mode with authentication.""" + mock_access_token = Mock() + mock_access_token.resource = "alice" + mock_access_token.client_id = "mcp_client_123" + mock_access_token.scopes = ["notes:read", "calendar:write"] + mock_access_token.expires_at = 1730678400 + mock_access_token.token = "test_token" + + mock_request = Mock() + mock_request.app = Mock() + mock_request.app.state = Mock() + mock_request.app.state.oauth_context = {"token_verifier": Mock()} + mock_request.user = Mock() + mock_request.user.access_token = mock_access_token + + response = await user_info_json(mock_request) + + assert response.status_code == 200 + body = response.body.decode() + assert "alice" in body + assert "oauth" in body + assert "mcp_client_123" in body + + +@pytest.mark.asyncio +async def test_user_info_html_basic_auth(mocker, monkeypatch): + """Test user_info_html endpoint in BasicAuth mode.""" + monkeypatch.setenv("NEXTCLOUD_USERNAME", "admin") + monkeypatch.setenv("NEXTCLOUD_HOST", "https://cloud.example.com") + + mock_request = Mock() + mock_request.app = Mock() + mock_request.app.state = Mock() + mock_request.app.state.oauth_context = None + + response = await user_info_html(mock_request) + + assert response.status_code == 200 + body = response.body.decode() + assert "" in body + assert "admin" in body + assert "basic" in body.lower() + + +@pytest.mark.asyncio +async def test_user_info_html_oauth_unauthenticated(mocker): + """Test user_info_html endpoint in OAuth mode without authentication.""" + mock_request = Mock() + mock_request.app = Mock() + mock_request.app.state = Mock() + mock_request.app.state.oauth_context = {"token_verifier": Mock()} + mock_request.user = Mock(spec=[]) # No access_token + + response = await user_info_html(mock_request) + + assert response.status_code == 401 + body = response.body.decode() + assert "" in body + assert "Authentication Required" in body + + +@pytest.mark.asyncio +async def test_user_info_html_oauth_authenticated(mocker): + """Test user_info_html endpoint in OAuth mode with authentication.""" + mock_access_token = Mock() + mock_access_token.resource = "bob" + mock_access_token.client_id = "mcp_client_456" + mock_access_token.scopes = ["notes:write"] + mock_access_token.expires_at = 1730678400 + mock_access_token.token = "test_token" + + mock_request = Mock() + mock_request.app = Mock() + mock_request.app.state = Mock() + mock_request.app.state.oauth_context = {"token_verifier": Mock()} + mock_request.user = Mock() + mock_request.user.access_token = mock_access_token + + response = await user_info_html(mock_request) + + assert response.status_code == 200 + body = response.body.decode() + assert "" in body + assert "bob" in body + assert "oauth" in body.lower() + assert "mcp_client_456" in body + + +@pytest.mark.asyncio +async def test_user_info_html_with_scopes(mocker): + """Test user_info_html displays scopes correctly.""" + mock_access_token = Mock() + mock_access_token.resource = "charlie" + mock_access_token.client_id = "mcp_client_789" + mock_access_token.scopes = ["notes:read", "notes:write", "calendar:read"] + mock_access_token.expires_at = 1730678400 + mock_access_token.token = "test_token" + + mock_request = Mock() + mock_request.app = Mock() + mock_request.app.state = Mock() + mock_request.app.state.oauth_context = {"token_verifier": Mock()} + mock_request.user = Mock() + mock_request.user.access_token = mock_access_token + + response = await user_info_html(mock_request) + + assert response.status_code == 200 + body = response.body.decode() + assert "notes:read" in body + assert "notes:write" in body + assert "calendar:read" in body + assert "

Scopes

" in body diff --git a/tests/server/oauth/test_userinfo_integration.py b/tests/server/oauth/test_userinfo_integration.py new file mode 100644 index 0000000..6c81c9b --- /dev/null +++ b/tests/server/oauth/test_userinfo_integration.py @@ -0,0 +1,307 @@ +"""OAuth integration tests for user info routes. + +Tests verify: +1. /user endpoint returns correct user info in OAuth mode +2. /user/page endpoint renders HTML correctly in OAuth mode +3. Endpoints return 401 when not authenticated +4. Integration with Nextcloud OIDC and Keycloak IdP +""" + +import json +import logging +import os + +import httpx +import pytest + +logger = logging.getLogger(__name__) + +pytestmark = [pytest.mark.integration, pytest.mark.oauth] + + +# ============================================================================ +# Helper Functions +# ============================================================================ + + +async def get_user_info_json(access_token: str, port: int = 8001) -> dict: + """Call /user endpoint with OAuth token. + + Args: + access_token: OAuth access token + port: MCP server port (8001 for mcp-oauth, 8002 for mcp-keycloak) + + Returns: + JSON response data + """ + async with httpx.AsyncClient() as client: + response = await client.get( + f"http://localhost:{port}/user", + headers={"Authorization": f"Bearer {access_token}"}, + ) + response.raise_for_status() + return response.json() + + +async def get_user_info_html(access_token: str, port: int = 8001) -> str: + """Call /user/page endpoint with OAuth token. + + Args: + access_token: OAuth access token + port: MCP server port (8001 for mcp-oauth, 8002 for mcp-keycloak) + + Returns: + HTML response text + """ + async with httpx.AsyncClient() as client: + response = await client.get( + f"http://localhost:{port}/user/page", + headers={"Authorization": f"Bearer {access_token}"}, + ) + response.raise_for_status() + return response.text + + +# ============================================================================ +# Nextcloud OAuth Tests (mcp-oauth on port 8001) +# ============================================================================ + + +async def test_user_info_json_with_nextcloud_oauth(playwright_oauth_token): + """Test /user endpoint with Nextcloud OAuth token.""" + user_info = await get_user_info_json(playwright_oauth_token, port=8001) + + # Verify response structure + assert "username" in user_info + assert "auth_mode" in user_info + assert user_info["auth_mode"] == "oauth" + + # Verify OAuth-specific fields + assert "client_id" in user_info + assert "scopes" in user_info + assert "token_expires_at" in user_info + assert isinstance(user_info["scopes"], list) + + # Verify username matches environment + expected_username = os.getenv("NEXTCLOUD_USERNAME", "admin") + assert user_info["username"] == expected_username + + logger.info(f"User info JSON: {json.dumps(user_info, indent=2)}") + + +async def test_user_info_html_with_nextcloud_oauth(playwright_oauth_token): + """Test /user/page endpoint with Nextcloud OAuth token.""" + html = await get_user_info_html(playwright_oauth_token, port=8001) + + # Verify HTML structure + assert "" in html + assert "Nextcloud MCP Server - User Info" in html + assert "oauth" in html.lower() + + # Verify username is displayed + expected_username = os.getenv("NEXTCLOUD_USERNAME", "admin") + assert expected_username in html + + # Verify OAuth-specific content + assert "Client ID" in html + assert "Scopes" in html + assert "Token Expires At" in html + + logger.info(f"User info HTML page rendered successfully ({len(html)} chars)") + + +async def test_user_info_json_unauthenticated(): + """Test /user endpoint without authentication returns 401.""" + async with httpx.AsyncClient() as client: + response = await client.get("http://localhost:8001/user") + + # Should return 401 without authentication + assert response.status_code == 401 + + # Verify error message + data = response.json() + assert "error" in data + assert data["error"] == "Not authenticated" + + logger.info("Unauthenticated request correctly returned 401") + + +async def test_user_info_html_unauthenticated(): + """Test /user/page endpoint without authentication returns 401 HTML.""" + async with httpx.AsyncClient() as client: + response = await client.get("http://localhost:8001/user/page") + + # Should return 401 without authentication + assert response.status_code == 401 + + # Verify HTML error page + html = response.text + assert "" in html + assert "Authentication Required" in html + assert "You must be authenticated to view this page" in html + + logger.info("Unauthenticated HTML request correctly returned 401 page") + + +async def test_user_info_with_alice_token(alice_oauth_token): + """Test /user endpoint with alice's OAuth token.""" + user_info = await get_user_info_json(alice_oauth_token, port=8001) + + # Verify alice's user info + assert user_info["username"] == "alice" + assert user_info["auth_mode"] == "oauth" + assert isinstance(user_info["scopes"], list) + assert len(user_info["scopes"]) > 0 + + logger.info( + f"Alice's user info: username={user_info['username']}, scopes={user_info['scopes']}" + ) + + +async def test_user_info_with_bob_token(bob_oauth_token): + """Test /user endpoint with bob's OAuth token.""" + user_info = await get_user_info_json(bob_oauth_token, port=8001) + + # Verify bob's user info + assert user_info["username"] == "bob" + assert user_info["auth_mode"] == "oauth" + + logger.info(f"Bob's user info: username={user_info['username']}") + + +async def test_user_info_scopes_reflect_token(playwright_oauth_token_read_only): + """Test that /user endpoint reflects token's scopes.""" + user_info = await get_user_info_json(playwright_oauth_token_read_only, port=8001) + + # Verify scopes are present and reflect read-only access + assert "scopes" in user_info + scopes = user_info["scopes"] + assert isinstance(scopes, list) + + # Read-only token should have read scopes but not write scopes + # Note: Actual scope names depend on configuration + logger.info(f"Read-only token scopes: {scopes}") + + +async def test_user_info_idp_profile_included(playwright_oauth_token): + """Test that /user endpoint includes IdP profile when available.""" + user_info = await get_user_info_json(playwright_oauth_token, port=8001) + + # Should have either idp_profile or idp_profile_error + has_profile = "idp_profile" in user_info + has_error = "idp_profile_error" in user_info + + assert has_profile or has_error, "Should have IdP profile data or error" + + if has_profile: + idp_profile = user_info["idp_profile"] + assert isinstance(idp_profile, dict) + # Common OIDC claims + assert "sub" in idp_profile, "IdP profile should include 'sub' claim" + logger.info(f"IdP profile included: {json.dumps(idp_profile, indent=2)}") + else: + logger.warning(f"IdP profile query failed: {user_info['idp_profile_error']}") + + +# ============================================================================ +# Keycloak OAuth Tests (mcp-keycloak on port 8002) +# ============================================================================ + + +@pytest.mark.keycloak +async def test_user_info_json_with_keycloak_oauth(keycloak_oauth_token): + """Test /user endpoint with Keycloak OAuth token.""" + user_info = await get_user_info_json(keycloak_oauth_token, port=8002) + + # Verify response structure + assert "username" in user_info + assert "auth_mode" in user_info + assert user_info["auth_mode"] == "oauth" + + # Verify Keycloak username (default admin user) + assert user_info["username"] == "admin" + + # Verify OAuth-specific fields + assert "client_id" in user_info + assert "scopes" in user_info + assert isinstance(user_info["scopes"], list) + + logger.info(f"Keycloak user info JSON: {json.dumps(user_info, indent=2)}") + + +@pytest.mark.keycloak +async def test_user_info_html_with_keycloak_oauth(keycloak_oauth_token): + """Test /user/page endpoint with Keycloak OAuth token.""" + html = await get_user_info_html(keycloak_oauth_token, port=8002) + + # Verify HTML structure + assert "" in html + assert "Nextcloud MCP Server - User Info" in html + + # Verify Keycloak username is displayed + assert "admin" in html + + logger.info( + f"Keycloak user info HTML page rendered successfully ({len(html)} chars)" + ) + + +@pytest.mark.keycloak +async def test_keycloak_user_info_idp_profile(keycloak_oauth_token): + """Test that Keycloak IdP profile includes extended claims.""" + user_info = await get_user_info_json(keycloak_oauth_token, port=8002) + + # Keycloak should provide IdP profile with extended claims + if "idp_profile" in user_info: + idp_profile = user_info["idp_profile"] + + # Standard OIDC claims + assert "sub" in idp_profile + + # Keycloak-specific claims (may vary by configuration) + # Common claims: email, preferred_username, name, groups, roles + logger.info(f"Keycloak IdP profile: {json.dumps(idp_profile, indent=2)}") + + # Verify at least one identity claim exists + identity_claims = ["email", "preferred_username", "name", "sub"] + has_identity = any(claim in idp_profile for claim in identity_claims) + assert has_identity, ( + f"IdP profile should include at least one identity claim: {identity_claims}" + ) + + +@pytest.mark.keycloak +async def test_keycloak_user_info_unauthenticated(): + """Test /user endpoint on Keycloak server without authentication.""" + async with httpx.AsyncClient() as client: + response = await client.get("http://localhost:8002/user") + + # Should return 401 + assert response.status_code == 401 + + data = response.json() + assert "error" in data + + logger.info("Keycloak server correctly returned 401 for unauthenticated request") + + +# ============================================================================ +# Cross-Mode Comparison Tests +# ============================================================================ + + +async def test_user_info_consistency_across_users(alice_oauth_token, bob_oauth_token): + """Test that user info structure is consistent across different users.""" + alice_info = await get_user_info_json(alice_oauth_token, port=8001) + bob_info = await get_user_info_json(bob_oauth_token, port=8001) + + # Both should have same structure + assert set(alice_info.keys()) == set(bob_info.keys()), ( + "User info structure should be consistent across users" + ) + + # But different usernames + assert alice_info["username"] == "alice" + assert bob_info["username"] == "bob" + + logger.info("User info structure is consistent across users")