From b07b7131468b45ed93fbbc2b35e66bffa92b1851 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Sun, 5 Apr 2026 19:29:23 +0200 Subject: [PATCH] fix: address PR review feedback for client registry and DCR proxy - Document wildcard scope policy in ClientRegistry class docstring - Add hostname None guard and IPv6 loopback (::1) to redirect URI validation - Simplify redirect URI scheme validation into single guard clause - Add try/finally cleanup to DCR client deletion test - Validate 302 Location header in unknown client rejection test - Add unit tests for IPv6 loopback, malformed URIs, and DCR proxy paths Co-Authored-By: Claude Opus 4.6 (1M context) --- nextcloud_mcp_server/auth/client_registry.py | 28 ++++++-- .../server/keycloak/test_keycloak_clients.py | 32 +++++++-- tests/unit/test_client_registry.py | 33 ++++++++++ tests/unit/test_dcr_proxy.py | 65 +++++++++++++++++++ 4 files changed, 144 insertions(+), 14 deletions(-) create mode 100644 tests/unit/test_dcr_proxy.py diff --git a/nextcloud_mcp_server/auth/client_registry.py b/nextcloud_mcp_server/auth/client_registry.py index 91958e7b..af46828b 100644 --- a/nextcloud_mcp_server/auth/client_registry.py +++ b/nextcloud_mcp_server/auth/client_registry.py @@ -36,6 +36,12 @@ class ClientRegistry: 2. Integrate with IdP client registry 3. Store client metadata in database 4. Support client updates and revocation + + Scope Policy: + All clients are registered with allowed_scopes=["*"] (wildcard). + The MCP server acts as an OAuth AS proxy — it validates client + identity and redirect URIs locally, but delegates scope enforcement + to the upstream IdP (Nextcloud or Keycloak). """ def __init__(self, allow_dynamic_registration: bool = False): @@ -80,13 +86,19 @@ class ClientRegistry: continue parsed = urlparse(redirect) - is_loopback = parsed.hostname in ("localhost", "127.0.0.1") + hostname = parsed.hostname + if hostname is None: + logger.warning( + f"Skipping client {cid!r}: cannot parse hostname " + f"from {redirect!r}" + ) + continue + is_loopback = hostname in ("localhost", "127.0.0.1", "::1") - if parsed.scheme == "https": - pass # HTTPS always allowed - elif parsed.scheme == "http" and is_loopback: - pass # HTTP localhost allowed - else: + if not ( + parsed.scheme == "https" + or (parsed.scheme == "http" and is_loopback) + ): logger.warning( f"Rejecting client {cid!r}: HTTP redirect URIs are only " f"allowed for localhost, got {redirect!r}" @@ -205,6 +217,8 @@ class ClientRegistry: """ # Parse the redirect URI parsed = urlparse(redirect_uri) + if not parsed.hostname: + return False # Check against registered patterns for pattern in client.redirect_uris: @@ -213,7 +227,7 @@ class ClientRegistry: pattern_base = pattern.replace(":*", "") if redirect_uri.startswith(pattern_base + ":"): # Validate it's localhost with a port - if parsed.hostname in ["localhost", "127.0.0.1"]: + if parsed.hostname in ("localhost", "127.0.0.1", "::1"): return True elif redirect_uri == pattern: return True diff --git a/tests/server/keycloak/test_keycloak_clients.py b/tests/server/keycloak/test_keycloak_clients.py index c6e98287..d0467aaa 100644 --- a/tests/server/keycloak/test_keycloak_clients.py +++ b/tests/server/keycloak/test_keycloak_clients.py @@ -141,13 +141,25 @@ async def test_dcr_client_deletion_via_keycloak(keycloak_mcp_available): client_id = data["client_id"] rat = data["registration_access_token"] - # Delete via Keycloak (external URL) - del_resp = await http.delete( - f"{KEYCLOAK_BASE_URL}/realms/{KEYCLOAK_REALM}" - f"/clients-registrations/openid-connect/{client_id}", - headers={"Authorization": f"Bearer {rat}"}, - ) - assert del_resp.status_code == 204 + try: + # Delete via Keycloak (external URL) + del_resp = await http.delete( + f"{KEYCLOAK_BASE_URL}/realms/{KEYCLOAK_REALM}" + f"/clients-registrations/openid-connect/{client_id}", + headers={"Authorization": f"Bearer {rat}"}, + ) + assert del_resp.status_code == 204 + except Exception: + # Best-effort cleanup if assertion failed + try: + await http.delete( + f"{KEYCLOAK_BASE_URL}/realms/{KEYCLOAK_REALM}" + f"/clients-registrations/openid-connect/{client_id}", + headers={"Authorization": f"Bearer {rat}"}, + ) + except Exception: + pass + raise # --- AS metadata tests --- @@ -198,3 +210,9 @@ async def test_authorize_rejects_unknown_client(keycloak_mcp_available): if resp.status_code == 400: data = resp.json() assert "error" in data + else: + # 302 redirect must carry an error parameter + location = resp.headers.get("location", "") + assert "error=" in location, ( + f"302 redirect should contain error= in Location, got: {location}" + ) diff --git a/tests/unit/test_client_registry.py b/tests/unit/test_client_registry.py index f8836805..30c5709b 100644 --- a/tests/unit/test_client_registry.py +++ b/tests/unit/test_client_registry.py @@ -165,3 +165,36 @@ def test_client_name_resolution(monkeypatch): registry = _get_registry(monkeypatch, "claude-desktop, custom-tool") assert registry.get_client("claude-desktop").name == "Claude Desktop" assert registry.get_client("custom-tool").name == "Custom Tool" + + +def test_ipv6_loopback_allowed(monkeypatch): + registry = _get_registry(monkeypatch, "ipv6-app|http://[::1]:3000/cb") + client = registry.get_client("ipv6-app") + assert client is not None + assert client.redirect_uris == ["http://[::1]:3000/cb"] + + +def test_malformed_uri_no_hostname_skipped(monkeypatch, caplog): + with caplog.at_level(logging.WARNING): + registry = _get_registry(monkeypatch, "bad|http:///no-host") + + assert registry.get_client("bad") is None + assert "cannot parse hostname" in caplog.text + + +def test_validate_redirect_uri_ipv6_loopback(monkeypatch): + """IPv6 loopback redirect URIs should match wildcard localhost patterns.""" + registry = _get_registry(monkeypatch, "ipv6-app|http://[::1]:3000/cb") + valid, err = registry.validate_client( + "ipv6-app", redirect_uri="http://[::1]:3000/cb" + ) + assert valid is True + assert err is None + + +def test_validate_redirect_uri_no_hostname(monkeypatch): + """Redirect URIs with no parseable hostname should be rejected.""" + registry = _get_registry(monkeypatch, "test-client") + valid, err = registry.validate_client("test-client", redirect_uri="not-a-uri") + assert valid is False + assert "redirect_uri" in err.lower() diff --git a/tests/unit/test_dcr_proxy.py b/tests/unit/test_dcr_proxy.py new file mode 100644 index 00000000..07eed461 --- /dev/null +++ b/tests/unit/test_dcr_proxy.py @@ -0,0 +1,65 @@ +"""Unit tests for DCR proxy registration_not_supported path.""" + +import json +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from nextcloud_mcp_server.auth.oauth_routes import oauth_register_proxy + +pytestmark = pytest.mark.unit + + +def _make_request(body: dict, oauth_config: dict) -> MagicMock: + """Create a mock Starlette Request.""" + request = AsyncMock() + request.json = AsyncMock(return_value=body) + request.client = MagicMock() + request.client.host = "127.0.0.1" + request.app = MagicMock() + request.app.state.oauth_context = {"config": oauth_config} + return request + + +_DCR_BODY = { + "client_name": "test", + "redirect_uris": ["http://localhost:9999/cb"], +} + + +async def test_registration_not_supported_when_no_endpoint(): + """When discovery doc lacks registration_endpoint, return 400.""" + request = _make_request( + body=_DCR_BODY, + oauth_config={ + "discovery_url": "https://idp.example.com/.well-known/openid-configuration" + }, + ) + + discovery_doc = { + "issuer": "https://idp.example.com", + "authorization_endpoint": "https://idp.example.com/auth", + } + + with patch( + "nextcloud_mcp_server.auth.oauth_routes._get_cached_discovery", + new_callable=AsyncMock, + return_value=discovery_doc, + ): + response = await oauth_register_proxy(request) + + assert response.status_code == 400 + body = json.loads(response.body) + assert body["error"] == "registration_not_supported" + assert "ALLOWED_MCP_CLIENTS" in body["error_description"] + + +async def test_registration_not_supported_when_no_discovery_url(): + """When no discovery_url is configured, return 400.""" + request = _make_request(body=_DCR_BODY, oauth_config={}) + + response = await oauth_register_proxy(request) + + assert response.status_code == 400 + body = json.loads(response.body) + assert body["error"] == "registration_not_supported"