diff --git a/docker-compose.yml b/docker-compose.yml index 8d993c6d..3a1c6c0e 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -153,10 +153,10 @@ services: - ENABLE_MULTI_USER_BASIC_AUTH=true - ENABLE_BACKGROUND_OPERATIONS=true - # Token storage (required for middleware initialization) - # DEVELOPMENT ONLY - generate a fresh key for production: - # python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())" - - TOKEN_ENCRYPTION_KEY=fqqI4G51yBCOcu9cvv6wCUJB7sf_CK2za5ClC6b86yY= + # Token storage (required for middleware initialization). + # Source the key from .env — see env.sample. To generate a fresh key: + # python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())" + - TOKEN_ENCRYPTION_KEY=${TOKEN_ENCRYPTION_KEY:?TOKEN_ENCRYPTION_KEY must be set in .env (see env.sample)} - TOKEN_STORAGE_DB=/app/data/tokens.db - ENABLE_SEMANTIC_SEARCH=true @@ -230,9 +230,9 @@ services: - NEXTCLOUD_RESOURCE_URI=nextcloud # ADR-005: Keycloak uses client IDs as audiences, not URLs - NEXTCLOUD_PUBLIC_ISSUER_URL=http://localhost:8888/realms/nextcloud-mcp - # Refresh token storage (ADR-002 Tier 1 & 2) + # Refresh token storage (ADR-002 Tier 1 & 2). Source from .env. - ENABLE_BACKGROUND_OPERATIONS=true - - TOKEN_ENCRYPTION_KEY=ESF1BvEQdGYsCluwMx9Cxvw3uh5pFowPH7Rg_nIliyo= + - TOKEN_ENCRYPTION_KEY=${TOKEN_ENCRYPTION_KEY:?TOKEN_ENCRYPTION_KEY must be set in .env (see env.sample)} - TOKEN_STORAGE_DB=/app/data/tokens.db # ADR-005: Token exchange mode (RFC 8693) @@ -278,10 +278,10 @@ services: # Login Flow v2 (ADR-022) - ENABLE_LOGIN_FLOW=true - # Token storage (required for app password + session persistence) - # DEVELOPMENT ONLY - generate a fresh key for production: - # python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())" - - TOKEN_ENCRYPTION_KEY=rxJvkBf7ZBjZZDL4a1sSqjhmjawhmbRMSOGfK8HDyKU= + # Token storage (required for app password + session persistence). + # Source the key from .env — see env.sample. To generate a fresh key: + # python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())" + - TOKEN_ENCRYPTION_KEY=${TOKEN_ENCRYPTION_KEY:?TOKEN_ENCRYPTION_KEY must be set in .env (see env.sample)} - TOKEN_STORAGE_DB=/app/data/tokens.db # Semantic search diff --git a/env.sample b/env.sample index 652d9435..bae9d96e 100644 --- a/env.sample +++ b/env.sample @@ -15,6 +15,15 @@ # Your Nextcloud instance URL (without trailing slash) NEXTCLOUD_HOST= +# Fernet key for encrypting refresh tokens / app passwords / browser +# sessions in SQLite. Required by every docker-compose profile that runs +# the MCP server (single-user, multi-user-basic, keycloak, login-flow). +# Generate one with: +# python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())" +# NEVER commit a real key. Each environment (dev / staging / prod) needs +# its own key. +TOKEN_ENCRYPTION_KEY= + # ============================================ # SINGLE-USER BASICAUTH MODE # ============================================ diff --git a/nextcloud_mcp_server/alembic/versions/20260502_1500_005_add_browser_sessions.py b/nextcloud_mcp_server/alembic/versions/20260502_1500_005_add_browser_sessions.py new file mode 100644 index 00000000..b85f20d1 --- /dev/null +++ b/nextcloud_mcp_server/alembic/versions/20260502_1500_005_add_browser_sessions.py @@ -0,0 +1,49 @@ +"""Add browser_sessions table for random-id browser cookie auth. + +Replaces the prior `mcp_session=` cookie pattern (issue #626 +finding 2) with a server-side mapping from a cryptographically random +session id to the authenticated user_id. The cookie value is now opaque +and revocable. + +Revision ID: 005 +Revises: 004 +Create Date: 2026-05-02 15:00:00.000000 +""" + +from alembic import op + +revision = "005" +down_revision = "004" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.execute( + """ + CREATE TABLE IF NOT EXISTS browser_sessions ( + session_id TEXT PRIMARY KEY, + user_id TEXT NOT NULL, + created_at INTEGER NOT NULL, + expires_at INTEGER NOT NULL + ) + """ + ) + op.execute( + """ + CREATE INDEX IF NOT EXISTS idx_browser_sessions_user + ON browser_sessions(user_id) + """ + ) + op.execute( + """ + CREATE INDEX IF NOT EXISTS idx_browser_sessions_expires + ON browser_sessions(expires_at) + """ + ) + + +def downgrade() -> None: + op.execute("DROP INDEX IF EXISTS idx_browser_sessions_expires") + op.execute("DROP INDEX IF EXISTS idx_browser_sessions_user") + op.execute("DROP TABLE IF EXISTS browser_sessions") diff --git a/nextcloud_mcp_server/auth/browser_oauth_routes.py b/nextcloud_mcp_server/auth/browser_oauth_routes.py index c1050532..577c3024 100644 --- a/nextcloud_mcp_server/auth/browser_oauth_routes.py +++ b/nextcloud_mcp_server/auth/browser_oauth_routes.py @@ -10,14 +10,18 @@ import os import secrets import time from base64 import urlsafe_b64encode +from html import escape as html_escape from urllib.parse import urlencode from urllib.parse import urlparse as parse_url import httpx -import jwt from starlette.requests import Request from starlette.responses import HTMLResponse, JSONResponse, RedirectResponse +from nextcloud_mcp_server.auth.token_utils import ( + IdTokenVerificationError, + verify_id_token, +) from nextcloud_mcp_server.auth.userinfo_routes import ( _get_userinfo_endpoint, _query_idp_userinfo, @@ -383,16 +387,44 @@ async def oauth_login_callback(request: Request) -> RedirectResponse | HTMLRespo 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 + # Resolve the discovery URL + audience used for THIS auth request so + # we can verify the ID token signature + claims (issue #626 finding 1). + if oauth_client: + # External IdP path + verification_audience = oauth_client.client_id + verification_discovery_url = getattr(oauth_client, "discovery_url", None) + else: + # Integrated Nextcloud OIDC path + verification_audience = oauth_config["client_id"] + verification_discovery_url = oauth_config.get("discovery_url") + + if not verification_discovery_url: + logger.error("Cannot verify ID token: no discovery_url available") + return HTMLResponse( + "

Login Failed

OIDC discovery URL not configured

", + status_code=500, + ) + 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" + userinfo = await verify_id_token( + id_token, + discovery_url=verification_discovery_url, + expected_audience=verification_audience, + ) + except IdTokenVerificationError as e: + logger.error("ID token verification failed: %s", e) + # html_escape: defense-in-depth. The exception text is currently + # server-constructed, but escape on the success path too so any + # future error wrapping that includes IdP response text can't + # smuggle markup into the login-failure page. + return HTMLResponse( + f"

Login Failed

ID token failed verification: {html_escape(str(e))}

