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

fix: conditionally include offline_access based on IdP discovery

Browse Source 
AWS Cognito provides refresh tokens automatically with the authorization
code flow but does not list offline_access as a supported scope. Check
the IdP's scopes_supported discovery field before including it in
requests, and always accept refresh tokens from responses regardless.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2026-04-07 23:48:49 +02:00
co-authored by Claude Opus 4.6
parent 776e2ce693
commit 7730f926cb
5 changed files with 331 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nextcloud_mcp_server/auth/browser_oauth_routes.py
+17 -2
View File
@@ -85,10 +85,11 @@ async def oauth_login(request: Request) -> RedirectResponse | JSONResponse:
mcp_server_url = oauth_config["mcp_server_url"]
callback_uri = f"{mcp_server_url}/oauth/callback"
# Request only basic OIDC scopes for browser session
# Request only basic OIDC scopes for browser session.
# offline_access is added conditionally below based on IdP discovery.
# 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"
scopes = "openid profile email"
# Generate PKCE values for ALL modes (both external and integrated IdP require PKCE)
code_verifier = secrets.token_urlsafe(32)
@@ -113,6 +114,12 @@ async def oauth_login(request: Request) -> RedirectResponse | JSONResponse:
if not oauth_client.authorization_endpoint:
await oauth_client.discover()
# Check if IdP supports offline_access via server metadata from discovery
idp_metadata = getattr(oauth_client, "server_metadata", None) or {}
idp_scopes = idp_metadata.get("scopes_supported")
if idp_scopes is None or "offline_access" in idp_scopes:
scopes += " offline_access"
# Get Nextcloud resource URI for audience (background sync needs Nextcloud-scoped tokens)
nextcloud_resource_uri = oauth_config.get(
"nextcloud_resource_uri", oauth_config.get("nextcloud_host")
@@ -151,6 +158,14 @@ async def oauth_login(request: Request) -> RedirectResponse | JSONResponse:
discovery = response.json()
authorization_endpoint = discovery["authorization_endpoint"]
# Include offline_access only if the IdP advertises it (or if
# scopes_supported is absent from the discovery document).
# IdPs like AWS Cognito provide refresh tokens automatically without
# supporting the offline_access scope.
idp_scopes = discovery.get("scopes_supported")
if idp_scopes is None or "offline_access" in idp_scopes:
scopes += " offline_access"
# Replace internal Docker hostname with public URL
public_issuer = os.getenv("NEXTCLOUD_PUBLIC_ISSUER_URL")
if public_issuer:
nextcloud_mcp_server/auth/oauth_routes.py
+11 -1
View File
@@ -456,7 +456,17 @@ async def oauth_authorize_nextcloud(
# Resource scopes are requested by client in Flow 1
scopes = "openid profile email"
if get_settings().enable_offline_access:
scopes += " offline_access"
# Only include offline_access if the IdP advertises it in scopes_supported.
# IdPs like AWS Cognito provide refresh tokens automatically without
# supporting the offline_access scope.
discovery_url = oauth_config.get("discovery_url")
if discovery_url:
disc = await _get_cached_discovery(discovery_url)
scopes_supported = disc.get("scopes_supported")
if scopes_supported is None or "offline_access" in scopes_supported:
scopes += " offline_access"
else:
scopes += " offline_access"
# Generate PKCE values (required by Nextcloud OIDC)
code_verifier = secrets.token_urlsafe(32)
nextcloud_mcp_server/auth/token_broker.py
+43 -6
View File
@@ -169,6 +169,24 @@ class TokenBrokerService:
self._oidc_config = response.json()
return self._oidc_config
async def _idp_supports_offline_access(self) -> bool:
"""Check if the IdP advertises ``offline_access`` in ``scopes_supported``.
Returns ``True`` when ``offline_access`` is explicitly listed **or**
when ``scopes_supported`` is absent from the discovery document (the
field is OPTIONAL per the OIDC spec, so absence means unknown —
include ``offline_access`` as a safe default).
Returns ``False`` when ``scopes_supported`` is present but does **not**
include ``offline_access`` (e.g. AWS Cognito, which provides refresh
tokens automatically without requiring the scope).
"""
config = await self._get_oidc_config()
scopes_supported = config.get("scopes_supported")
if scopes_supported is None:
return True
return "offline_access" in scopes_supported
async def get_nextcloud_token(self, user_id: str) -> Optional[str]:
"""
Get a valid Nextcloud access token for the user.
@@ -334,10 +352,18 @@ class TokenBrokerService:
# Request new access token using refresh token
# Include client credentials as required by most OAuth servers
# Only request offline_access if the IdP advertises it (e.g. Cognito does not)
base_scopes = ["openid", "profile", "email"]
if await self._idp_supports_offline_access():
base_scopes.append("offline_access")
scope_str = " ".join(
base_scopes
+ ["notes.read", "notes.write", "calendar.read", "calendar.write"]
)
data = {
"grant_type": "refresh_token",
"refresh_token": refresh_token,
"scope": "openid profile email offline_access notes.read notes.write calendar.read calendar.write",
"scope": scope_str,
"client_id": self.client_id,
"client_secret": self.client_secret,
}
@@ -402,10 +428,13 @@ class TokenBrokerService:
client = await self._get_http_client()
# Always include basic OpenID scopes + offline_access to get new refresh token
scopes = list(
set(["openid", "profile", "email", "offline_access"] + required_scopes)
)
# Always include basic OpenID scopes; only add offline_access if the IdP
# advertises it (e.g. AWS Cognito provides refresh tokens automatically
# without supporting the offline_access scope).
base_scopes = ["openid", "profile", "email"]
if await self._idp_supports_offline_access():
base_scopes.append("offline_access")
scopes = list(set(base_scopes + required_scopes))
# Request new access token with specific scopes
# Include client credentials as required by most OAuth servers
@@ -518,10 +547,18 @@ class TokenBrokerService:
client = await self._get_http_client()
# Request new refresh token
# Only request offline_access if the IdP advertises it
base_scopes = ["openid", "profile", "email"]
if await self._idp_supports_offline_access():
base_scopes.append("offline_access")
scope_str = " ".join(
base_scopes
+ ["notes.read", "notes.write", "calendar.read", "calendar.write"]
)
data = {
"grant_type": "refresh_token",
"refresh_token": current_refresh_token,
"scope": "openid profile email offline_access notes.read notes.write calendar.read calendar.write",
"scope": scope_str,
}
response = await client.post(
nextcloud_mcp_server/server/oauth_tools.py
+6 -1
View File
@@ -468,11 +468,14 @@ async def check_logged_in(ctx: Context, user_id: Optional[str] = None) -> str:
)
# Define scopes for Nextcloud access
# Note: offline_access is only included when enabled in settings.
# The actual scope sent to the IdP is determined by
# oauth_authorize_nextcloud() based on IdP discovery, so this list
# is informational (generate_oauth_url_for_flow2 marks it as unused).
scopes = [
"openid",
"profile",
"email",
"offline_access", # Critical for background operations
"notes.read",
"notes.write",
"calendar.read",
@@ -482,6 +485,8 @@ async def check_logged_in(ctx: Context, user_id: Optional[str] = None) -> str:
"files.read",
"files.write",
]
if get_settings().enable_offline_access:
scopes.insert(3, "offline_access")
# Generate authorization URL
auth_url = generate_oauth_url_for_flow2(
tests/unit/test_offline_access_detection.py
+254
View File
@@ -0,0 +1,254 @@
"""Tests for discovery-driven offline_access scope handling.
Verifies that the server conditionally includes ``offline_access`` in OAuth
scope requests based on the IdP's ``scopes_supported`` discovery field, and
that refresh tokens are always accepted from responses regardless of whether
``offline_access`` was requested (AWS Cognito behavior).
"""
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from nextcloud_mcp_server.auth.token_broker import TokenBrokerService
pytestmark = pytest.mark.unit
@pytest.fixture
def mock_storage():
"""Mock RefreshTokenStorage."""
storage = AsyncMock()
storage.get_refresh_token = AsyncMock(return_value=None)
storage.store_refresh_token = AsyncMock()
storage.delete_refresh_token = AsyncMock()
return storage
def _make_broker(mock_storage):
"""Create a TokenBrokerService instance for testing."""
return TokenBrokerService(
storage=mock_storage,
oidc_discovery_url="https://idp.example.com/.well-known/openid-configuration",
nextcloud_host="https://nextcloud.example.com",
client_id="test_client_id",
client_secret="test_client_secret",
cache_ttl=300,
)
class TestIdpSupportsOfflineAccess:
"""Test _idp_supports_offline_access() discovery-driven detection."""
async def test_supports_when_listed(self, mock_storage):
"""Returns True when scopes_supported includes offline_access."""
broker = _make_broker(mock_storage)
discovery = {
"token_endpoint": "https://idp.example.com/token",
"scopes_supported": ["openid", "profile", "email", "offline_access"],
}
with patch.object(broker, "_get_oidc_config", return_value=discovery):
assert await broker._idp_supports_offline_access() is True
await broker.close()
async def test_not_supported_when_absent_from_list(self, mock_storage):
"""Returns False when scopes_supported is present but lacks offline_access (Cognito)."""
broker = _make_broker(mock_storage)
discovery = {
"token_endpoint": "https://idp.example.com/token",
"scopes_supported": ["openid"],
}
with patch.object(broker, "_get_oidc_config", return_value=discovery):
assert await broker._idp_supports_offline_access() is False
await broker.close()
async def test_supports_when_field_missing(self, mock_storage):
"""Returns True (safe default) when scopes_supported is absent from discovery."""
broker = _make_broker(mock_storage)
discovery = {
"token_endpoint": "https://idp.example.com/token",
# No scopes_supported field at all
}
with patch.object(broker, "_get_oidc_config", return_value=discovery):
assert await broker._idp_supports_offline_access() is True
await broker.close()
class TestRefreshScopeConditional:
"""Test that refresh methods conditionally include offline_access."""
async def test_refresh_omits_offline_access_for_cognito(self, mock_storage):
"""When IdP does not support offline_access, scope string omits it."""
broker = _make_broker(mock_storage)
discovery = {
"token_endpoint": "https://idp.example.com/token",
"scopes_supported": ["openid", "profile", "email"],
}
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {
"access_token": "new_access_token",
"expires_in": 3600,
}
with patch.object(broker, "_get_oidc_config", return_value=discovery):
with patch.object(broker, "_get_http_client") as mock_client:
mock_post = AsyncMock(return_value=mock_response)
mock_client.return_value.post = mock_post
await broker._refresh_access_token_with_scopes(
"test_refresh", ["notes.read"], user_id=None
)
# Verify the scope sent in the POST
call_kwargs = mock_post.call_args
posted_data = call_kwargs.kwargs.get("data") or call_kwargs[1].get(
"data"
)
scope_str = posted_data["scope"]
assert "offline_access" not in scope_str
assert "openid" in scope_str
assert "notes.read" in scope_str
await broker.close()
async def test_refresh_includes_offline_access_for_nextcloud(self, mock_storage):
"""When IdP supports offline_access, scope string includes it."""
broker = _make_broker(mock_storage)
discovery = {
"token_endpoint": "https://idp.example.com/token",
"scopes_supported": ["openid", "profile", "email", "offline_access"],
}
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {
"access_token": "new_access_token",
"expires_in": 3600,
}
with patch.object(broker, "_get_oidc_config", return_value=discovery):
with patch.object(broker, "_get_http_client") as mock_client:
mock_post = AsyncMock(return_value=mock_response)
mock_client.return_value.post = mock_post
await broker._refresh_access_token_with_scopes(
"test_refresh", ["notes.read"], user_id=None
)
call_kwargs = mock_post.call_args
posted_data = call_kwargs.kwargs.get("data") or call_kwargs[1].get(
"data"
)
scope_str = posted_data["scope"]
assert "offline_access" in scope_str
assert "openid" in scope_str
await broker.close()
async def test_refresh_token_stored_without_offline_access_scope(
self, mock_storage
):
"""Refresh token from response is stored even when offline_access wasn't requested.
This is the key Cognito scenario: the IdP returns a refresh token
automatically even though offline_access was not in the scope request.
"""
broker = _make_broker(mock_storage)
# Cognito-like discovery: no offline_access in scopes_supported
discovery = {
"token_endpoint": "https://idp.example.com/token",
"scopes_supported": ["openid", "profile", "email"],
}
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {
"access_token": "new_access_token",
"expires_in": 3600,
# Cognito returns refresh token automatically
"refresh_token": "rotated_refresh_token",
}
with patch.object(broker, "_get_oidc_config", return_value=discovery):
with patch.object(broker, "_get_http_client") as mock_client:
mock_client.return_value.post = AsyncMock(return_value=mock_response)
await broker._refresh_access_token_with_scopes(
"old_refresh_token", ["notes.read"], user_id="user1"
)
# Verify the rotated refresh token was stored
mock_storage.store_refresh_token.assert_called_once()
call_kwargs = mock_storage.store_refresh_token.call_args[1]
assert call_kwargs["user_id"] == "user1"
assert call_kwargs["refresh_token"] == "rotated_refresh_token"
await broker.close()
async def test_deprecated_refresh_omits_offline_access_for_cognito(
self, mock_storage
):
"""The deprecated _refresh_access_token also respects IdP discovery."""
broker = _make_broker(mock_storage)
discovery = {
"token_endpoint": "https://idp.example.com/token",
"scopes_supported": ["openid", "profile"],
}
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {
"access_token": "new_access_token",
"expires_in": 3600,
}
with patch.object(broker, "_get_oidc_config", return_value=discovery):
with patch.object(broker, "_get_http_client") as mock_client:
mock_post = AsyncMock(return_value=mock_response)
mock_client.return_value.post = mock_post
await broker._refresh_access_token("test_refresh")
call_kwargs = mock_post.call_args
posted_data = call_kwargs.kwargs.get("data") or call_kwargs[1].get(
"data"
)
scope_str = posted_data["scope"]
assert "offline_access" not in scope_str
await broker.close()
async def test_master_token_refresh_omits_offline_access_for_cognito(
self, mock_storage
):
"""refresh_master_token also respects IdP discovery."""
broker = _make_broker(mock_storage)
discovery = {
"token_endpoint": "https://idp.example.com/token",
"scopes_supported": ["openid"],
}
mock_storage.get_refresh_token.return_value = {
"refresh_token": "current_refresh",
}
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {
"access_token": "new_access",
"refresh_token": "rotated_refresh",
"expires_in": 3600,
}
with patch.object(broker, "_get_oidc_config", return_value=discovery):
with patch.object(broker, "_get_http_client") as mock_client:
mock_post = AsyncMock(return_value=mock_response)
mock_client.return_value.post = mock_post
await broker.refresh_master_token("user1")
call_kwargs = mock_post.call_args
posted_data = call_kwargs.kwargs.get("data") or call_kwargs[1].get(
"data"
)
scope_str = posted_data["scope"]
assert "offline_access" not in scope_str
await broker.close()