From e92582257453cc17a7338e464b05a7812db75191 Mon Sep 17 00:00:00 2001 From: Herculino Trotta Date: Sat, 10 Oct 2026 17:48:14 -0300 Subject: [PATCH] fix(oauth): keep specific DCR errors without exposing exception text Parsing helpers now return (value, error) instead of raising, so no exception object flows into the response while clients keep the precise messages they had before. full_clean() failures map to fixed per-field messages; redirect URI errors surface as NON_FIELD_ERRORS from Application.clean(). Also reject non-string token_endpoint_auth_method with a 400 instead of crashing with a TypeError. --- app/apps/common/oauth_views.py | 83 +++++++++++++++++------------ app/apps/common/tests/test_oauth.py | 43 +++++++++++++++ 2 files changed, 93 insertions(+), 33 deletions(-) diff --git a/app/apps/common/oauth_views.py b/app/apps/common/oauth_views.py index b508eb1e..54733580 100644 --- a/app/apps/common/oauth_views.py +++ b/app/apps/common/oauth_views.py @@ -5,7 +5,7 @@ import time from secrets import token_urlsafe from django.conf import settings -from django.core.exceptions import ValidationError +from django.core.exceptions import NON_FIELD_ERRORS, ValidationError from django.http import JsonResponse from django.views.decorators.csrf import csrf_exempt from django.views.decorators.http import require_http_methods @@ -47,31 +47,31 @@ def _set_no_store_headers(response): def _parse_json_request_body(request): try: payload = json.loads(request.body.decode("utf-8")) - except (UnicodeDecodeError, json.JSONDecodeError) as exc: - raise ValueError("Request body must be valid JSON.") from exc + except (UnicodeDecodeError, json.JSONDecodeError): + return None, "Request body must be valid JSON." if not isinstance(payload, dict): - raise ValueError("Request body must be a JSON object.") + return None, "Request body must be a JSON object." - return payload + return payload, None def _get_string_list(payload, field_name, *, required=False, default=None): value = payload.get(field_name, default) if value is None: if required: - raise ValueError(f"'{field_name}' is required.") - return None + return None, f"'{field_name}' is required." + return None, None if not isinstance(value, list) or not value: - raise ValueError(f"'{field_name}' must be a non-empty array of strings.") + return None, f"'{field_name}' must be a non-empty array of strings." normalized = [] for item in value: if not isinstance(item, str) or not item.strip(): - raise ValueError(f"'{field_name}' must contain only non-empty strings.") + return None, f"'{field_name}' must contain only non-empty strings." normalized.append(item.strip()) - return normalized + return normalized, None def _get_supported_scopes(): @@ -133,22 +133,29 @@ def dynamic_client_registration(request): status=401, ) - try: - payload = _parse_json_request_body(request) - redirect_uris = _get_string_list(payload, "redirect_uris", required=True) - grant_types = _get_string_list( - payload, - "grant_types", - default=["authorization_code"], - ) - response_types = _get_string_list( - payload, - "response_types", - default=["code"], - ) - except ValueError as exc: - logger.warning("Invalid dynamic client registration payload: %s", exc) - return _json_error("invalid_client_metadata", "Client metadata is invalid.") + payload, error = _parse_json_request_body(request) + if error: + return _json_error("invalid_client_metadata", error) + + redirect_uris, error = _get_string_list(payload, "redirect_uris", required=True) + if error: + return _json_error("invalid_client_metadata", error) + + grant_types, error = _get_string_list( + payload, + "grant_types", + default=["authorization_code"], + ) + if error: + return _json_error("invalid_client_metadata", error) + + response_types, error = _get_string_list( + payload, + "response_types", + default=["code"], + ) + if error: + return _json_error("invalid_client_metadata", error) unsupported_grant_types = sorted(set(grant_types) - SUPPORTED_GRANT_TYPES) if unsupported_grant_types: @@ -181,11 +188,13 @@ def dynamic_client_registration(request): "token_endpoint_auth_method", "client_secret_basic", ) - if token_endpoint_auth_method not in SUPPORTED_TOKEN_ENDPOINT_AUTH_METHODS: + if ( + not isinstance(token_endpoint_auth_method, str) + or token_endpoint_auth_method not in SUPPORTED_TOKEN_ENDPOINT_AUTH_METHODS + ): return _json_error( "invalid_client_metadata", - "Unsupported token_endpoint_auth_method: " - + token_endpoint_auth_method, + "Unsupported token_endpoint_auth_method.", ) supported_scopes = _get_supported_scopes() @@ -226,11 +235,19 @@ def dynamic_client_registration(request): try: application.full_clean() except ValidationError as exc: - logger.warning("Dynamic client registration validation failed: %s", exc) - return _json_error( - "invalid_client_metadata", - "Client metadata is invalid.", + logger.warning( + "Dynamic client registration validation failed: %s", exc.message_dict ) + # Map to fixed messages so validator text never reaches the client. + # Application.clean() validates redirect URIs as non-field errors; with + # the fields set above, that is the only non-field error it can raise. + if "redirect_uris" in exc.message_dict or NON_FIELD_ERRORS in exc.message_dict: + description = "redirect_uris contains an invalid or disallowed URI." + elif "name" in exc.message_dict: + description = "client_name is invalid." + else: + description = "Client metadata is invalid." + return _json_error("invalid_client_metadata", description) application.save() diff --git a/app/apps/common/tests/test_oauth.py b/app/apps/common/tests/test_oauth.py index b6d7edd6..b096ada9 100644 --- a/app/apps/common/tests/test_oauth.py +++ b/app/apps/common/tests/test_oauth.py @@ -150,6 +150,49 @@ class DynamicClientRegistrationTests(TestCase): self.assertEqual(response.json()["error"], "invalid_client_metadata") self.assertIn("redirect_uris", response.json()["error_description"]) + def test_rejects_invalid_json_body(self): + response = self.client.post( + reverse("oauth-dynamic-client-registration"), + data="{not json", + content_type="application/json", + ) + + self.assertEqual(response.status_code, 400) + self.assertEqual(response.json()["error"], "invalid_client_metadata") + self.assertEqual( + response.json()["error_description"], "Request body must be valid JSON." + ) + + def test_rejects_disallowed_redirect_uri_scheme(self): + response = self.client.post( + reverse("oauth-dynamic-client-registration"), + data=json.dumps({"redirect_uris": ["ftp://example.com/callback"]}), + content_type="application/json", + ) + + self.assertEqual(response.status_code, 400) + self.assertEqual(response.json()["error"], "invalid_client_metadata") + self.assertEqual( + response.json()["error_description"], + "redirect_uris contains an invalid or disallowed URI.", + ) + + def test_rejects_non_string_token_auth_method(self): + response = self.client.post( + reverse("oauth-dynamic-client-registration"), + data=json.dumps( + { + "redirect_uris": ["http://127.0.0.1:8765/callback"], + "token_endpoint_auth_method": ["none"], + } + ), + content_type="application/json", + ) + + self.assertEqual(response.status_code, 400) + self.assertEqual(response.json()["error"], "invalid_client_metadata") + self.assertIn("token_endpoint_auth_method", response.json()["error_description"]) + @override_settings(OAUTH2_DCR_ENABLED=False) def test_returns_404_when_dcr_disabled(self): response = self.client.post(