", + status_code=400, + ) + + user_id = userinfo["sub"] + username = userinfo.get("preferred_username") or userinfo.get("email") + logger.info("Browser login successful: %s (sub=%s)", username, user_id) # Calculate refresh token expiration from token response refresh_expires_in = token_data.get("refresh_expires_in") @@ -455,40 +487,118 @@ async def oauth_login_callback(request: Request) -> RedirectResponse | HTMLRespo logger.error(f"Error caching user profile: {e}") # Continue anyway - profile cache is optional for browser UI - # Create response and set session cookie - # Redirect to stored next_url (from OAuth session) or /app as default + # Create a server-side browser session: a random opaque session_id is + # mapped to the verified user_id in `browser_sessions`. The cookie value + # is the session_id (never the raw user_id — see issue #626 finding 2). + session_id = secrets.token_urlsafe(32) + session_ttl = 86400 * 30 # 30 days + await storage.create_browser_session( + session_id=session_id, user_id=user_id, ttl_seconds=session_ttl + ) + response = RedirectResponse(next_url, status_code=302) response.set_cookie( key="mcp_session", - value=user_id, - max_age=86400 * 30, # 30 days + value=session_id, + max_age=session_ttl, httponly=True, secure=_should_use_secure_cookies(), samesite="lax", ) - logger.info(f"Session cookie set for user: {username}") + logger.info("Session cookie set for user %s (sid=%s…)", username, session_id[:8]) return response async def oauth_logout(request: Request) -> RedirectResponse: - """Browser OAuth logout - clears session cookie. + """Browser OAuth logout — invalidate session and revoke refresh token. + + Issue #626 finding 4: prior implementation only cleared the cookie, + leaving the refresh token in storage (valid up to 90 days). This now: + 1. Resolves the user_id for the current browser session_id. + 2. Calls the IdP `revocation_endpoint` for the stored refresh token + when the IdP advertises one. + 3. Deletes the stored refresh token regardless of revocation success. + 4. Deletes the browser_sessions row so the cookie is unusable even + if it leaks. + 5. Clears the cookie on the response. 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") + session_id = request.cookies.get("mcp_session") - # TODO: Optionally revoke refresh token from storage - # session_id = request.cookies.get("mcp_session") - # if session_id: - # await storage.delete_refresh_token(session_id) + oauth_ctx = getattr(request.app.state, "oauth_context", None) + storage = oauth_ctx.get("storage") if oauth_ctx else None + + if session_id and storage and oauth_ctx: + try: + user_id = await storage.get_browser_session_user(session_id) + if user_id: + token_data = await storage.get_refresh_token(user_id) + refresh_token = token_data.get("refresh_token") if token_data else None + + if refresh_token: + await _revoke_refresh_token_at_idp(oauth_ctx, refresh_token) + await storage.delete_refresh_token(user_id) + logger.info("Refresh token revoked + deleted for user %s", user_id) + + await storage.delete_browser_session(session_id) + except Exception as e: + # Logout must always succeed locally; log and continue. + logger.warning("Logout cleanup failed (continuing): %s", e) response = RedirectResponse(next_url, status_code=302) response.delete_cookie("mcp_session") logger.info("User logged out, session cookie cleared") return response + + +async def _revoke_refresh_token_at_idp(oauth_ctx: dict, refresh_token: str) -> None: + """Best-effort RFC 7009 revocation against the IdP. + + Silent on failure: revoking remotely is a defense-in-depth step on top + of deleting the local copy, and we don't want logout to error if the + IdP is unreachable or doesn't advertise a revocation endpoint. + """ + try: + discovery_url = oauth_ctx.get("discovery_url") or os.getenv( + "OIDC_DISCOVERY_URL", + f"{os.getenv('NEXTCLOUD_HOST', '')}/.well-known/openid-configuration", + ) + if not discovery_url: + return + + async with nextcloud_httpx_client() as http_client: + discovery_response = await http_client.get(discovery_url) + discovery_response.raise_for_status() + discovery = discovery_response.json() + revocation_endpoint = discovery.get("revocation_endpoint") + if not revocation_endpoint: + logger.debug("IdP advertises no revocation_endpoint; skipping") + return + + client_id = oauth_ctx.get("client_id") or os.getenv("OIDC_CLIENT_ID") + client_secret = oauth_ctx.get("client_secret") or os.getenv( + "OIDC_CLIENT_SECRET" + ) + if not (client_id and client_secret): + logger.debug("No OIDC client credentials available for revocation") + return + + response = await http_client.post( + revocation_endpoint, + data={ + "token": refresh_token, + "token_type_hint": "refresh_token", + }, + auth=(client_id, client_secret), + ) + if response.status_code >= 400: + logger.warning( + "Refresh token revocation returned HTTP %s", response.status_code + ) + except Exception as e: + logger.warning("Refresh token revocation failed: %s", e) diff --git a/nextcloud_mcp_server/auth/oauth_routes.py b/nextcloud_mcp_server/auth/oauth_routes.py index 1eb39286..1ea81551 100644 --- a/nextcloud_mcp_server/auth/oauth_routes.py +++ b/nextcloud_mcp_server/auth/oauth_routes.py @@ -30,13 +30,16 @@ from typing import Any from urllib.parse import unquote, urlencode from urllib.parse import urlparse as parse_url -import jwt from starlette.requests import Request from starlette.responses import HTMLResponse, JSONResponse, RedirectResponse from nextcloud_mcp_server.auth.browser_oauth_routes import oauth_login_callback from nextcloud_mcp_server.auth.client_registry import get_client_registry from nextcloud_mcp_server.auth.storage import RefreshTokenStorage +from nextcloud_mcp_server.auth.token_utils import ( + IdTokenVerificationError, + verify_id_token, +) from nextcloud_mcp_server.config import get_settings from ..http import nextcloud_httpx_client @@ -643,23 +646,29 @@ async def oauth_callback_nextcloud(request: Request): refresh_token = token_data.get("refresh_token") id_token = token_data.get("id_token") - # Decode ID token to get user info - logger.info("=" * 60) - logger.info("oauth_callback_nextcloud: Extracting user_id from ID token") - logger.info("=" * 60) + # Verify ID token signature + claims (issue #626 finding 1). + logger.info("oauth_callback_nextcloud: Verifying ID token") 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(" ✓ ID token decode SUCCESSFUL") - logger.info(f" Extracted user_id: {user_id}") - logger.info(f" Username: {username}") - logger.info(f" ID token payload keys: {list(userinfo.keys())}") - logger.info(f"Flow 2: User {username} provisioned resource access") - except Exception as e: - logger.error(f" ✗ ID token decode FAILED: {type(e).__name__}: {e}") - user_id = "unknown" - logger.error(f" Using fallback user_id: {user_id}") + userinfo = await verify_id_token( + id_token, + discovery_url=discovery_url, + expected_audience=mcp_server_client_id, + ) + except IdTokenVerificationError as e: + logger.error("ID token verification failed: %s", e) + return JSONResponse( + { + "error": "invalid_token", + "error_description": "ID token failed verification", + }, + status_code=400, + ) + + user_id = userinfo["sub"] + username = userinfo.get("preferred_username") or userinfo.get("email") + logger.info( + "Flow 2: User %s (sub=%s) provisioned resource access", username, user_id + ) # Store master refresh token for Flow 2 if refresh_token: diff --git a/nextcloud_mcp_server/auth/provisioning_decorator.py b/nextcloud_mcp_server/auth/provisioning_decorator.py index 585b6ff2..f9534e31 100644 --- a/nextcloud_mcp_server/auth/provisioning_decorator.py +++ b/nextcloud_mcp_server/auth/provisioning_decorator.py @@ -9,7 +9,7 @@ import functools import logging from typing import Callable -import jwt +from mcp.server.auth.middleware.auth_context import get_access_token from mcp.server.fastmcp import Context from mcp.shared.exceptions import McpError from mcp.types import ErrorData @@ -65,16 +65,12 @@ def require_provisioning(func: Callable) -> Callable: return await func(*args, **kwargs) # Offline access mode - check if user has completed Flow 2 provisioning - # Get user_id from authorization token - user_id = None - if hasattr(ctx, "authorization") and ctx.authorization: - try: - token = ctx.authorization.token - payload = jwt.decode(token, options={"verify_signature": False}) - user_id = payload.get("sub") - logger.debug(f"Checking provisioning for user: {user_id}") - except Exception as e: - logger.warning(f"Failed to extract user_id from token: {e}") + # Read user_id from the verified AccessToken populated by + # UnifiedTokenVerifier; no second decode of the raw JWT here. + access_token = get_access_token() + user_id = access_token.resource if access_token else None + if user_id: + logger.debug("Checking provisioning for user: %s", user_id) if not user_id: raise McpError( @@ -149,12 +145,8 @@ def require_provisioning_or_suggest(func: Callable) -> Callable: if ctx: # Try to check provisioning status try: - # Get user_id from authorization token - user_id = None - if hasattr(ctx, "authorization") and ctx.authorization: - token = ctx.authorization.token - payload = jwt.decode(token, options={"verify_signature": False}) - user_id = payload.get("sub") + access_token = get_access_token() + user_id = access_token.resource if access_token else None if user_id: # Check provisioning status diff --git a/nextcloud_mcp_server/auth/session_backend.py b/nextcloud_mcp_server/auth/session_backend.py index 1a3dc714..f4654b64 100644 --- a/nextcloud_mcp_server/auth/session_backend.py +++ b/nextcloud_mcp_server/auth/session_backend.py @@ -52,44 +52,46 @@ class SessionAuthBackend(AuthenticationBackend): username = os.getenv("NEXTCLOUD_USERNAME", "admin") return AuthCredentials(["authenticated", "admin"]), SimpleUser(username) - # OAuth mode: Check for session cookie + # OAuth mode: opaque random session_id cookie -> user_id mapping. + # Replaces the prior `mcp_session=` cookie pattern (issue + # #626 finding 2). The cookie value is no longer the user identity; + # we look it up server-side and reject unknown / expired sessions. 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]}..." + user_id = await storage.get_browser_session_user(session_id) + if not user_id: + logger.info( + "Browser session not found or expired (sid=%s…)", session_id[:8] ) 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]}...") + # Defense-in-depth: only authenticate sessions for users that + # actually have a refresh token persisted. Logout deletes both, + # so an expired/revoked user state will fail closed here. + token_data = await storage.get_refresh_token(user_id) + if not token_data: + logger.warning( + "Session %s… has no refresh token for user %s; rejecting", + session_id[:8], + user_id, + ) + return None - return AuthCredentials(["authenticated"]), SimpleUser(username) + return AuthCredentials(["authenticated"]), SimpleUser(user_id) except Exception as e: logger.warning(f"Session validation error: {e}") diff --git a/nextcloud_mcp_server/auth/storage.py b/nextcloud_mcp_server/auth/storage.py index ed6675df..38da1d11 100644 --- a/nextcloud_mcp_server/auth/storage.py +++ b/nextcloud_mcp_server/auth/storage.py @@ -1133,6 +1133,90 @@ class RefreshTokenStorage: return deleted + # ============================================================================ + # Browser Sessions (OAuth admin UI) + # ============================================================================ + # + # Maps a cryptographically random `session_id` (cookie value) to the + # authenticated user_id. Replaces the prior `mcp_session=` + # cookie pattern (issue #626 finding 2). Cookie value is opaque, expires, + # and can be revoked server-side without forcing the user to roll their + # IdP `sub`. + + async def create_browser_session( + self, + session_id: str, + user_id: str, + ttl_seconds: int = 86400 * 30, + ) -> None: + """Persist a random session_id → user_id mapping for browser auth.""" + if not self._initialized: + await self.initialize() + + now = int(time.time()) + expires_at = now + ttl_seconds + + async with aiosqlite.connect(self.db_path) as db: + await db.execute( + """ + INSERT OR REPLACE INTO browser_sessions + (session_id, user_id, created_at, expires_at) + VALUES (?, ?, ?, ?) + """, + (session_id, user_id, now, expires_at), + ) + await db.commit() + + logger.debug( + "Stored browser session %s for user %s (expires in %ss)", + session_id[:8], + user_id, + ttl_seconds, + ) + + async def get_browser_session_user(self, session_id: str) -> Optional[str]: + """Look up the user_id bound to a browser session_id, or None. + + Returns None when the session is unknown or expired. Expired rows + are deleted on encounter to keep the table small. + """ + if not self._initialized: + await self.initialize() + + async with aiosqlite.connect(self.db_path) as db: + db.row_factory = aiosqlite.Row + async with db.execute( + "SELECT user_id, expires_at FROM browser_sessions WHERE session_id = ?", + (session_id,), + ) as cursor: + row = await cursor.fetchone() + + if not row: + return None + + if row["expires_at"] < time.time(): + logger.debug("Browser session %s expired", session_id[:8]) + await self.delete_browser_session(session_id) + return None + + return row["user_id"] + + async def delete_browser_session(self, session_id: str) -> bool: + """Delete a browser session row. Returns True when a row was removed.""" + if not self._initialized: + await self.initialize() + + async with aiosqlite.connect(self.db_path) as db: + cursor = await db.execute( + "DELETE FROM browser_sessions WHERE session_id = ?", (session_id,) + ) + await db.commit() + deleted = cursor.rowcount > 0 + + if deleted: + logger.debug("Deleted browser session %s", session_id[:8]) + return deleted + # ============================================================================ # Webhook Registration Tracking (both BasicAuth and OAuth modes) # ============================================================================ diff --git a/nextcloud_mcp_server/auth/token_broker.py b/nextcloud_mcp_server/auth/token_broker.py index c4729edb..9885246f 100644 --- a/nextcloud_mcp_server/auth/token_broker.py +++ b/nextcloud_mcp_server/auth/token_broker.py @@ -20,7 +20,6 @@ from typing import Dict, Optional, Tuple import anyio import httpx -import jwt from nextcloud_mcp_server.auth.storage import RefreshTokenStorage @@ -489,35 +488,6 @@ class TokenBrokerService: ) return access_token, expires_in - async def _validate_token_audience(self, token: str, expected_audience: str): - """ - Validate that token has correct audience claim. - - Args: - token: JWT token to validate - expected_audience: Expected audience value - - Raises: - ValueError: If audience doesn't match - """ - try: - # Decode without verification to check claims - # In production, should verify signature - claims = jwt.decode(token, options={"verify_signature": False}) - - audience = claims.get("aud", []) - if isinstance(audience, str): - audience = [audience] - - if expected_audience not in audience: - raise ValueError( - f"Token audience {audience} doesn't include {expected_audience}" - ) - - except jwt.DecodeError as e: - # Token might be opaque, skip validation - logger.debug(f"Cannot decode token for audience validation: {e}") - async def refresh_master_token(self, user_id: str) -> bool: """ Refresh the master refresh token (periodic rotation). diff --git a/nextcloud_mcp_server/auth/token_utils.py b/nextcloud_mcp_server/auth/token_utils.py index d9a4cb4f..51020ec1 100644 --- a/nextcloud_mcp_server/auth/token_utils.py +++ b/nextcloud_mcp_server/auth/token_utils.py @@ -5,81 +5,155 @@ between server/ and auth/ layers. """ import logging -import os +from typing import Any import jwt +from jwt import PyJWKSet from mcp.server.auth.middleware.auth_context import get_access_token from mcp.server.auth.provider import AccessToken from mcp.server.fastmcp import Context -from nextcloud_mcp_server.auth.userinfo_routes import _query_idp_userinfo - from ..http import nextcloud_httpx_client logger = logging.getLogger(__name__) -async def extract_user_id_from_token(ctx: Context) -> str: - """Extract user_id from the MCP access token (Flow 1). +class IdTokenVerificationError(Exception): + """Raised when an OIDC ID token fails signature or claim verification.""" - Handles both JWT and opaque tokens: - - JWT: Decode and extract 'sub' claim - - Opaque: Call userinfo endpoint to get 'sub' + +async def verify_id_token( + id_token: str, + *, + discovery_url: str, + expected_audience: str, + expected_nonce: str | None = None, +) -> dict[str, Any]: + """Verify an OIDC ID token's signature and standard claims. + + Implements the verification steps required by OIDC core spec section + 3.1.3.7 (ID Token Validation) for the authorization-code flow: + - Signature against JWKS (RS256) + - Issuer matches the OP that issued the token + - Audience contains the expected client_id + - Token is not expired (`exp`) + - `iat` is well-formed (PyJWT default) + - `nonce` matches when one was included in the auth request + + Replaces the prior `jwt.decode(id_token, options={"verify_signature": False})` + pattern (issue #626 finding 1) on the OAuth callback paths. Args: - ctx: MCP context with access token + id_token: Raw ID token (JWT) string. + discovery_url: OIDC `.well-known/openid-configuration` URL of the IdP. + expected_audience: The MCP-server-side OAuth client_id used for this + authorization request. + expected_nonce: When the auth request included a nonce, the same value + so it can be checked here. None disables the nonce check (callers + that didn't bind a nonce in the auth request). Returns: - user_id extracted from token, or "default_user" as fallback + Decoded, verified ID-token claims. + + Raises: + IdTokenVerificationError: On any verification failure. """ - # Use MCP SDK's get_access_token() which uses contextvars - access_token: AccessToken | None = get_access_token() + if not id_token: + raise IdTokenVerificationError("ID token missing from token response") - if not access_token or not access_token.token: - logger.warning(" ✗ No access token found via get_access_token()") - return "default_user" - - token = access_token.token - is_jwt = "." in token and token.count(".") >= 2 - logger.info(f" Token type: {'JWT' if is_jwt else 'Opaque'}") - - # Try JWT decode first - if is_jwt: - try: - payload = jwt.decode(token, options={"verify_signature": False}) - user_id = payload.get("sub", "unknown") - logger.info(f" ✓ JWT decode successful: user_id={user_id}") - return user_id - except Exception as e: - logger.error(f" ✗ JWT decode failed: {type(e).__name__}: {e}") - - # Opaque token - call userinfo endpoint - logger.info(" Opaque token detected, calling userinfo endpoint...") try: - # Get userinfo endpoint from OIDC discovery - oidc_discovery_uri = os.getenv( - "OIDC_DISCOVERY_URI", - "http://localhost:8080/.well-known/openid-configuration", - ) async with nextcloud_httpx_client() as http_client: - discovery_response = await http_client.get(oidc_discovery_uri) + discovery_response = await http_client.get(discovery_url) discovery_response.raise_for_status() discovery = discovery_response.json() - userinfo_endpoint = discovery.get("userinfo_endpoint") - if userinfo_endpoint: - userinfo = await _query_idp_userinfo(token, userinfo_endpoint) - if userinfo: - user_id = userinfo.get("sub", "unknown") - logger.info(f" ✓ Userinfo query successful: user_id={user_id}") - return user_id - else: - logger.error(" ✗ Userinfo query failed") - else: - logger.error(" ✗ No userinfo_endpoint available") + issuer = discovery.get("issuer") + jwks_uri = discovery.get("jwks_uri") + if not issuer or not jwks_uri: + raise IdTokenVerificationError( + "OIDC discovery response missing issuer or jwks_uri" + ) + + jwks_response = await http_client.get(jwks_uri) + jwks_response.raise_for_status() + jwks_data = jwks_response.json() + except IdTokenVerificationError: + raise except Exception as e: - logger.error(f" ✗ Userinfo query failed: {type(e).__name__}: {e}") + raise IdTokenVerificationError( + f"Failed to fetch OIDC discovery / JWKS: {e}" + ) from e - # Fallback - logger.warning(" Using fallback user_id: default_user") - return "default_user" + try: + jwks = PyJWKSet.from_dict(jwks_data) + unverified_header = jwt.get_unverified_header(id_token) + kid = unverified_header.get("kid") + if not kid: + raise IdTokenVerificationError("ID token header missing 'kid'") + try: + signing_key = jwks[kid] + except KeyError as e: + raise IdTokenVerificationError( + f"No JWKS key matches ID token kid {kid!r}" + ) from e + + payload: dict[str, Any] = jwt.decode( + id_token, + signing_key.key, + algorithms=["RS256"], + audience=expected_audience, + issuer=issuer, + options={ + "verify_signature": True, + "verify_exp": True, + "verify_iat": True, + "verify_aud": True, + "verify_iss": True, + "require": ["sub", "iss", "aud", "exp", "iat"], + }, + ) + except IdTokenVerificationError: + raise + except jwt.PyJWTError as e: + raise IdTokenVerificationError(f"ID token verification failed: {e}") from e + except Exception as e: + raise IdTokenVerificationError( + f"Unexpected error verifying ID token: {e}" + ) from e + + if expected_nonce is not None and payload.get("nonce") != expected_nonce: + raise IdTokenVerificationError("ID token nonce does not match request nonce") + + return payload + + +async def extract_user_id_from_token(ctx: Context) -> str: + """Extract user_id from the verified MCP access token. + + Reads the `sub` claim from `AccessToken.resource`, which is populated by + `UnifiedTokenVerifier` after JWT signature verification (or token + introspection for opaque tokens). We never re-decode the raw token here: + the verifier has already validated the signature and extracted the + identity claim. + + Args: + ctx: MCP context with access token (unused — kept for the public API) + + Returns: + user_id from the verified token, or "default_user" when no token is + present (e.g. BasicAuth mode where this should not be called). + """ + access_token: AccessToken | None = get_access_token() + + if not access_token: + logger.warning("No access token found via get_access_token()") + return "default_user" + + user_id = access_token.resource + if not user_id: + logger.error( + "Access token has no resource (sub) claim — verifier should have rejected it" + ) + return "default_user" + + return user_id diff --git a/nextcloud_mcp_server/server/oauth_tools.py b/nextcloud_mcp_server/server/oauth_tools.py index 3698aa82..90769bf5 100644 --- a/nextcloud_mcp_server/server/oauth_tools.py +++ b/nextcloud_mcp_server/server/oauth_tools.py @@ -198,9 +198,7 @@ def generate_oauth_url_for_flow2( return f"{auth_endpoint}?{urlencode(params)}" -async def provision_nextcloud_access( - ctx: Context, user_id: Optional[str] = None -) -> ProvisioningResult: +async def provision_nextcloud_access(ctx: Context, user_id: str) -> ProvisioningResult: """ MCP Tool: Provision offline access to Nextcloud resources. @@ -211,16 +209,13 @@ async def provision_nextcloud_access( Args: ctx: MCP context with user's Flow 1 token - user_id: Optional user identifier (extracted from token if not provided) + user_id: Authenticated user identifier (must be derived from the + verified access token by the caller; never accept from MCP input). Returns: ProvisioningResult with Astrolabe settings URL or status """ try: - # Extract user ID from the MCP access token (Flow 1 token) - if not user_id: - user_id = await extract_user_id_from_token(ctx) - # Check if already provisioned status = await get_provisioning_status(ctx, user_id) if status.is_provisioned: @@ -271,9 +266,7 @@ async def provision_nextcloud_access( ) -async def revoke_nextcloud_access( - ctx: Context, user_id: Optional[str] = None -) -> RevocationResult: +async def revoke_nextcloud_access(ctx: Context, user_id: str) -> RevocationResult: """ MCP Tool: Revoke offline access to Nextcloud resources. @@ -281,19 +274,14 @@ async def revoke_nextcloud_access( that was granted via Flow 2. Args: - mcp: MCP context - user_id: Optional user identifier + ctx: MCP context + user_id: Authenticated user identifier (must be derived from the + verified access token by the caller; never accept from MCP input). Returns: RevocationResult with status """ try: - # Get user ID from token if not provided - if not user_id: - logger.info("Extracting user_id from access token for revoke...") - user_id = await extract_user_id_from_token(ctx) - logger.info(f" Revoke using user_id: {user_id}") - # Check current status status = await get_provisioning_status(ctx, user_id) if not status.is_provisioned: @@ -350,9 +338,7 @@ async def revoke_nextcloud_access( ) -async def check_provisioning_status( - ctx: Context, user_id: Optional[str] = None -) -> ProvisioningStatus: +async def check_provisioning_status(ctx: Context, user_id: str) -> ProvisioningStatus: """ MCP Tool: Check the current provisioning status. @@ -360,24 +346,17 @@ async def check_provisioning_status( Nextcloud access and see details about their current authorization. Args: - mcp: MCP context - user_id: Optional user identifier + ctx: MCP context + user_id: Authenticated user identifier (must be derived from the + verified access token by the caller; never accept from MCP input). Returns: ProvisioningStatus with current state """ - # Get user ID from context if not provided - if not user_id: - user_id = ( - ctx.context.get("user_id", "default_user") # type: ignore - if hasattr(ctx, "context") - else "default_user" - ) - return await get_provisioning_status(ctx, user_id) -async def check_logged_in(ctx: Context, user_id: Optional[str] = None) -> str: +async def check_logged_in(ctx: Context, user_id: str) -> str: """ MCP Tool: Check if user is logged in and elicit login if needed. @@ -387,23 +366,13 @@ async def check_logged_in(ctx: Context, user_id: Optional[str] = None) -> str: Args: ctx: MCP context with user's Flow 1 token - user_id: Optional user identifier (extracted from token if not provided) + user_id: Authenticated user identifier (must be derived from the + verified access token by the caller; never accept from MCP input). Returns: "yes" if logged in, or elicitation prompting for login """ try: - # Extract user ID from the MCP access token (Flow 1 token) - logger.info("=" * 60) - logger.info("check_logged_in: Starting user_id extraction") - logger.info("=" * 60) - - if not user_id: - user_id = await extract_user_id_from_token(ctx) - logger.info(f" Final user_id for check_logged_in: {user_id}") - else: - logger.info(f" user_id provided as argument: {user_id}") - # Check if already logged in logger.info(f"Checking provisioning status for user_id: {user_id}") status = await get_provisioning_status(ctx, user_id) @@ -591,10 +560,8 @@ def register_oauth_tools(mcp): ), ) @require_scopes("openid") - async def tool_provision_access( - ctx: Context, - user_id: Optional[str] = None, - ) -> ProvisioningResult: + async def tool_provision_access(ctx: Context) -> ProvisioningResult: + user_id = await extract_user_id_from_token(ctx) return await provision_nextcloud_access(ctx, user_id) @mcp.tool( @@ -608,9 +575,8 @@ def register_oauth_tools(mcp): ), ) @require_scopes("openid") - async def tool_revoke_access( - ctx: Context, user_id: Optional[str] = None - ) -> RevocationResult: + async def tool_revoke_access(ctx: Context) -> RevocationResult: + user_id = await extract_user_id_from_token(ctx) return await revoke_nextcloud_access(ctx, user_id) @mcp.tool( @@ -623,9 +589,8 @@ def register_oauth_tools(mcp): ), ) @require_scopes("openid") - async def tool_check_status( - ctx: Context, user_id: Optional[str] = None - ) -> ProvisioningStatus: + async def tool_check_status(ctx: Context) -> ProvisioningStatus: + user_id = await extract_user_id_from_token(ctx) return await check_provisioning_status(ctx, user_id) @mcp.tool( @@ -641,5 +606,6 @@ def register_oauth_tools(mcp): ), ) @require_scopes("openid") - async def tool_check_logged_in(ctx: Context, user_id: Optional[str] = None) -> str: + async def tool_check_logged_in(ctx: Context) -> str: + user_id = await extract_user_id_from_token(ctx) return await check_logged_in(ctx, user_id) diff --git a/tests/unit/test_browser_sessions.py b/tests/unit/test_browser_sessions.py new file mode 100644 index 00000000..7851d933 --- /dev/null +++ b/tests/unit/test_browser_sessions.py @@ -0,0 +1,75 @@ +"""Unit tests for browser_sessions storage (issue #626 finding 2). + +The browser admin UI no longer uses the raw user_id as the cookie value +— it uses a cryptographically random session_id mapped server-side to +user_id. These tests pin the storage contract. +""" + +import secrets +import tempfile +import time +from pathlib import Path + +import pytest +from cryptography.fernet import Fernet + +from nextcloud_mcp_server.auth.storage import RefreshTokenStorage + +pytestmark = pytest.mark.unit + + +@pytest.fixture +async def storage(): + with tempfile.TemporaryDirectory() as tmpdir: + db_path = Path(tmpdir) / "test_browser_sessions.db" + s = RefreshTokenStorage( + db_path=str(db_path), encryption_key=Fernet.generate_key().decode() + ) + await s.initialize() + yield s + + +async def test_create_and_get_browser_session(storage): + sid = secrets.token_urlsafe(32) + await storage.create_browser_session(session_id=sid, user_id="alice") + + user_id = await storage.get_browser_session_user(sid) + assert user_id == "alice" + + +async def test_get_browser_session_unknown_returns_none(storage): + assert await storage.get_browser_session_user("does-not-exist") is None + + +async def test_delete_browser_session(storage): + sid = secrets.token_urlsafe(32) + await storage.create_browser_session(session_id=sid, user_id="alice") + + deleted = await storage.delete_browser_session(sid) + assert deleted is True + assert await storage.get_browser_session_user(sid) is None + + +async def test_expired_browser_session_rejected_and_deleted(storage): + sid = secrets.token_urlsafe(32) + # ttl_seconds=0 so the row is immediately expired (now == expires_at) + await storage.create_browser_session(session_id=sid, user_id="alice", ttl_seconds=0) + # Make sure clock advances past expires_at + time.sleep(0.01) + + assert await storage.get_browser_session_user(sid) is None + # Expired row should be deleted on encounter + assert await storage.delete_browser_session(sid) is False + + +async def test_replace_existing_session_id(storage): + """INSERT OR REPLACE so re-using a session_id rebinds the user. + + Not a recommended call pattern (session_ids are random), but the + storage layer must not raise UNIQUE constraint errors if it happens. + """ + sid = secrets.token_urlsafe(32) + await storage.create_browser_session(session_id=sid, user_id="alice") + await storage.create_browser_session(session_id=sid, user_id="bob") + + assert await storage.get_browser_session_user(sid) == "bob" diff --git a/tests/unit/test_id_token_verification.py b/tests/unit/test_id_token_verification.py new file mode 100644 index 00000000..84ed854a --- /dev/null +++ b/tests/unit/test_id_token_verification.py @@ -0,0 +1,233 @@ +"""Unit tests for OIDC ID token verification (issue #626 finding 1). + +The OAuth callback handlers used to call +`jwt.decode(id_token, options={"verify_signature": False})` and trust the +result. They now go through `verify_id_token`, which checks signature +against JWKS and validates issuer / audience / exp / nonce per OIDC core +spec §3.1.3.7. +""" + +import json +import time +from base64 import urlsafe_b64encode +from unittest.mock import patch + +import httpx +import jwt +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import rsa + +from nextcloud_mcp_server.auth.token_utils import ( + IdTokenVerificationError, + verify_id_token, +) + +pytestmark = pytest.mark.unit + + +# Generated once per process — RSA keypair generation is slow. +_KEY = rsa.generate_private_key(public_exponent=65537, key_size=2048) +_PRIVATE_PEM = _KEY.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.TraditionalOpenSSL, + encryption_algorithm=serialization.NoEncryption(), +) +_OTHER_KEY = rsa.generate_private_key(public_exponent=65537, key_size=2048) +_OTHER_PRIVATE_PEM = _OTHER_KEY.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.TraditionalOpenSSL, + encryption_algorithm=serialization.NoEncryption(), +) + +ISSUER = "https://idp.example.com" +DISCOVERY_URL = f"{ISSUER}/.well-known/openid-configuration" +JWKS_URI = f"{ISSUER}/jwks" + + +def _b64u_uint(n: int) -> str: + raw = n.to_bytes((n.bit_length() + 7) // 8, "big") + return urlsafe_b64encode(raw).rstrip(b"=").decode("ascii") + + +def _build_jwks() -> dict: + pub = _KEY.public_key().public_numbers() + return { + "keys": [ + { + "kty": "RSA", + "use": "sig", + "kid": "test-key-1", + "alg": "RS256", + "n": _b64u_uint(pub.n), + "e": _b64u_uint(pub.e), + } + ] + } + + +def _sign( + claims: dict, *, kid: str = "test-key-1", key_pem: bytes = _PRIVATE_PEM +) -> str: + return jwt.encode(claims, key_pem, algorithm="RS256", headers={"kid": kid}) + + +def _idp_handler(request: httpx.Request) -> httpx.Response: + if str(request.url) == DISCOVERY_URL: + return httpx.Response(200, json={"issuer": ISSUER, "jwks_uri": JWKS_URI}) + if str(request.url) == JWKS_URI: + return httpx.Response( + 200, + content=json.dumps(_build_jwks()).encode(), + headers={"content-type": "application/json"}, + ) + return httpx.Response(404) + + +@pytest.fixture +def mock_idp(): + """Patch nextcloud_httpx_client used inside token_utils.verify_id_token.""" + transport = httpx.MockTransport(_idp_handler) + + def fake_client(**kwargs): + kwargs["transport"] = transport + return httpx.AsyncClient(**kwargs) + + with patch( + "nextcloud_mcp_server.auth.token_utils.nextcloud_httpx_client", + side_effect=fake_client, + ): + yield + + +async def test_verify_id_token_accepts_valid_token(mock_idp): + now = int(time.time()) + token = _sign( + { + "iss": ISSUER, + "aud": "test-client", + "sub": "alice", + "iat": now, + "exp": now + 60, + } + ) + payload = await verify_id_token( + token, discovery_url=DISCOVERY_URL, expected_audience="test-client" + ) + assert payload["sub"] == "alice" + + +async def test_verify_id_token_rejects_wrong_audience(mock_idp): + now = int(time.time()) + token = _sign( + { + "iss": ISSUER, + "aud": "other-client", + "sub": "alice", + "iat": now, + "exp": now + 60, + } + ) + with pytest.raises(IdTokenVerificationError): + await verify_id_token( + token, discovery_url=DISCOVERY_URL, expected_audience="test-client" + ) + + +async def test_verify_id_token_rejects_expired_token(mock_idp): + now = int(time.time()) + token = _sign( + { + "iss": ISSUER, + "aud": "test-client", + "sub": "alice", + "iat": now - 120, + "exp": now - 60, + } + ) + with pytest.raises(IdTokenVerificationError): + await verify_id_token( + token, discovery_url=DISCOVERY_URL, expected_audience="test-client" + ) + + +async def test_verify_id_token_rejects_wrong_issuer(mock_idp): + now = int(time.time()) + token = _sign( + { + "iss": "https://evil.example.com", + "aud": "test-client", + "sub": "alice", + "iat": now, + "exp": now + 60, + } + ) + with pytest.raises(IdTokenVerificationError): + await verify_id_token( + token, discovery_url=DISCOVERY_URL, expected_audience="test-client" + ) + + +async def test_verify_id_token_rejects_wrong_signature(mock_idp): + """Token signed with a different key but matching kid header must fail.""" + now = int(time.time()) + forged = _sign( + { + "iss": ISSUER, + "aud": "test-client", + "sub": "alice", + "iat": now, + "exp": now + 60, + }, + key_pem=_OTHER_PRIVATE_PEM, + ) + with pytest.raises(IdTokenVerificationError): + await verify_id_token( + forged, discovery_url=DISCOVERY_URL, expected_audience="test-client" + ) + + +async def test_verify_id_token_rejects_unknown_kid(mock_idp): + now = int(time.time()) + token = _sign( + { + "iss": ISSUER, + "aud": "test-client", + "sub": "alice", + "iat": now, + "exp": now + 60, + }, + kid="not-in-jwks", + ) + with pytest.raises(IdTokenVerificationError, match="No JWKS key matches"): + await verify_id_token( + token, discovery_url=DISCOVERY_URL, expected_audience="test-client" + ) + + +async def test_verify_id_token_nonce_mismatch_rejected(mock_idp): + now = int(time.time()) + token = _sign( + { + "iss": ISSUER, + "aud": "test-client", + "sub": "alice", + "iat": now, + "exp": now + 60, + "nonce": "actual", + } + ) + with pytest.raises(IdTokenVerificationError, match="nonce"): + await verify_id_token( + token, + discovery_url=DISCOVERY_URL, + expected_audience="test-client", + expected_nonce="expected", + ) + + +async def test_verify_id_token_missing_token_rejected(): + with pytest.raises(IdTokenVerificationError, match="missing"): + await verify_id_token( + "", discovery_url=DISCOVERY_URL, expected_audience="test-client" + ) diff --git a/tests/unit/test_oauth_logout.py b/tests/unit/test_oauth_logout.py new file mode 100644 index 00000000..57925581 --- /dev/null +++ b/tests/unit/test_oauth_logout.py @@ -0,0 +1,331 @@ +"""Unit tests for OAuth logout (issue #626 finding 4) and the +SessionAuthBackend (finding 2). + +These cover the new server-side session lifecycle: + - logout deletes refresh token + browser session + - logout calls IdP revocation_endpoint when available + - logout still succeeds when IdP/storage errors + - SessionAuthBackend resolves random session_id -> user_id, fails + closed when the session is unknown / expired / has no refresh token +""" + +import json +import tempfile +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch + +import httpx +import pytest +from cryptography.fernet import Fernet +from starlette.requests import HTTPConnection + +from nextcloud_mcp_server.auth.browser_oauth_routes import ( + _revoke_refresh_token_at_idp, + oauth_logout, +) +from nextcloud_mcp_server.auth.session_backend import SessionAuthBackend +from nextcloud_mcp_server.auth.storage import RefreshTokenStorage + +pytestmark = pytest.mark.unit + + +# --------------------------------------------------------------------------- +# storage fixture (real SQLite backend; lighter than mocking every call) +# --------------------------------------------------------------------------- + + +@pytest.fixture +async def storage(): + with tempfile.TemporaryDirectory() as tmpdir: + db_path = Path(tmpdir) / "test_logout.db" + s = RefreshTokenStorage( + db_path=str(db_path), encryption_key=Fernet.generate_key().decode() + ) + await s.initialize() + yield s + + +def _build_request(*, cookie: str | None, oauth_context: dict | None): + """Build a minimal Starlette-style request stub for oauth_logout.""" + request = MagicMock() + request.query_params = {} + request.cookies = {"mcp_session": cookie} if cookie else {} + request.app.state.oauth_context = oauth_context + return request + + +# --------------------------------------------------------------------------- +# oauth_logout +# --------------------------------------------------------------------------- + + +async def test_logout_deletes_refresh_token_and_session(storage): + """Happy path: logout removes the refresh token and the browser session.""" + await storage.create_browser_session(session_id="sid-1", user_id="alice") + await storage.store_refresh_token( + user_id="alice", refresh_token="rt-abc", flow_type="browser" + ) + + request = _build_request( + cookie="sid-1", + oauth_context={"storage": storage, "discovery_url": None}, + ) + + with patch( + "nextcloud_mcp_server.auth.browser_oauth_routes._revoke_refresh_token_at_idp", + new=AsyncMock(), + ): + response = await oauth_logout(request) + + assert response.status_code == 302 + assert await storage.get_refresh_token("alice") is None + assert await storage.get_browser_session_user("sid-1") is None + + +async def test_logout_calls_revocation_when_refresh_token_present(storage): + """The IdP revocation helper is called with the stored refresh token.""" + await storage.create_browser_session(session_id="sid-2", user_id="bob") + await storage.store_refresh_token( + user_id="bob", refresh_token="rt-xyz", flow_type="browser" + ) + + revoke = AsyncMock() + request = _build_request( + cookie="sid-2", + oauth_context={"storage": storage, "discovery_url": "http://idp/.well-known"}, + ) + + with patch( + "nextcloud_mcp_server.auth.browser_oauth_routes._revoke_refresh_token_at_idp", + new=revoke, + ): + await oauth_logout(request) + + revoke.assert_awaited_once() + args = revoke.await_args.args + # Second arg is the refresh token string + assert args[1] == "rt-xyz" + + +async def test_logout_no_session_cookie_returns_302(storage): + """Without a cookie, logout still 302s and doesn't touch storage.""" + request = _build_request( + cookie=None, oauth_context={"storage": storage, "discovery_url": None} + ) + response = await oauth_logout(request) + assert response.status_code == 302 + + +async def test_logout_swallows_storage_errors(storage): + """Logout is best-effort — a storage failure must not 500 the response.""" + await storage.create_browser_session(session_id="sid-3", user_id="carol") + broken_storage = MagicMock() + broken_storage.get_browser_session_user = AsyncMock( + side_effect=RuntimeError("db down") + ) + broken_storage.delete_browser_session = AsyncMock() + + request = _build_request( + cookie="sid-3", + oauth_context={"storage": broken_storage, "discovery_url": None}, + ) + response = await oauth_logout(request) + assert response.status_code == 302 # logout still succeeds + + +async def test_logout_handles_session_with_no_refresh_token(storage): + """Cookie + session row exist but refresh token already gone — logout is idempotent.""" + await storage.create_browser_session(session_id="sid-4", user_id="dave") + + revoke = AsyncMock() + request = _build_request( + cookie="sid-4", + oauth_context={"storage": storage, "discovery_url": None}, + ) + with patch( + "nextcloud_mcp_server.auth.browser_oauth_routes._revoke_refresh_token_at_idp", + new=revoke, + ): + await oauth_logout(request) + + # Revoke not called — no token to revoke + revoke.assert_not_called() + # Browser session still cleared + assert await storage.get_browser_session_user("sid-4") is None + + +# --------------------------------------------------------------------------- +# _revoke_refresh_token_at_idp +# --------------------------------------------------------------------------- + + +def _httpx_handler(routes: dict[str, httpx.Response]): + def handler(request: httpx.Request) -> httpx.Response: + return routes.get(str(request.url), httpx.Response(404)) + + return handler + + +async def test_revoke_helper_posts_to_revocation_endpoint(): + discovery_url = "http://idp.example/.well-known" + revocation_url = "http://idp.example/revoke" + + received: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + if str(request.url) == discovery_url: + return httpx.Response( + 200, + content=json.dumps({"revocation_endpoint": revocation_url}).encode(), + headers={"content-type": "application/json"}, + ) + if str(request.url) == revocation_url: + received.append(request) + return httpx.Response(200) + return httpx.Response(404) + + transport = httpx.MockTransport(handler) + + def fake_client(**kwargs): + kwargs["transport"] = transport + return httpx.AsyncClient(**kwargs) + + with patch( + "nextcloud_mcp_server.auth.browser_oauth_routes.nextcloud_httpx_client", + side_effect=fake_client, + ): + await _revoke_refresh_token_at_idp( + { + "discovery_url": discovery_url, + "client_id": "test-client", + "client_secret": "test-secret", + }, + "rt-secret", + ) + + assert len(received) == 1 + body = received[0].content.decode() + assert "token=rt-secret" in body + assert "token_type_hint=refresh_token" in body + + +async def test_revoke_helper_skips_when_no_revocation_endpoint(): + """IdPs without a revocation_endpoint advertised: helper must no-op silently.""" + discovery_url = "http://idp.example/.well-known" + + def handler(request: httpx.Request) -> httpx.Response: + if str(request.url) == discovery_url: + return httpx.Response(200, json={}) # no revocation_endpoint + return httpx.Response(404) + + transport = httpx.MockTransport(handler) + + def fake_client(**kwargs): + kwargs["transport"] = transport + return httpx.AsyncClient(**kwargs) + + with patch( + "nextcloud_mcp_server.auth.browser_oauth_routes.nextcloud_httpx_client", + side_effect=fake_client, + ): + # Returns None and does not raise + result = await _revoke_refresh_token_at_idp( + { + "discovery_url": discovery_url, + "client_id": "x", + "client_secret": "y", + }, + "rt", + ) + assert result is None + + +async def test_revoke_helper_silent_on_idp_error(): + """If the IdP 500s, the helper must not raise — caller treats it as best-effort.""" + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(500, content=b"boom") + + transport = httpx.MockTransport(handler) + + def fake_client(**kwargs): + kwargs["transport"] = transport + return httpx.AsyncClient(**kwargs) + + with patch( + "nextcloud_mcp_server.auth.browser_oauth_routes.nextcloud_httpx_client", + side_effect=fake_client, + ): + result = await _revoke_refresh_token_at_idp( + { + "discovery_url": "http://x/.well-known", + "client_id": "x", + "client_secret": "y", + }, + "rt", + ) + assert result is None + + +# --------------------------------------------------------------------------- +# SessionAuthBackend +# --------------------------------------------------------------------------- + + +def _build_conn(*, cookie: str | None, oauth_context: dict | None): + conn = MagicMock(spec=HTTPConnection) + conn.cookies = {"mcp_session": cookie} if cookie else {} + conn.url = SimpleNamespace(path="/app") + conn.app = MagicMock() + conn.app.state.oauth_context = oauth_context + return conn + + +async def test_session_backend_authenticates_known_session_with_token(storage): + await storage.create_browser_session(session_id="sid-A", user_id="alice") + await storage.store_refresh_token( + user_id="alice", refresh_token="rt", flow_type="browser" + ) + + backend = SessionAuthBackend(oauth_enabled=True) + conn = _build_conn(cookie="sid-A", oauth_context={"storage": storage}) + + result = await backend.authenticate(conn) + assert result is not None + creds, user = result + assert "authenticated" in creds.scopes + assert user.username == "alice" + + +async def test_session_backend_rejects_unknown_session(storage): + backend = SessionAuthBackend(oauth_enabled=True) + conn = _build_conn(cookie="not-a-real-sid", oauth_context={"storage": storage}) + assert await backend.authenticate(conn) is None + + +async def test_session_backend_rejects_session_without_refresh_token(storage): + """Defense-in-depth: session row exists but user has no refresh token.""" + await storage.create_browser_session(session_id="sid-B", user_id="bob") + # Note: NO refresh token stored for bob + + backend = SessionAuthBackend(oauth_enabled=True) + conn = _build_conn(cookie="sid-B", oauth_context={"storage": storage}) + assert await backend.authenticate(conn) is None + + +async def test_session_backend_rejects_when_no_cookie(storage): + backend = SessionAuthBackend(oauth_enabled=True) + conn = _build_conn(cookie=None, oauth_context={"storage": storage}) + assert await backend.authenticate(conn) is None + + +async def test_session_backend_basicauth_mode_short_circuits(monkeypatch, storage): + """In BasicAuth mode (oauth_enabled=False) the backend never touches storage.""" + monkeypatch.setenv("NEXTCLOUD_USERNAME", "admin-user") + backend = SessionAuthBackend(oauth_enabled=False) + conn = _build_conn(cookie=None, oauth_context=None) + result = await backend.authenticate(conn) + assert result is not None + _, user = result + assert user.username == "admin-user" diff --git a/tests/unit/test_oauth_tools_signatures.py b/tests/unit/test_oauth_tools_signatures.py new file mode 100644 index 00000000..cbeba12b --- /dev/null +++ b/tests/unit/test_oauth_tools_signatures.py @@ -0,0 +1,56 @@ +"""Unit tests for OAuth tool input-schema hardening (issue #626 finding 3). + +These tools must derive `user_id` from the verified MCP access token and +must never accept it as an MCP-level input. Otherwise an LLM (or any MCP +client) could supply an arbitrary user_id and reach cross-user revoke or +status-disclosure operations. +""" + +import pytest +from mcp.server.fastmcp import FastMCP + +from nextcloud_mcp_server.server.oauth_tools import register_oauth_tools + +pytestmark = pytest.mark.unit + + +HARDENED_TOOLS = ( + "provision_nextcloud_access", + "revoke_nextcloud_access", + "check_provisioning_status", + "check_logged_in", +) + + +@pytest.fixture +def registered_tools(): + """Register the OAuth tools against a fresh FastMCP and return them by name. + + Uses FastMCP's `_tool_manager.list_tools()`; flagged as internal and may + break on SDK upgrades, but this is the supported way to inspect a tool's + JSON input schema in unit tests (see tests/unit/test_stdio.py). + """ + mcp = FastMCP("test-oauth-tools") + register_oauth_tools(mcp) + tools = mcp._tool_manager.list_tools() + return {t.name: t for t in tools} + + +def test_oauth_tools_registered(registered_tools): + for name in HARDENED_TOOLS: + assert name in registered_tools, f"{name} should be registered" + + +@pytest.mark.parametrize("tool_name", HARDENED_TOOLS) +def test_oauth_tool_schema_does_not_accept_user_id(tool_name, registered_tools): + """user_id must not appear in the tool's JSON input schema.""" + tool = registered_tools[tool_name] + properties = tool.parameters.get("properties", {}) + required = tool.parameters.get("required", []) + + assert "user_id" not in properties, ( + f"{tool_name} accepts user_id as an MCP input — must be derived from " + f"the verified access token (issue #626 finding 3). " + f"properties={list(properties.keys())}" + ) + assert "user_id" not in required diff --git a/tests/unit/test_token_broker.py b/tests/unit/test_token_broker.py index 32105b2e..8aebcd1a 100644 --- a/tests/unit/test_token_broker.py +++ b/tests/unit/test_token_broker.py @@ -10,7 +10,6 @@ from datetime import datetime, timedelta, timezone from unittest.mock import AsyncMock, MagicMock, patch import httpx -import jwt import pytest from cryptography.fernet import Fernet @@ -283,23 +282,6 @@ class TestTokenBrokerService: # Verify cache was cleared assert await token_broker.cache.get("user1") is None - async def test_validate_token_audience(self, token_broker): - """Test token audience validation.""" - # Create test token with audience - test_payload = { - "sub": "user1", - "aud": ["nextcloud", "other-service"], - "exp": datetime.now(timezone.utc) + timedelta(hours=1), - } - test_token = jwt.encode(test_payload, "secret", algorithm="HS256") - - # Should not raise for correct audience - await token_broker._validate_token_audience(test_token, "nextcloud") - - # Should raise for wrong audience - with pytest.raises(ValueError, match="doesn't include wrong-audience"): - await token_broker._validate_token_audience(test_token, "wrong-audience") - async def test_token_refresh_with_network_error(self, token_broker, mock_storage): """Test handling network errors during token refresh.""" # Storage returns already-decrypted refresh token