fix: address PR review for OIDC scope prefix feature
Add offline_access to OIDC standard scopes exclusion list to prevent it from being incorrectly prefixed, which would break Cognito refresh token flows. Extract scope transformation into testable _transform_scopes_for_idp() helper, add debug logging for prefixed scopes, remove unused Settings field (oauth_routes.py consistently uses os.getenv), and add unit tests. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.6
parent
e21ddd91b9
commit
f67d4d1116
@@ -109,6 +109,28 @@ _DCR_RATE_LIMIT_MAX = 10 # max requests
|
|||||||
_DCR_RATE_LIMIT_WINDOW = 60 # per 60 seconds
|
_DCR_RATE_LIMIT_WINDOW = 60 # per 60 seconds
|
||||||
|
|
||||||
|
|
||||||
|
# OIDC standard scopes that must never be prefixed with a resource server identifier.
|
||||||
|
_OIDC_STANDARD_SCOPES = {"openid", "profile", "email", "offline_access"}
|
||||||
|
|
||||||
|
|
||||||
|
def _transform_scopes_for_idp(scopes: str, resource_server_id: str) -> str:
|
||||||
|
"""Prefix resource scopes with an IdP resource server identifier.
|
||||||
|
|
||||||
|
IdPs like AWS Cognito require resource scopes in ``{identifier}/{scope}``
|
||||||
|
format. Standard OIDC scopes (openid, profile, email, offline_access) are
|
||||||
|
forwarded unchanged.
|
||||||
|
|
||||||
|
When *resource_server_id* is empty the original scope string is returned
|
||||||
|
as-is.
|
||||||
|
"""
|
||||||
|
if not resource_server_id:
|
||||||
|
return scopes
|
||||||
|
return " ".join(
|
||||||
|
f"{resource_server_id}/{s}" if s not in _OIDC_STANDARD_SCOPES else s
|
||||||
|
for s in scopes.split()
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
async def _get_cached_discovery(url: str) -> dict[str, Any]:
|
async def _get_cached_discovery(url: str) -> dict[str, Any]:
|
||||||
"""Fetch OIDC discovery document with caching (5-minute TTL)."""
|
"""Fetch OIDC discovery document with caching (5-minute TTL)."""
|
||||||
now = time.time()
|
now = time.time()
|
||||||
@@ -347,17 +369,10 @@ async def oauth_authorize(request: Request) -> RedirectResponse | JSONResponse:
|
|||||||
|
|
||||||
# Prefix resource scopes with the resource server identifier if configured.
|
# Prefix resource scopes with the resource server identifier if configured.
|
||||||
# Required for IdPs like Cognito that use {identifier}/{scope} format.
|
# Required for IdPs like Cognito that use {identifier}/{scope} format.
|
||||||
# OIDC standard scopes are forwarded as-is.
|
resource_server_id = os.getenv("OIDC_RESOURCE_SERVER_ID", "").strip()
|
||||||
oidc_scopes = {"openid", "profile", "email"}
|
idp_scope_str = _transform_scopes_for_idp(scopes, resource_server_id)
|
||||||
resource_server_id = os.getenv("OIDC_RESOURCE_SERVER_ID", "")
|
|
||||||
if resource_server_id:
|
if resource_server_id:
|
||||||
idp_scope_list = [
|
logger.info(f" IdP scopes (prefixed): {idp_scope_str}")
|
||||||
f"{resource_server_id}/{s}" if s not in oidc_scopes else s
|
|
||||||
for s in scopes.split()
|
|
||||||
]
|
|
||||||
idp_scope_str = " ".join(idp_scope_list)
|
|
||||||
else:
|
|
||||||
idp_scope_str = scopes
|
|
||||||
|
|
||||||
# Redirect to Nextcloud with MCP server's own client_id (no PKCE — confidential client)
|
# Redirect to Nextcloud with MCP server's own client_id (no PKCE — confidential client)
|
||||||
idp_params = {
|
idp_params = {
|
||||||
|
|||||||
@@ -213,7 +213,6 @@ class Settings:
|
|||||||
oidc_client_id: str | None = None
|
oidc_client_id: str | None = None
|
||||||
oidc_client_secret: str | None = None
|
oidc_client_secret: str | None = None
|
||||||
oidc_issuer: str | None = None
|
oidc_issuer: str | None = None
|
||||||
oidc_resource_server_id: str | None = None
|
|
||||||
|
|
||||||
# Nextcloud settings
|
# Nextcloud settings
|
||||||
nextcloud_host: str | None = None
|
nextcloud_host: str | None = None
|
||||||
@@ -569,7 +568,6 @@ def get_settings() -> Settings:
|
|||||||
"oidc_client_id": "NEXTCLOUD_OIDC_CLIENT_ID",
|
"oidc_client_id": "NEXTCLOUD_OIDC_CLIENT_ID",
|
||||||
"oidc_client_secret": "NEXTCLOUD_OIDC_CLIENT_SECRET",
|
"oidc_client_secret": "NEXTCLOUD_OIDC_CLIENT_SECRET",
|
||||||
"oidc_issuer": "OIDC_ISSUER",
|
"oidc_issuer": "OIDC_ISSUER",
|
||||||
"oidc_resource_server_id": "OIDC_RESOURCE_SERVER_ID",
|
|
||||||
# Nextcloud settings
|
# Nextcloud settings
|
||||||
"nextcloud_host": "NEXTCLOUD_HOST",
|
"nextcloud_host": "NEXTCLOUD_HOST",
|
||||||
"nextcloud_username": "NEXTCLOUD_USERNAME",
|
"nextcloud_username": "NEXTCLOUD_USERNAME",
|
||||||
|
|||||||
@@ -37,7 +37,6 @@ oidc_issuer = "@none"
|
|||||||
jwks_uri = "@none"
|
jwks_uri = "@none"
|
||||||
introspection_uri = "@none"
|
introspection_uri = "@none"
|
||||||
userinfo_uri = "@none"
|
userinfo_uri = "@none"
|
||||||
oidc_resource_server_id = "@none"
|
|
||||||
|
|
||||||
# --- Mode flags ---
|
# --- Mode flags ---
|
||||||
enable_multi_user_basic_auth = false
|
enable_multi_user_basic_auth = false
|
||||||
|
|||||||
@@ -0,0 +1,72 @@
|
|||||||
|
"""Tests for OIDC resource server scope prefixing."""
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from nextcloud_mcp_server.auth.oauth_routes import _transform_scopes_for_idp
|
||||||
|
|
||||||
|
|
||||||
|
class TestTransformScopesForIdp:
|
||||||
|
"""Test _transform_scopes_for_idp scope transformation."""
|
||||||
|
|
||||||
|
def test_no_prefix_when_resource_server_id_empty(self):
|
||||||
|
"""Scopes are returned unchanged when resource_server_id is empty."""
|
||||||
|
scopes = "openid profile notes.read notes.write"
|
||||||
|
assert _transform_scopes_for_idp(scopes, "") == scopes
|
||||||
|
|
||||||
|
def test_oidc_scopes_not_prefixed(self):
|
||||||
|
"""Standard OIDC scopes are never prefixed."""
|
||||||
|
result = _transform_scopes_for_idp(
|
||||||
|
"openid profile email", "https://api.example.com"
|
||||||
|
)
|
||||||
|
assert result == "openid profile email"
|
||||||
|
|
||||||
|
def test_offline_access_not_prefixed(self):
|
||||||
|
"""offline_access is a standard OIDC scope and must not be prefixed."""
|
||||||
|
result = _transform_scopes_for_idp(
|
||||||
|
"openid offline_access notes.read", "https://api.example.com"
|
||||||
|
)
|
||||||
|
assert result == "openid offline_access https://api.example.com/notes.read"
|
||||||
|
|
||||||
|
def test_resource_scopes_prefixed(self):
|
||||||
|
"""Non-OIDC scopes are prefixed with the resource server identifier."""
|
||||||
|
result = _transform_scopes_for_idp(
|
||||||
|
"notes.read notes.write", "https://api.example.com"
|
||||||
|
)
|
||||||
|
assert (
|
||||||
|
result
|
||||||
|
== "https://api.example.com/notes.read https://api.example.com/notes.write"
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_mixed_scopes(self):
|
||||||
|
"""Mixed OIDC and resource scopes are handled correctly."""
|
||||||
|
result = _transform_scopes_for_idp(
|
||||||
|
"openid profile notes.read calendar.write offline_access",
|
||||||
|
"https://api.example.com",
|
||||||
|
)
|
||||||
|
assert result == (
|
||||||
|
"openid profile https://api.example.com/notes.read "
|
||||||
|
"https://api.example.com/calendar.write offline_access"
|
||||||
|
)
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
("resource_server_id", "expected_prefix"),
|
||||||
|
[
|
||||||
|
("https://api.example.com", "https://api.example.com/notes.read"),
|
||||||
|
("my-api", "my-api/notes.read"),
|
||||||
|
("urn:api:prod", "urn:api:prod/notes.read"),
|
||||||
|
],
|
||||||
|
)
|
||||||
|
def test_various_identifier_formats(self, resource_server_id, expected_prefix):
|
||||||
|
"""Different resource server identifier formats are supported."""
|
||||||
|
result = _transform_scopes_for_idp("notes.read", resource_server_id)
|
||||||
|
assert result == expected_prefix
|
||||||
|
|
||||||
|
def test_single_resource_scope(self):
|
||||||
|
"""A single non-OIDC scope is prefixed."""
|
||||||
|
result = _transform_scopes_for_idp("notes.read", "https://api.example.com")
|
||||||
|
assert result == "https://api.example.com/notes.read"
|
||||||
|
|
||||||
|
def test_empty_scopes_string(self):
|
||||||
|
"""An empty scopes string returns empty."""
|
||||||
|
result = _transform_scopes_for_idp("", "https://api.example.com")
|
||||||
|
assert result == ""
|
||||||
Reference in New Issue
Block a user