Chris Coutinho
db1e0606ad
Consolidate three independent RefreshTokenStorage lazy singletons into a single lock-protected get_shared_storage() function, eliminating race conditions on concurrent first-access. Remove blanket try/except in _get_stored_scopes so storage errors propagate as proper MCP errors instead of silently triggering "please provision" messages. Handle declined/cancelled elicitation results in Login Flow tools by cleaning up sessions and returning clear status. Add update_app_password_scopes() to avoid unnecessary decrypt/re-encrypt when only scopes change. Add unprovisioned-user early exit and no-op detection to nc_auth_update_scopes. Remove four dead config fields and misleading NEXTCLOUD_PASSWORD deprecation warning. Add periodic login flow session cleanup task. Generate separate Fernet keys per service. Add board cleanup in deck integration test. Gate CI unit tests on linting and skip Astrolabe build for single-user profile. Fix test markers from oauth to multi_user_basic for astrolabe integration tests. Update login_flow.py docstrings to document outbound HTTP calls. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
501 lines
18 KiB
Python
501 lines
18 KiB
Python
"""Scope-based authorization for MCP tools."""
|
|
|
|
import logging
|
|
from functools import wraps
|
|
from typing import Any, Callable
|
|
|
|
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 mcp.server.fastmcp.utilities.context_injection import find_context_parameter
|
|
|
|
from nextcloud_mcp_server.auth.storage import get_shared_storage
|
|
from nextcloud_mcp_server.config import get_settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ScopeAuthorizationError(Exception):
|
|
"""Raised when a request lacks required scopes."""
|
|
|
|
pass
|
|
|
|
|
|
class InsufficientScopeError(ScopeAuthorizationError):
|
|
"""Raised when request lacks required scopes (enables step-up auth).
|
|
|
|
This exception triggers a 403 response with WWW-Authenticate header
|
|
containing the missing scopes, allowing clients to perform step-up
|
|
authorization to obtain additional permissions.
|
|
"""
|
|
|
|
def __init__(self, missing_scopes: list[str], message: str | None = None):
|
|
self.missing_scopes = missing_scopes
|
|
super().__init__(
|
|
message or f"Missing required scopes: {', '.join(missing_scopes)}"
|
|
)
|
|
|
|
|
|
class ProvisioningRequiredError(ScopeAuthorizationError):
|
|
"""Raised when Nextcloud resource access requires provisioning (Flow 2).
|
|
|
|
In Progressive Consent mode, users must explicitly provision Nextcloud
|
|
access using the provision_nextcloud_access MCP tool.
|
|
"""
|
|
|
|
def __init__(self, message: str | None = None):
|
|
super().__init__(
|
|
message
|
|
or (
|
|
"Nextcloud resource access not provisioned. "
|
|
"Please run the 'provision_nextcloud_access' tool to grant access."
|
|
)
|
|
)
|
|
|
|
|
|
def require_scopes(*required_scopes: str):
|
|
"""
|
|
Decorator to require specific OAuth scopes for MCP tool execution.
|
|
|
|
This decorator:
|
|
1. Stores scope requirements as function metadata (_required_scopes attribute)
|
|
2. Checks that the access token contains all required scopes before execution
|
|
3. Raises ScopeAuthorizationError if any required scope is missing
|
|
|
|
The stored metadata enables dynamic tool filtering - tools can be hidden from
|
|
users who lack the necessary scopes.
|
|
|
|
Args:
|
|
*required_scopes: Variable number of scope strings required (e.g., "notes:read", "notes:write")
|
|
|
|
Returns:
|
|
Decorated function that checks scopes before execution
|
|
|
|
Example:
|
|
```python
|
|
@mcp.tool()
|
|
@require_scopes("notes:read")
|
|
async def nc_notes_get_note(ctx: Context, note_id: int):
|
|
# This tool requires the notes:read scope
|
|
...
|
|
|
|
@mcp.tool()
|
|
@require_scopes("notes:write")
|
|
async def nc_notes_create_note(ctx: Context, ...):
|
|
# This tool requires the notes:write scope
|
|
...
|
|
```
|
|
|
|
Raises:
|
|
ScopeAuthorizationError: If required scopes are not present in the access token
|
|
"""
|
|
|
|
def decorator(func: Callable) -> Callable:
|
|
# Store scope requirements as function metadata for dynamic filtering
|
|
func._required_scopes = list(required_scopes) # type: ignore[attr-defined]
|
|
|
|
# Get function name for logging (works for any callable)
|
|
func_name = getattr(func, "__name__", repr(func))
|
|
|
|
# Find which parameter receives the Context (FastMCP injects it by name)
|
|
context_param_name = find_context_parameter(func)
|
|
|
|
@wraps(func)
|
|
async def wrapper(*args: Any, **kwargs: Any) -> Any:
|
|
# Extract context from kwargs (where FastMCP injected it)
|
|
ctx: Context | None = (
|
|
kwargs.get(context_param_name) if context_param_name else None
|
|
)
|
|
|
|
if ctx is None:
|
|
# No context parameter found - likely BasicAuth mode
|
|
# In BasicAuth mode, all operations are allowed
|
|
logger.debug(
|
|
f"No context parameter for {func_name} - allowing (BasicAuth mode)"
|
|
)
|
|
return await func(*args, **kwargs)
|
|
|
|
# Check if we're in OAuth mode (access token available)
|
|
access_token: AccessToken | None = getattr(
|
|
ctx.request_context, "access_token", None
|
|
)
|
|
|
|
if access_token is None:
|
|
# Check if single-user BasicAuth mode (env var app password)
|
|
# If NEXTCLOUD_APP_PASSWORD or NEXTCLOUD_PASSWORD is set, bypass scope checks
|
|
settings = get_settings()
|
|
if settings.nextcloud_app_password or settings.nextcloud_password:
|
|
logger.debug(
|
|
f"No access token for {func_name} - allowing (env var app password)"
|
|
)
|
|
return await func(*args, **kwargs)
|
|
|
|
# Not in OAuth mode (BasicAuth or no auth)
|
|
# In BasicAuth mode, all operations are allowed
|
|
logger.debug(
|
|
f"No access token present for {func_name} - allowing (BasicAuth mode)"
|
|
)
|
|
return await func(*args, **kwargs)
|
|
|
|
# ── Login Flow v2: Check stored app password scopes ──
|
|
# In Login Flow v2 multi-user mode, OAuth tokens provide MCP session
|
|
# identity only. Nextcloud API access uses stored app passwords.
|
|
# Check if the user has a stored app password with appropriate scopes.
|
|
if get_settings().enable_login_flow:
|
|
from nextcloud_mcp_server.server.oauth_tools import ( # noqa: PLC0415
|
|
extract_user_id_from_token,
|
|
)
|
|
|
|
user_id = await extract_user_id_from_token(ctx)
|
|
if user_id and user_id != "default_user":
|
|
stored_scopes = await _get_stored_scopes(user_id)
|
|
|
|
if stored_scopes is None:
|
|
# No stored app password → require provisioning
|
|
error_msg = (
|
|
f"Access denied to {func_name}: "
|
|
f"Nextcloud access not provisioned. "
|
|
f"Please call 'nc_auth_provision_access' first."
|
|
)
|
|
logger.warning(error_msg)
|
|
raise ProvisioningRequiredError(error_msg)
|
|
|
|
if stored_scopes == "all":
|
|
# NULL scopes in DB = legacy app password = all allowed
|
|
logger.debug(
|
|
f"Stored app password scope check passed for {func_name}: all scopes"
|
|
)
|
|
return await func(*args, **kwargs)
|
|
|
|
# Check stored scopes against required
|
|
stored_set = set(stored_scopes)
|
|
missing = set(required_scopes) - stored_set
|
|
if missing:
|
|
error_msg = (
|
|
f"Access denied to {func_name}: "
|
|
f"Missing scopes: {', '.join(sorted(missing))}. "
|
|
f"Call 'nc_auth_update_scopes' to add permissions."
|
|
)
|
|
logger.warning(error_msg)
|
|
raise InsufficientScopeError(list(missing), error_msg)
|
|
|
|
logger.debug(
|
|
f"Stored app password scope check passed for {func_name}"
|
|
)
|
|
return await func(*args, **kwargs)
|
|
|
|
# Extract scopes from access token
|
|
token_scopes = set(access_token.scopes or [])
|
|
required_scopes_set = set(required_scopes)
|
|
|
|
# Check if offline access is enabled
|
|
# Use settings.enable_offline_access which handles both ENABLE_BACKGROUND_OPERATIONS (new)
|
|
# and ENABLE_OFFLINE_ACCESS (deprecated) environment variables
|
|
settings = get_settings()
|
|
enable_offline_access = settings.enable_offline_access
|
|
|
|
# In offline access mode, check if Nextcloud scopes require provisioning
|
|
if enable_offline_access:
|
|
# Check if any required scopes are Nextcloud-specific
|
|
nextcloud_scopes = [
|
|
s
|
|
for s in required_scopes
|
|
if any(
|
|
s.startswith(prefix)
|
|
for prefix in [
|
|
"notes:",
|
|
"calendar:",
|
|
"contacts:",
|
|
"files:",
|
|
"tables:",
|
|
"deck:",
|
|
]
|
|
)
|
|
]
|
|
|
|
if nextcloud_scopes:
|
|
# Check if user has completed Flow 2 provisioning
|
|
# This would be indicated by having a stored refresh token
|
|
# In production, we'd check the token broker or storage
|
|
# For now, we check if the token has the required scopes
|
|
# (Flow 1 tokens won't have Nextcloud scopes)
|
|
has_nextcloud_scopes = any(
|
|
s.startswith(prefix)
|
|
for s in token_scopes
|
|
for prefix in [
|
|
"notes:",
|
|
"calendar:",
|
|
"contacts:",
|
|
"files:",
|
|
"tables:",
|
|
"deck:",
|
|
]
|
|
)
|
|
|
|
if not has_nextcloud_scopes:
|
|
error_msg = (
|
|
f"Access denied to {func_name}: "
|
|
f"Nextcloud resource access not provisioned. "
|
|
f"Please run the 'provision_nextcloud_access' tool first."
|
|
)
|
|
logger.warning(error_msg)
|
|
raise ProvisioningRequiredError(error_msg)
|
|
|
|
# Check if all required scopes are present
|
|
missing_scopes = required_scopes_set - token_scopes
|
|
if missing_scopes:
|
|
error_msg = (
|
|
f"Access denied to {func_name}: "
|
|
f"Missing required scopes: {', '.join(sorted(missing_scopes))}. "
|
|
f"Token has scopes: {', '.join(sorted(token_scopes)) if token_scopes else 'none'}"
|
|
)
|
|
logger.warning(error_msg)
|
|
raise InsufficientScopeError(list(missing_scopes), error_msg)
|
|
|
|
# All required scopes present - allow execution
|
|
logger.debug(
|
|
f"Scope authorization passed for {func_name}: {required_scopes}"
|
|
)
|
|
return await func(*args, **kwargs)
|
|
|
|
return wrapper
|
|
|
|
return decorator
|
|
|
|
|
|
def get_access_token_scopes(ctx: Context | None = None) -> set[str]:
|
|
"""
|
|
Extract scopes from the authenticated user's access token.
|
|
|
|
This function uses MCP SDK's contextvar to access the token, which works
|
|
across all request types including list_tools.
|
|
|
|
Args:
|
|
ctx: FastMCP context object (unused, kept for compatibility)
|
|
|
|
Returns:
|
|
Set of scope strings, empty set if no token or no scopes
|
|
"""
|
|
# Use MCP SDK's get_access_token() which uses contextvars
|
|
# This works for all request types, including list_tools
|
|
access_token: AccessToken | None = get_access_token()
|
|
|
|
if access_token is None:
|
|
logger.debug("No access token found in auth context (likely BasicAuth mode)")
|
|
return set()
|
|
|
|
scopes = set(access_token.scopes or [])
|
|
logger.info(f"✅ Extracted scopes from access token: {scopes}")
|
|
return scopes
|
|
|
|
|
|
def check_scopes(ctx: Context, *required_scopes: str) -> tuple[bool, set[str]]:
|
|
"""
|
|
Check if the request context has all required scopes.
|
|
|
|
Utility function for manual scope checking without decorator.
|
|
|
|
Args:
|
|
ctx: FastMCP context object
|
|
*required_scopes: Variable number of required scope strings
|
|
|
|
Returns:
|
|
Tuple of (has_all_scopes: bool, missing_scopes: set[str])
|
|
|
|
Example:
|
|
```python
|
|
async def my_tool(ctx: Context):
|
|
has_scopes, missing = check_scopes(ctx, "notes:read", "notes:write")
|
|
if not has_scopes:
|
|
# Handle missing scopes
|
|
...
|
|
```
|
|
"""
|
|
token_scopes = get_access_token_scopes(ctx)
|
|
|
|
# If no access token, assume BasicAuth mode (all operations allowed)
|
|
if not token_scopes and getattr(ctx.request_context, "access_token", None) is None:
|
|
return True, set()
|
|
|
|
required_scopes_set = set(required_scopes)
|
|
missing_scopes = required_scopes_set - token_scopes
|
|
|
|
return len(missing_scopes) == 0, missing_scopes
|
|
|
|
|
|
def get_required_scopes(func: Callable) -> list[str]:
|
|
"""
|
|
Extract required scopes from a function decorated with @require_scopes.
|
|
|
|
Args:
|
|
func: Function to check (may be decorated)
|
|
|
|
Returns:
|
|
List of required scope strings, empty list if no scopes required
|
|
|
|
Example:
|
|
```python
|
|
@require_scopes("notes:read", "notes:write")
|
|
async def my_tool():
|
|
pass
|
|
|
|
scopes = get_required_scopes(my_tool) # ["notes:read", "notes:write"]
|
|
```
|
|
"""
|
|
return getattr(func, "_required_scopes", [])
|
|
|
|
|
|
def is_jwt_token() -> bool:
|
|
"""
|
|
Check if the current access token is in JWT format.
|
|
|
|
JWT tokens have 3 parts separated by dots (header.payload.signature).
|
|
Opaque tokens are random strings without this structure.
|
|
|
|
Returns:
|
|
True if current token is JWT format, False if opaque or no token
|
|
"""
|
|
access_token: AccessToken | None = get_access_token()
|
|
|
|
if access_token is None:
|
|
logger.debug("No access token found - not JWT")
|
|
return False
|
|
|
|
# JWT tokens have exactly 2 dots (3 parts)
|
|
token_string = access_token.token
|
|
is_jwt = "." in token_string and token_string.count(".") == 2
|
|
|
|
logger.debug(f"Token format check: is_jwt={is_jwt}")
|
|
return is_jwt
|
|
|
|
|
|
def has_required_scopes(func: Callable, user_scopes: set[str]) -> bool:
|
|
"""
|
|
Check if a user has all scopes required by a function.
|
|
|
|
Used for dynamic tool filtering - determines if a tool should be visible
|
|
to a user based on their token scopes.
|
|
|
|
Args:
|
|
func: Function decorated with @require_scopes
|
|
user_scopes: Set of scopes the user possesses
|
|
|
|
Returns:
|
|
True if user has all required scopes (or no scopes required), False otherwise
|
|
|
|
Example:
|
|
```python
|
|
@require_scopes("notes:write")
|
|
async def create_note():
|
|
pass
|
|
|
|
user_scopes = {"notes:read", "notes:write"}
|
|
can_see = has_required_scopes(create_note, user_scopes) # True
|
|
|
|
limited_user_scopes = {"notes:read"}
|
|
can_see = has_required_scopes(create_note, limited_user_scopes) # False
|
|
```
|
|
"""
|
|
required = get_required_scopes(func)
|
|
|
|
# No scopes required → always allow
|
|
if not required:
|
|
return True
|
|
|
|
# Empty user_scopes but scopes required → deny
|
|
if not user_scopes:
|
|
return False
|
|
|
|
# Check if user has all required scopes
|
|
return set(required).issubset(user_scopes)
|
|
|
|
|
|
def discover_all_scopes(mcp) -> list[str]:
|
|
"""
|
|
Dynamically discover all OAuth scopes required by registered MCP tools.
|
|
|
|
This function inspects all registered tools and extracts their required scopes
|
|
from the @require_scopes decorator metadata. It provides a single source of truth
|
|
for available scopes based on the actual tool implementations.
|
|
|
|
Args:
|
|
mcp: FastMCP instance with registered tools
|
|
|
|
Returns:
|
|
Sorted list of unique scope strings, including base OIDC scopes
|
|
|
|
Example:
|
|
```python
|
|
from mcp.server.fastmcp import FastMCP
|
|
|
|
mcp = FastMCP("My Server")
|
|
|
|
@mcp.tool()
|
|
@require_scopes("notes:read")
|
|
async def get_notes():
|
|
pass
|
|
|
|
@mcp.tool()
|
|
@require_scopes("notes:write")
|
|
async def create_note():
|
|
pass
|
|
|
|
scopes = discover_all_scopes(mcp)
|
|
# Returns: ["notes:read", "notes:write", "openid", "profile", "email"]
|
|
```
|
|
|
|
Note:
|
|
- Base OIDC scopes (openid, profile, email) are always included
|
|
- Scopes are deduplicated and sorted alphabetically
|
|
- Only scopes from decorated tools are included
|
|
- Must be called after tools are registered
|
|
"""
|
|
# Start with base OIDC scopes that are always required
|
|
all_scopes = {"openid", "profile", "email"}
|
|
|
|
# Get all registered tools
|
|
try:
|
|
tools = mcp._tool_manager.list_tools()
|
|
except AttributeError:
|
|
logger.warning("FastMCP instance does not have _tool_manager attribute")
|
|
return sorted(all_scopes)
|
|
|
|
# Extract scopes from each tool
|
|
for tool in tools:
|
|
# Get the original function (tools have a .fn attribute)
|
|
func = getattr(tool, "fn", None)
|
|
if func is None:
|
|
continue
|
|
|
|
# Extract scopes using existing helper
|
|
tool_scopes = get_required_scopes(func)
|
|
all_scopes.update(tool_scopes)
|
|
|
|
# Return sorted list of unique scopes
|
|
return sorted(all_scopes)
|
|
|
|
|
|
# ── Login Flow v2 helpers ────────────────────────────────────────────────
|
|
|
|
|
|
async def _get_stored_scopes(user_id: str) -> list[str] | str | None:
|
|
"""Look up stored app password scopes for a user.
|
|
|
|
Returns:
|
|
- list[str]: Specific scopes granted
|
|
- "all": NULL scopes in DB (legacy = all allowed)
|
|
- None: No stored app password (provisioning required)
|
|
|
|
Raises:
|
|
Storage/infrastructure exceptions propagate to the caller
|
|
(require_scopes decorator) for proper MCP error responses.
|
|
"""
|
|
storage = await get_shared_storage()
|
|
|
|
data = await storage.get_app_password_with_scopes(user_id)
|
|
if data is None:
|
|
return None
|
|
if data["scopes"] is None:
|
|
return "all"
|
|
return data["scopes"]
|