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) <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2026-04-05 19:29:23 +02:00
co-authored by Claude Opus 4.6
parent 7d775d2a52
commit b07b713146
4 changed files with 144 additions and 14 deletions
+21 -7
View File
@@ -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
+25 -7
View File
@@ -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}"
)
+33
View File
@@ -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()
+65
View File
@@ -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"