mirror of
https://github.com/eitchtee/WYGIWYH.git
synced 2026-11-13 12:23:44 +01:00
Merge pull request #569 from eitchtee/copilot/fix-code-scanning-alerts
Stop exposing exception text in OAuth dynamic client registration errors
This commit is contained in:
2 files changed
+95
-34
No files matched your search
@@ -1,10 +1,11 @@
|
||||
import hmac
|
||||
import json
|
||||
import logging
|
||||
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
|
||||
@@ -12,6 +13,7 @@ from oauth2_provider.models import get_application_model
|
||||
|
||||
|
||||
Application = get_application_model()
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
SUPPORTED_TOKEN_ENDPOINT_AUTH_METHODS = {
|
||||
"none": Application.CLIENT_PUBLIC,
|
||||
@@ -45,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():
|
||||
@@ -131,21 +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:
|
||||
return _json_error("invalid_client_metadata", str(exc))
|
||||
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:
|
||||
@@ -178,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()
|
||||
@@ -223,13 +235,19 @@ def dynamic_client_registration(request):
|
||||
try:
|
||||
application.full_clean()
|
||||
except ValidationError as exc:
|
||||
errors = []
|
||||
for field, messages in exc.message_dict.items():
|
||||
errors.extend(f"{field}: {message}" for message in messages)
|
||||
return _json_error(
|
||||
"invalid_client_metadata",
|
||||
"; ".join(errors),
|
||||
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()
|
||||
|
||||
|
||||
@@ -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(
|
||||
|
||||
Reference in new issue
Block a user