diff --git a/api/app/urls.py b/api/app/urls.py index 670aa258aab8..69661e620a3c 100644 --- a/api/app/urls.py +++ b/api/app/urls.py @@ -8,6 +8,7 @@ from oauth2_provider import views as oauth2_views from oauth2_metadata.views import ( + CIMDTokenView, DynamicClientRegistrationView, OAuthAuthorizeView, authorization_server_metadata, @@ -73,7 +74,7 @@ include( ( [ - path("token/", oauth2_views.TokenView.as_view(), name="token"), + path("token/", CIMDTokenView.as_view(), name="token"), path( "revoke_token/", oauth2_views.RevokeTokenView.as_view(), diff --git a/api/oauth2_metadata/cimd.py b/api/oauth2_metadata/cimd.py new file mode 100644 index 000000000000..71d6505de21e --- /dev/null +++ b/api/oauth2_metadata/cimd.py @@ -0,0 +1,204 @@ +"""Client ID Metadata Document (CIMD) resolution. + +When a client_id is an HTTPS URL, the authorisation server fetches the +client metadata document from that URL instead of requiring DCR. +""" + +import ipaddress +import socket +from urllib.parse import urlparse + +import requests +import structlog +from django.core.cache import cache +from django.core.exceptions import ValidationError +from oauth2_provider.models import Application + +from oauth2_metadata.metrics import flagsmith_oauth2_cimd_resolutions_total +from oauth2_metadata.services import validate_redirect_uri + +logger = structlog.get_logger("oauth2_metadata") + +# Cache resolved CIMD applications for 10 minutes to avoid hammering the +# client's metadata endpoint on every authorize/token call. +CIMD_CACHE_TTL_SECONDS = 60 * 10 +CIMD_FETCH_TIMEOUT_SECONDS = 5 + +# Auth methods that require a shared secret — impossible without a +# registration step, so we reject these. +_SECRET_BASED_AUTH_METHODS = frozenset({"client_secret_basic", "client_secret_post"}) +# Auth methods not yet implemented. +_UNSUPPORTED_AUTH_METHODS = frozenset({"private_key_jwt"}) + + +# DOT's Application.client_id field has max_length=100. +# TODO: real-world CIMD URLs (e.g. Claude Code) can be long; consider a +# migration to increase Application.client_id max_length if this proves +# too restrictive. +_CLIENT_ID_MAX_LENGTH = 100 + + +def is_cimd_client_id(client_id: str) -> bool: + """Return True if client_id looks like an HTTPS URL.""" + return client_id.startswith("https://") + + +def _is_public_hostname(hostname: str) -> bool: + """Return True if hostname resolves to at least one public IP address.""" + try: + addrinfo = socket.getaddrinfo(hostname, None) + except socket.gaierror: + return False + + for family, _type, _proto, _canonname, sockaddr in addrinfo: + ip = ipaddress.ip_address(sockaddr[0]) + if ip.is_private or ip.is_loopback or ip.is_reserved or ip.is_link_local: + return False + return bool(addrinfo) + + +def _fetch_cimd_document(client_id_url: str) -> dict: + """Fetch and return the JSON metadata document at client_id_url. + + Raises ValueError on any fetch/parse failure. + """ + parsed = urlparse(client_id_url) + if parsed.scheme != "https": + raise ValueError(f"client_id must be an HTTPS URL: {client_id_url}") + + if len(client_id_url) > _CLIENT_ID_MAX_LENGTH: + raise ValueError( + f"client_id URL exceeds {_CLIENT_ID_MAX_LENGTH} characters: {client_id_url}" + ) + + hostname = parsed.hostname + if not hostname: + raise ValueError(f"client_id URL has no hostname: {client_id_url}") + + if not _is_public_hostname(hostname): + raise ValueError( + f"client_id hostname does not resolve to a public address: {hostname}" + ) + + # TODO: _is_public_hostname resolves DNS independently from requests.get, + # leaving a small TOCTOU window for DNS rebinding attacks. A robust fix + # would pin the resolved IP and connect to it directly. + try: + response = requests.get( + client_id_url, + timeout=CIMD_FETCH_TIMEOUT_SECONDS, + allow_redirects=False, + headers={"Accept": "application/json"}, + ) + response.raise_for_status() + except requests.RequestException as exc: + raise ValueError(f"Failed to fetch CIMD document: {exc}") from exc + + try: + return response.json() + except ValueError as exc: + raise ValueError(f"CIMD document is not valid JSON: {exc}") from exc + + +def _validate_cimd_document(client_id_url: str, doc: dict) -> dict: + """Validate a CIMD document and return normalised metadata. + + Raises ValueError on validation failure. + """ + # The document's client_id MUST match the URL it was fetched from. + doc_client_id = doc.get("client_id") + if doc_client_id != client_id_url: + raise ValueError( + f"CIMD client_id mismatch: document says {doc_client_id!r}, " + f"expected {client_id_url!r}" + ) + + # redirect_uris is required. + redirect_uris = doc.get("redirect_uris") + if not redirect_uris or not isinstance(redirect_uris, list): + raise ValueError("CIMD document must contain a non-empty redirect_uris array") + + # Validate each redirect URI against the same policy as DCR. + for uri in redirect_uris: + try: + validate_redirect_uri(uri) + except ValidationError as exc: + raise ValueError( + f"Invalid redirect_uri in CIMD document: {exc.message}" + ) from exc + + # token_endpoint_auth_method: default to "none" if absent. + auth_method = doc.get("token_endpoint_auth_method", "none") + + if auth_method in _SECRET_BASED_AUTH_METHODS: + raise ValueError( + f"CIMD clients cannot use secret-based auth method: {auth_method}. " + f"No registration step exists to distribute a secret." + ) + + if auth_method in _UNSUPPORTED_AUTH_METHODS: + raise ValueError( + f"Auth method {auth_method} is not yet implemented for CIMD clients." + ) + + # client_name falls back to the hostname. + parsed = urlparse(client_id_url) + client_name = doc.get("client_name") or parsed.hostname or "CIMD client" + + return { + "client_name": client_name, + "redirect_uris": redirect_uris, + "token_endpoint_auth_method": auth_method, + } + + +def resolve_cimd_client(client_id_url: str) -> Application | None: + """Resolve a CIMD client_id URL to a DOT Application. + + Returns the Application on success, None on failure (all failures + are logged and counted). + """ + cache_key = f"cimd:{client_id_url}" + cached_app_pk = cache.get(cache_key) + if cached_app_pk is not None: + try: + return Application.objects.get(pk=cached_app_pk) + except Application.DoesNotExist: + cache.delete(cache_key) + + try: + doc = _fetch_cimd_document(client_id_url) + metadata = _validate_cimd_document(client_id_url, doc) + except ValueError as exc: + logger.error( + "cimd.rejected", + client_id=client_id_url, + reason=str(exc), + ) + flagsmith_oauth2_cimd_resolutions_total.labels(outcome="rejected").inc() + return None + + # Upsert: reuse an existing Application row keyed by the URL client_id, + # or create one. This avoids littering Application rows. + application, created = Application.objects.update_or_create( + client_id=client_id_url, + defaults={ + "name": metadata["client_name"], + "client_type": Application.CLIENT_PUBLIC, + "authorization_grant_type": Application.GRANT_AUTHORIZATION_CODE, + "client_secret": "", + "redirect_uris": " ".join(metadata["redirect_uris"]), + "skip_authorization": False, + }, + ) + + action = "created" if created else "refreshed" + logger.info( + f"cimd.{action}", + client_id=client_id_url, + client_name=metadata["client_name"], + ) + flagsmith_oauth2_cimd_resolutions_total.labels(outcome="resolved").inc() + + cache.set(cache_key, application.pk, CIMD_CACHE_TTL_SECONDS) + return application diff --git a/api/oauth2_metadata/metrics.py b/api/oauth2_metadata/metrics.py index ec3b7ea64a69..8ae1044aeab1 100644 --- a/api/oauth2_metadata/metrics.py +++ b/api/oauth2_metadata/metrics.py @@ -7,3 +7,10 @@ "was accepted or rejected.", ["token_endpoint_auth_method", "outcome"], ) + +flagsmith_oauth2_cimd_resolutions_total = prometheus_client.Counter( + "flagsmith_oauth2_cimd_resolutions_total", + "Total OAuth2 CIMD (Client ID Metadata Document) resolution attempts, " + "labelled by whether the resolution was accepted or rejected.", + ["outcome"], +) diff --git a/api/oauth2_metadata/views.py b/api/oauth2_metadata/views.py index a9fb137f368c..dd1c804e49d1 100644 --- a/api/oauth2_metadata/views.py +++ b/api/oauth2_metadata/views.py @@ -2,21 +2,22 @@ from urllib.parse import urlencode, urlparse, urlunparse import structlog -from django.http import HttpRequest, JsonResponse, QueryDict +from django.http import HttpRequest, HttpResponse, JsonResponse, QueryDict from django.views.decorators.csrf import csrf_exempt from django.views.decorators.http import require_GET from oauth2_provider.exceptions import OAuthToolkitError from oauth2_provider.models import get_application_model from oauth2_provider.scopes import get_scopes_backend +from oauth2_provider.views import TokenView from oauth2_provider.views.mixins import OAuthLibMixin from rest_framework import status -from rest_framework import status as drf_status from rest_framework.permissions import AllowAny, IsAuthenticated from rest_framework.request import Request from rest_framework.response import Response from rest_framework.throttling import ScopedRateThrottle from rest_framework.views import APIView +from oauth2_metadata.cimd import is_cimd_client_id, resolve_cimd_client from oauth2_metadata.dataclasses import OAuthConfig from oauth2_metadata.mappers import map_drf_error_to_rfc7591_error_body from oauth2_metadata.metrics import flagsmith_oauth2_dcr_registrations_total @@ -53,6 +54,7 @@ def authorization_server_metadata(request: HttpRequest) -> JsonResponse: "none", ], "introspection_endpoint_auth_methods_supported": ["none"], + "client_id_metadata_document_supported": True, } return JsonResponse(metadata) @@ -63,11 +65,36 @@ class OAuthAuthorizeView(OAuthLibMixin, APIView): # type: ignore[misc] permission_classes = [IsAuthenticated] + def _ensure_cimd_client(self, request: HttpRequest) -> str | None: + """If client_id is a CIMD URL, resolve it and return the client_id. + + Returns None if resolution fails, so the caller can return an error. + The client_id in the request is NOT mutated — DOT will look it up + by the URL value which is now stored as Application.client_id. + """ + client_id = request.GET.get("client_id", "") + if not is_cimd_client_id(client_id): + return client_id # Not a CIMD client_id, let DOT handle it. + app = resolve_cimd_client(client_id) + if app is None: + return None + return client_id + def get(self, request: Request, *args: Any, **kwargs: Any) -> Response: """Validate an authorisation request and return application info.""" # Bridge DRF auth to Django request so DOT sees the authenticated user. request._request.user = request.user + resolved = self._ensure_cimd_client(request._request) + if resolved is None: + return Response( + { + "error": "invalid_client", + "error_description": "Could not resolve CIMD client metadata.", + }, + status=status.HTTP_400_BAD_REQUEST, + ) + try: scopes, credentials = self.validate_authorization_request(request._request) except OAuthToolkitError as e: @@ -122,6 +149,16 @@ def post(self, request: Request, *args: Any, **kwargs: Any) -> Response: request._request.GET = query # type: ignore[assignment] request._request.META["QUERY_STRING"] = query.urlencode() + resolved = self._ensure_cimd_client(request._request) + if resolved is None: + return Response( + { + "error": "invalid_client", + "error_description": "Could not resolve CIMD client metadata.", + }, + status=status.HTTP_400_BAD_REQUEST, + ) + try: scopes, credentials = self.validate_authorization_request(request._request) except OAuthToolkitError as e: @@ -178,7 +215,7 @@ def post(self, request: Request) -> Response: ) return Response( error_body, - status=drf_status.HTTP_400_BAD_REQUEST, + status=status.HTTP_400_BAD_REQUEST, ) data = serializer.validated_data @@ -207,7 +244,7 @@ def post(self, request: Request) -> Response: # 0 means the secret never expires, per RFC 7591 §3.2.1. response_body["client_secret_expires_at"] = 0 - return Response(response_body, status=drf_status.HTTP_201_CREATED) + return Response(response_body, status=status.HTTP_201_CREATED) def _count_registration(self, auth_method: Any, outcome: str) -> None: # Requested method is client input; collapse unknown values to keep @@ -219,3 +256,23 @@ def _count_registration(self, auth_method: Any, outcome: str) -> None: flagsmith_oauth2_dcr_registrations_total.labels( token_endpoint_auth_method=auth_method, outcome=outcome ).inc() + + +class CIMDTokenView(TokenView): + """Token endpoint that resolves CIMD client_ids before DOT processing. + + Wraps DOT's TokenView so that when a client_id in the POST body is + an HTTPS URL, we ensure the corresponding Application row exists + before DOT attempts to look it up. + """ + + def post(self, request: HttpRequest, *args: Any, **kwargs: Any) -> HttpResponse: + client_id = request.POST.get("client_id", "") + if is_cimd_client_id(client_id): + app = resolve_cimd_client(client_id) + if app is None: + return JsonResponse( + {"error": "invalid_client"}, + status=400, + ) + return super().post(request, *args, **kwargs) diff --git a/api/tests/unit/oauth2_metadata/test_authorize_view.py b/api/tests/unit/oauth2_metadata/test_authorize_view.py index e5fa19e887a0..e38239c1ea58 100644 --- a/api/tests/unit/oauth2_metadata/test_authorize_view.py +++ b/api/tests/unit/oauth2_metadata/test_authorize_view.py @@ -1,6 +1,7 @@ import base64 import hashlib import secrets +from unittest.mock import patch from urllib.parse import parse_qs, urlparse import pytest @@ -332,3 +333,164 @@ def test_get__flagsmith_cli_requests_admin_api__returns_application_info( assert data["application"]["client_id"] == FLAGSMITH_CLI_CLIENT_ID assert "admin-api" in data["scopes"] assert data["is_verified"] is True + + +# --------------------------------------------------------------------------- +# CIMD integration — OAuthAuthorizeView with HTTPS URL client_ids +# --------------------------------------------------------------------------- + +CIMD_CLIENT_ID_URL = "https://cimd.example.com/oauth/metadata" + + +def _mock_cimd_doc( + client_id: str = CIMD_CLIENT_ID_URL, + client_name: str = "CIMD Test App", + redirect_uris: list[str] | None = None, +) -> dict: + return { + "client_id": client_id, + "client_name": client_name, + "redirect_uris": redirect_uris or ["https://example.com/callback"], + } + + +def test_get__cimd_client_id__resolves_and_returns_application_info( + auth_client: APIClient, + pkce_pair: tuple[str, str], + db: None, +) -> None: + # Given + _verifier, challenge = pkce_pair + doc = _mock_cimd_doc() + + # When + with ( + patch("oauth2_metadata.cimd._fetch_cimd_document", return_value=doc), + patch("oauth2_metadata.cimd._is_public_hostname", return_value=True), + patch("oauth2_metadata.cimd.cache"), + ): + response = auth_client.get( + "/api/v1/oauth/authorize/", + { + "client_id": CIMD_CLIENT_ID_URL, + "response_type": "code", + "redirect_uri": "https://example.com/callback", + "scope": "mcp", + "code_challenge": challenge, + "code_challenge_method": "S256", + }, + ) + + # Then + assert response.status_code == status.HTTP_200_OK + data = response.json() + assert data["application"]["name"] == "CIMD Test App" + assert data["application"]["client_id"] == CIMD_CLIENT_ID_URL + assert data["is_verified"] is False + + +def test_get__cimd_client_id_resolution_fails__returns_400( + auth_client: APIClient, + pkce_pair: tuple[str, str], + db: None, +) -> None: + # Given + _verifier, challenge = pkce_pair + + # When + with ( + patch( + "oauth2_metadata.cimd._fetch_cimd_document", + side_effect=ValueError("unreachable"), + ), + patch("oauth2_metadata.cimd.cache"), + ): + response = auth_client.get( + "/api/v1/oauth/authorize/", + { + "client_id": CIMD_CLIENT_ID_URL, + "response_type": "code", + "redirect_uri": "https://example.com/callback", + "scope": "mcp", + "code_challenge": challenge, + "code_challenge_method": "S256", + }, + ) + + # Then + assert response.status_code == status.HTTP_400_BAD_REQUEST + data = response.json() + assert data["error"] == "invalid_client" + + +def test_post__cimd_consent_allow__returns_redirect( + auth_client: APIClient, + db: None, +) -> None: + # Given + _verifier, challenge = _pkce_pair() + doc = _mock_cimd_doc() + + # When + with ( + patch("oauth2_metadata.cimd._fetch_cimd_document", return_value=doc), + patch("oauth2_metadata.cimd._is_public_hostname", return_value=True), + patch("oauth2_metadata.cimd.cache"), + ): + response = auth_client.post( + "/api/v1/oauth/authorize/", + { + "allow": True, + "client_id": CIMD_CLIENT_ID_URL, + "response_type": "code", + "redirect_uri": "https://example.com/callback", + "scope": "mcp", + "code_challenge": challenge, + "code_challenge_method": "S256", + "state": "cimd-test-state", + }, + format="json", + ) + + # Then + assert response.status_code == status.HTTP_200_OK + redirect_uri = response.json()["redirect_uri"] + parsed = urlparse(redirect_uri) + query_params = parse_qs(parsed.query) + assert "code" in query_params + assert query_params["state"] == ["cimd-test-state"] + + +def test_post__cimd_client_id_resolution_fails__returns_400( + auth_client: APIClient, + db: None, +) -> None: + # Given + _verifier, challenge = _pkce_pair() + + # When + with ( + patch( + "oauth2_metadata.cimd._fetch_cimd_document", + side_effect=ValueError("unreachable"), + ), + patch("oauth2_metadata.cimd.cache"), + ): + response = auth_client.post( + "/api/v1/oauth/authorize/", + { + "allow": True, + "client_id": CIMD_CLIENT_ID_URL, + "response_type": "code", + "redirect_uri": "https://example.com/callback", + "scope": "mcp", + "code_challenge": challenge, + "code_challenge_method": "S256", + }, + format="json", + ) + + # Then + assert response.status_code == status.HTTP_400_BAD_REQUEST + data = response.json() + assert data["error"] == "invalid_client" diff --git a/api/tests/unit/oauth2_metadata/test_cimd.py b/api/tests/unit/oauth2_metadata/test_cimd.py new file mode 100644 index 000000000000..bbac25a78d63 --- /dev/null +++ b/api/tests/unit/oauth2_metadata/test_cimd.py @@ -0,0 +1,566 @@ +"""Tests for Client ID Metadata Document (CIMD) resolution.""" + +from unittest.mock import MagicMock, patch + +import pytest +import requests +from common.test_tools import AssertMetricFixture +from django.test import Client +from django.urls import reverse +from oauth2_provider.models import Application +from pytest_structlog import StructuredLogCapture +from rest_framework import status + +METADATA_URL = "oauth-authorization-server-metadata" + + +@pytest.fixture() +def client() -> Client: + return Client() + + +# --------------------------------------------------------------------------- +# Metadata endpoint +# --------------------------------------------------------------------------- + + +def test_metadata_endpoint__get__advertises_cimd_support( + client: Client, + settings: "SettingsWrapper", # noqa: F821 +) -> None: + # Given + settings.FLAGSMITH_API_URL = "https://api.flagsmith.com" + settings.FLAGSMITH_FRONTEND_URL = "https://app.flagsmith.com" + + # When + response = client.get(reverse(METADATA_URL)) + + # Then + assert response.status_code == status.HTTP_200_OK + data = response.json() + assert data["client_id_metadata_document_supported"] is True + + +# --------------------------------------------------------------------------- +# cimd module: is_cimd_client_id +# --------------------------------------------------------------------------- + + +def test_is_cimd_client_id__https_url__returns_true() -> None: + from oauth2_metadata.cimd import is_cimd_client_id + + # Given + client_id = "https://claude.ai/oauth/metadata" + + # When / Then + assert is_cimd_client_id(client_id) is True + + +def test_is_cimd_client_id__opaque_id__returns_false() -> None: + from oauth2_metadata.cimd import is_cimd_client_id + + # Given + client_id = "abc123-client-id" + + # When / Then + assert is_cimd_client_id(client_id) is False + + +def test_is_cimd_client_id__http_url__returns_false() -> None: + from oauth2_metadata.cimd import is_cimd_client_id + + # Given + client_id = "http://example.com/metadata" + + # When / Then + assert is_cimd_client_id(client_id) is False + + +# --------------------------------------------------------------------------- +# cimd module: _is_public_hostname +# --------------------------------------------------------------------------- + + +def test_is_public_hostname__localhost__returns_false() -> None: + from oauth2_metadata.cimd import _is_public_hostname + + # Given + hostname = "localhost" + + # When / Then + assert _is_public_hostname(hostname) is False + + +def test_is_public_hostname__nonexistent_host__returns_false() -> None: + from oauth2_metadata.cimd import _is_public_hostname + + # Given + hostname = "this-host-does-not-exist.invalid" + + # When / Then + assert _is_public_hostname(hostname) is False + + +def test_is_public_hostname__private_ip__returns_false() -> None: + from oauth2_metadata.cimd import _is_public_hostname + + # Given + # Mock getaddrinfo to return a private IP. + with patch("oauth2_metadata.cimd.socket.getaddrinfo") as mock_gai: + mock_gai.return_value = [ + (2, 1, 6, "", ("192.168.1.1", 0)), + ] + + # When / Then + assert _is_public_hostname("internal.example.com") is False + + +def test_is_public_hostname__public_ip__returns_true() -> None: + from oauth2_metadata.cimd import _is_public_hostname + + # Given + with patch("oauth2_metadata.cimd.socket.getaddrinfo") as mock_gai: + mock_gai.return_value = [ + (2, 1, 6, "", ("93.184.216.34", 0)), + ] + + # When / Then + assert _is_public_hostname("example.com") is True + + +# --------------------------------------------------------------------------- +# cimd module: _validate_cimd_document +# --------------------------------------------------------------------------- + +CLIENT_ID_URL = "https://example.com/oauth/metadata" + + +def test_validate_cimd_document__valid_document__returns_metadata() -> None: + from oauth2_metadata.cimd import _validate_cimd_document + + # Given + doc = { + "client_id": CLIENT_ID_URL, + "client_name": "Test App", + "redirect_uris": ["https://example.com/callback"], + } + + # When + result = _validate_cimd_document(CLIENT_ID_URL, doc) + + # Then + assert result["client_name"] == "Test App" + assert result["redirect_uris"] == ["https://example.com/callback"] + assert result["token_endpoint_auth_method"] == "none" + + +def test_validate_cimd_document__missing_token_endpoint_auth_method__defaults_to_none() -> ( + None +): + from oauth2_metadata.cimd import _validate_cimd_document + + # Given + doc = { + "client_id": CLIENT_ID_URL, + "redirect_uris": ["https://example.com/callback"], + } + + # When + result = _validate_cimd_document(CLIENT_ID_URL, doc) + + # Then + assert result["token_endpoint_auth_method"] == "none" + + +def test_validate_cimd_document__client_id_mismatch__raises_value_error() -> None: + from oauth2_metadata.cimd import _validate_cimd_document + + # Given + doc = { + "client_id": "https://other.com/metadata", + "redirect_uris": ["https://example.com/callback"], + } + + # When / Then + with pytest.raises(ValueError, match="mismatch"): + _validate_cimd_document(CLIENT_ID_URL, doc) + + +def test_validate_cimd_document__missing_redirect_uris__raises_value_error() -> None: + from oauth2_metadata.cimd import _validate_cimd_document + + # Given + doc = {"client_id": CLIENT_ID_URL} + + # When / Then + with pytest.raises(ValueError, match="redirect_uris"): + _validate_cimd_document(CLIENT_ID_URL, doc) + + +def test_validate_cimd_document__empty_redirect_uris__raises_value_error() -> None: + from oauth2_metadata.cimd import _validate_cimd_document + + # Given + doc = {"client_id": CLIENT_ID_URL, "redirect_uris": []} + + # When / Then + with pytest.raises(ValueError, match="redirect_uris"): + _validate_cimd_document(CLIENT_ID_URL, doc) + + +@pytest.mark.parametrize( + "auth_method", + ["client_secret_basic", "client_secret_post"], + ids=["secret-basic", "secret-post"], +) +def test_validate_cimd_document__secret_based_auth_method__raises_value_error( + auth_method: str, +) -> None: + from oauth2_metadata.cimd import _validate_cimd_document + + # Given + doc = { + "client_id": CLIENT_ID_URL, + "redirect_uris": ["https://example.com/callback"], + "token_endpoint_auth_method": auth_method, + } + + # When / Then + with pytest.raises(ValueError, match="secret-based"): + _validate_cimd_document(CLIENT_ID_URL, doc) + + +def test_validate_cimd_document__private_key_jwt__raises_value_error() -> None: + from oauth2_metadata.cimd import _validate_cimd_document + + # Given + doc = { + "client_id": CLIENT_ID_URL, + "redirect_uris": ["https://example.com/callback"], + "token_endpoint_auth_method": "private_key_jwt", + } + + # When / Then + with pytest.raises(ValueError, match="not yet implemented"): + _validate_cimd_document(CLIENT_ID_URL, doc) + + +def test_validate_cimd_document__no_client_name__falls_back_to_hostname() -> None: + from oauth2_metadata.cimd import _validate_cimd_document + + # Given + doc = { + "client_id": CLIENT_ID_URL, + "redirect_uris": ["https://example.com/callback"], + } + + # When + result = _validate_cimd_document(CLIENT_ID_URL, doc) + + # Then + assert result["client_name"] == "example.com" + + +def test_validate_cimd_document__invalid_redirect_uri_in_doc__raises_value_error() -> ( + None +): + from oauth2_metadata.cimd import _validate_cimd_document + + # Given + doc = { + "client_id": CLIENT_ID_URL, + "redirect_uris": ["http://evil.example.com/callback"], + } + + # When / Then + with pytest.raises(ValueError, match="redirect_uri"): + _validate_cimd_document(CLIENT_ID_URL, doc) + + +# --------------------------------------------------------------------------- +# cimd module: _fetch_cimd_document +# --------------------------------------------------------------------------- + + +def test_fetch_cimd_document__non_https__raises_value_error() -> None: + from oauth2_metadata.cimd import _fetch_cimd_document + + # Given + url = "http://example.com/metadata" + + # When / Then + with pytest.raises(ValueError, match="HTTPS"): + _fetch_cimd_document(url) + + +def test_fetch_cimd_document__url_too_long__raises_value_error() -> None: + from oauth2_metadata.cimd import _fetch_cimd_document + + # Given + long_url = "https://example.com/" + "a" * 200 + + # When / Then + with pytest.raises(ValueError, match="exceeds"): + _fetch_cimd_document(long_url) + + +def test_fetch_cimd_document__no_hostname__raises_value_error() -> None: + from oauth2_metadata.cimd import _fetch_cimd_document + + # Given + url = "https:///path-only" + + # When / Then + with pytest.raises(ValueError, match="hostname"): + _fetch_cimd_document(url) + + +def test_fetch_cimd_document__private_hostname__raises_value_error() -> None: + from oauth2_metadata.cimd import _fetch_cimd_document + + # Given + with patch("oauth2_metadata.cimd._is_public_hostname", return_value=False): + # When / Then + with pytest.raises(ValueError, match="public address"): + _fetch_cimd_document("https://internal.corp/metadata") + + +def test_fetch_cimd_document__http_error__raises_value_error() -> None: + from oauth2_metadata.cimd import _fetch_cimd_document + + # Given + with ( + patch("oauth2_metadata.cimd._is_public_hostname", return_value=True), + patch("oauth2_metadata.cimd.requests.get") as mock_get, + ): + mock_get.side_effect = requests.ConnectionError("unreachable") + + # When / Then + with pytest.raises(ValueError, match="Failed to fetch"): + _fetch_cimd_document("https://example.com/metadata") + + +def test_fetch_cimd_document__non_json_response__raises_value_error() -> None: + from oauth2_metadata.cimd import _fetch_cimd_document + + # Given + mock_response = MagicMock() + mock_response.raise_for_status.return_value = None + mock_response.json.side_effect = ValueError("not JSON") + + with ( + patch("oauth2_metadata.cimd._is_public_hostname", return_value=True), + patch("oauth2_metadata.cimd.requests.get", return_value=mock_response), + ): + # When / Then + with pytest.raises(ValueError, match="not valid JSON"): + _fetch_cimd_document("https://example.com/metadata") + + +def test_fetch_cimd_document__success__returns_dict() -> None: + from oauth2_metadata.cimd import _fetch_cimd_document + + # Given + expected = {"client_id": "https://example.com/metadata", "redirect_uris": []} + mock_response = MagicMock() + mock_response.raise_for_status.return_value = None + mock_response.json.return_value = expected + + # When + with ( + patch("oauth2_metadata.cimd._is_public_hostname", return_value=True), + patch("oauth2_metadata.cimd.requests.get", return_value=mock_response), + ): + result = _fetch_cimd_document("https://example.com/metadata") + + # Then + assert result == expected + + +# --------------------------------------------------------------------------- +# cimd module: resolve_cimd_client +# --------------------------------------------------------------------------- + + +def _mock_cimd_doc( + client_id: str = "https://example.com/oauth/metadata", + client_name: str = "Test CIMD App", + redirect_uris: list[str] | None = None, + **extra: object, +) -> dict: + doc: dict = { + "client_id": client_id, + "client_name": client_name, + "redirect_uris": redirect_uris or ["https://example.com/callback"], + } + doc.update(extra) + return doc + + +@pytest.mark.django_db() +def test_resolve_cimd_client__valid_document__creates_application() -> None: + from oauth2_metadata.cimd import resolve_cimd_client + + # Given + doc = _mock_cimd_doc(client_id=CLIENT_ID_URL) + + # When + with ( + patch("oauth2_metadata.cimd._fetch_cimd_document", return_value=doc), + patch("oauth2_metadata.cimd.cache"), + ): + app = resolve_cimd_client(CLIENT_ID_URL) + + # Then + assert app is not None + assert app.client_id == CLIENT_ID_URL + assert app.name == "Test CIMD App" + assert app.client_type == Application.CLIENT_PUBLIC + + +@pytest.mark.django_db() +def test_resolve_cimd_client__valid_document__upserts_existing_application() -> None: + from oauth2_metadata.cimd import resolve_cimd_client + + # Given + doc = _mock_cimd_doc(client_id=CLIENT_ID_URL, client_name="V1") + with ( + patch("oauth2_metadata.cimd._fetch_cimd_document", return_value=doc), + patch("oauth2_metadata.cimd.cache"), + ): + app1 = resolve_cimd_client(CLIENT_ID_URL) + + doc2 = _mock_cimd_doc(client_id=CLIENT_ID_URL, client_name="V2") + + # When + with ( + patch("oauth2_metadata.cimd._fetch_cimd_document", return_value=doc2), + patch("oauth2_metadata.cimd.cache"), + ): + app2 = resolve_cimd_client(CLIENT_ID_URL) + + # Then + assert app1 is not None + assert app2 is not None + assert app1.pk == app2.pk + app2.refresh_from_db() + assert app2.name == "V2" + + +@pytest.mark.django_db() +def test_resolve_cimd_client__fetch_failure__returns_none_and_logs( + log: StructuredLogCapture, +) -> None: + from oauth2_metadata.cimd import resolve_cimd_client + + # Given / When + with ( + patch( + "oauth2_metadata.cimd._fetch_cimd_document", + side_effect=ValueError("unreachable"), + ), + patch("oauth2_metadata.cimd.cache"), + ): + app = resolve_cimd_client(CLIENT_ID_URL) + + # Then + assert app is None + assert any(e["event"] == "cimd.rejected" for e in log.events) + + +@pytest.mark.django_db() +def test_resolve_cimd_client__fetch_failure__increments_rejected_metric( + assert_metric: AssertMetricFixture, +) -> None: + from oauth2_metadata.cimd import resolve_cimd_client + + # Given / When + with ( + patch( + "oauth2_metadata.cimd._fetch_cimd_document", + side_effect=ValueError("bad"), + ), + patch("oauth2_metadata.cimd.cache"), + ): + resolve_cimd_client(CLIENT_ID_URL) + + # Then + assert_metric( + name="flagsmith_oauth2_cimd_resolutions_total", + labels={"outcome": "rejected"}, + value=1, + ) + + +@pytest.mark.django_db() +def test_resolve_cimd_client__valid_document__increments_resolved_metric( + assert_metric: AssertMetricFixture, +) -> None: + from oauth2_metadata.cimd import resolve_cimd_client + + # Given + doc = _mock_cimd_doc(client_id=CLIENT_ID_URL) + + # When + with ( + patch("oauth2_metadata.cimd._fetch_cimd_document", return_value=doc), + patch("oauth2_metadata.cimd.cache"), + ): + resolve_cimd_client(CLIENT_ID_URL) + + # Then + assert_metric( + name="flagsmith_oauth2_cimd_resolutions_total", + labels={"outcome": "resolved"}, + value=1, + ) + + +@pytest.mark.django_db() +def test_resolve_cimd_client__cached_app__skips_fetch() -> None: + from oauth2_metadata.cimd import resolve_cimd_client + + # Given + # Pre-create an Application. + app = Application.objects.create( + client_id=CLIENT_ID_URL, + name="Cached App", + client_type=Application.CLIENT_PUBLIC, + authorization_grant_type=Application.GRANT_AUTHORIZATION_CODE, + redirect_uris="https://example.com/callback", + ) + + # When + with ( + patch("oauth2_metadata.cimd.cache") as mock_cache, + patch("oauth2_metadata.cimd._fetch_cimd_document") as mock_fetch, + ): + mock_cache.get.return_value = app.pk + result = resolve_cimd_client(CLIENT_ID_URL) + + # Then + assert result is not None + assert result.pk == app.pk + mock_fetch.assert_not_called() + + +@pytest.mark.django_db() +def test_resolve_cimd_client__cached_app_deleted__falls_through_to_fetch() -> None: + from oauth2_metadata.cimd import resolve_cimd_client + + # Given + doc = _mock_cimd_doc(client_id=CLIENT_ID_URL) + + # When + with ( + patch("oauth2_metadata.cimd.cache") as mock_cache, + patch("oauth2_metadata.cimd._fetch_cimd_document", return_value=doc), + ): + # Cache returns a PK that no longer exists. + mock_cache.get.return_value = 999999 + result = resolve_cimd_client(CLIENT_ID_URL) + + # Then + assert result is not None + assert result.client_id == CLIENT_ID_URL + mock_cache.delete.assert_called_once() diff --git a/api/tests/unit/oauth2_metadata/test_cimd_token_view.py b/api/tests/unit/oauth2_metadata/test_cimd_token_view.py new file mode 100644 index 000000000000..e351d3314c08 --- /dev/null +++ b/api/tests/unit/oauth2_metadata/test_cimd_token_view.py @@ -0,0 +1,177 @@ +"""Tests for CIMDTokenView — token endpoint with CIMD client_id support.""" + +import base64 +import hashlib +import secrets +from unittest.mock import patch + +import pytest +from django.contrib.auth.models import AbstractUser +from oauth2_provider.models import Application +from rest_framework import status +from rest_framework.test import APIClient + +CIMD_CLIENT_ID_URL = "https://cimd.example.com/oauth/metadata" + + +def _pkce_pair() -> tuple[str, str]: + """Return (code_verifier, code_challenge) for S256 PKCE.""" + code_verifier = secrets.token_urlsafe(32) + digest = hashlib.sha256(code_verifier.encode()).digest() + code_challenge = base64.urlsafe_b64encode(digest).rstrip(b"=").decode() + return code_verifier, code_challenge + + +def _mock_cimd_doc( + client_id: str = CIMD_CLIENT_ID_URL, + client_name: str = "CIMD Token App", + redirect_uris: list[str] | None = None, +) -> dict: + return { + "client_id": client_id, + "client_name": client_name, + "redirect_uris": redirect_uris or ["https://example.com/callback"], + } + + +@pytest.fixture() +def auth_client(admin_user: AbstractUser) -> APIClient: + client = APIClient() + client.force_authenticate(user=admin_user) + return client + + +def _obtain_auth_code( + auth_client: APIClient, + client_id: str, + code_challenge: str, +) -> str: + """Use the authorize endpoint to obtain an authorization code.""" + response = auth_client.post( + "/api/v1/oauth/authorize/", + { + "allow": True, + "client_id": client_id, + "response_type": "code", + "redirect_uri": "https://example.com/callback", + "scope": "mcp", + "code_challenge": code_challenge, + "code_challenge_method": "S256", + }, + format="json", + ) + assert response.status_code == status.HTTP_200_OK, response.json() + from urllib.parse import parse_qs, urlparse + + redirect_uri = response.json()["redirect_uri"] + query_params = parse_qs(urlparse(redirect_uri).query) + return query_params["code"][0] + + +def test_token__cimd_resolution_fails__returns_400( + db: None, +) -> None: + # Given + token_client = APIClient() + + # When + with ( + patch( + "oauth2_metadata.cimd._fetch_cimd_document", + side_effect=ValueError("unreachable"), + ), + patch("oauth2_metadata.cimd.cache"), + ): + response = token_client.post( + "/o/token/", + { + "grant_type": "authorization_code", + "code": "dummy-code", + "redirect_uri": "https://example.com/callback", + "client_id": CIMD_CLIENT_ID_URL, + "code_verifier": "dummy-verifier", + }, + ) + + # Then + assert response.status_code == status.HTTP_400_BAD_REQUEST + data = response.json() + assert data["error"] == "invalid_client" + + +def test_token__cimd_full_flow__issues_token( + auth_client: APIClient, + db: None, +) -> None: + # Given — obtain an auth code through the authorize endpoint. + code_verifier, code_challenge = _pkce_pair() + doc = _mock_cimd_doc() + + with ( + patch("oauth2_metadata.cimd._fetch_cimd_document", return_value=doc), + patch("oauth2_metadata.cimd._is_public_hostname", return_value=True), + patch("oauth2_metadata.cimd.cache"), + ): + code = _obtain_auth_code(auth_client, CIMD_CLIENT_ID_URL, code_challenge) + + # When — exchange the code for a token at the CIMD-aware token endpoint. + token_client = APIClient() + with ( + patch("oauth2_metadata.cimd._fetch_cimd_document", return_value=doc), + patch("oauth2_metadata.cimd._is_public_hostname", return_value=True), + patch("oauth2_metadata.cimd.cache"), + ): + response = token_client.post( + "/o/token/", + { + "grant_type": "authorization_code", + "code": code, + "redirect_uri": "https://example.com/callback", + "client_id": CIMD_CLIENT_ID_URL, + "code_verifier": code_verifier, + }, + ) + + # Then + assert response.status_code == status.HTTP_200_OK + token_data = response.json() + assert "access_token" in token_data + assert "refresh_token" in token_data + assert token_data["token_type"] == "Bearer" + + +def test_token__non_cimd_client_id__still_works( + auth_client: APIClient, + admin_user: AbstractUser, + db: None, +) -> None: + """DCR-registered (opaque) client_ids still work through CIMDTokenView.""" + # Given — create a regular application and obtain an auth code. + application = Application.objects.create( + name="Regular DCR App", + user=admin_user, + client_type=Application.CLIENT_PUBLIC, + authorization_grant_type=Application.GRANT_AUTHORIZATION_CODE, + redirect_uris="https://example.com/callback", + ) + code_verifier, code_challenge = _pkce_pair() + code = _obtain_auth_code(auth_client, application.client_id, code_challenge) + + # When — exchange the code at the token endpoint (no CIMD resolution needed). + token_client = APIClient() + response = token_client.post( + "/o/token/", + { + "grant_type": "authorization_code", + "code": code, + "redirect_uri": "https://example.com/callback", + "client_id": application.client_id, + "code_verifier": code_verifier, + }, + ) + + # Then + assert response.status_code == status.HTTP_200_OK + token_data = response.json() + assert "access_token" in token_data + assert token_data["token_type"] == "Bearer"