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.
This commit is contained in:
Herculino Trotta committed 2026-10-10 17:48:14 -03:00
1 parent f7bf96eef5
commit e925822574
2 files changed
+93 -33

No files matched your search

+50 -33
View File
@@ -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()
+43
View File
@@ -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(