Fixed client_id sanitization to prevent DB errors

Author: juanifioren

docs/sections/changelog.rst

@@ -10,6 +10,7 @@ Unreleased
 
 * Added: Translation to Russian.
 * Changed: Ruff as a fast Python linter and code formatter.
+* Fixed: client_id sanitization to prevent database errors.
 
 0.8.4
 =====

oidc_provider/admin.py

@@ -6,6 +6,7 @@
 from django.forms import ModelForm
 from django.utils.translation import gettext_lazy as _
 
+from oidc_provider.lib.utils.sanitization import sanitize_client_id
 from oidc_provider.models import Client
 from oidc_provider.models import Code
 from oidc_provider.models import RSAKey
@@ -23,12 +24,15 @@ def __init__(self, *args, **kwargs):
         self.fields["client_id"].widget.attrs["disabled"] = "true"
         self.fields["client_secret"].required = False
         self.fields["client_secret"].widget.attrs["disabled"] = "true"
+        self.fields["jwt_alg"].required = False
 
     def clean_client_id(self):
         instance = getattr(self, "instance", None)
         if instance and instance.pk:
-            return instance.client_id
+            # Sanitize existing client_id to remove any problematic characters
+            return sanitize_client_id(instance.client_id)
         else:
+            # Generate new client_id (digits only)
             return str(randint(1, 999999)).zfill(6)
 
     def clean_client_secret(self):

oidc_provider/lib/endpoints/authorize.py

@@ -28,6 +28,7 @@
 from oidc_provider.lib.errors import ClientIdError
 from oidc_provider.lib.errors import RedirectUriError
 from oidc_provider.lib.utils.common import get_browser_state_or_default
+from oidc_provider.lib.utils.sanitization import sanitize_client_id
 from oidc_provider.lib.utils.token import create_code
 from oidc_provider.lib.utils.token import create_id_token
 from oidc_provider.lib.utils.token import create_token
@@ -72,7 +73,10 @@ def _extract_params(self):
         # and POST request.
         query_dict = self.request.POST if self.request.method == "POST" else self.request.GET
 
-        self.params["client_id"] = query_dict.get("client_id", "")
+        # Sanitize client_id to remove control characters that cause PostgreSQL errors
+        client_id = query_dict.get("client_id", "")
+        self.params["client_id"] = sanitize_client_id(client_id)
+
         self.params["redirect_uri"] = query_dict.get("redirect_uri", "")
         self.params["response_type"] = query_dict.get("response_type", "")
         self.params["scope"] = query_dict.get("scope", "").split()

oidc_provider/lib/endpoints/introspection.py

@@ -6,6 +6,7 @@
 from oidc_provider.lib.errors import TokenIntrospectionError
 from oidc_provider.lib.utils.common import run_processing_hook
 from oidc_provider.lib.utils.oauth2 import extract_client_auth
+from oidc_provider.lib.utils.sanitization import sanitize_client_id
 from oidc_provider.models import Client
 from oidc_provider.models import Token
 
@@ -27,7 +28,8 @@ def _extract_params(self):
         # Introspection only supports POST requests
         self.params["token"] = self.request.POST.get("token")
         client_id, client_secret = extract_client_auth(self.request)
-        self.params["client_id"] = client_id
+        # Sanitize client_id to remove control characters that cause PostgreSQL errors
+        self.params["client_id"] = sanitize_client_id(client_id)
         self.params["client_secret"] = client_secret
 
     def validate_params(self):

oidc_provider/lib/endpoints/token.py

@@ -11,6 +11,7 @@
 from oidc_provider.lib.errors import TokenError
 from oidc_provider.lib.errors import UserAuthError
 from oidc_provider.lib.utils.oauth2 import extract_client_auth
+from oidc_provider.lib.utils.sanitization import sanitize_client_id
 from oidc_provider.lib.utils.token import create_id_token
 from oidc_provider.lib.utils.token import create_token
 from oidc_provider.lib.utils.token import encode_id_token
@@ -31,7 +32,8 @@ def __init__(self, request):
     def _extract_params(self):
         client_id, client_secret = extract_client_auth(self.request)
 
-        self.params["client_id"] = client_id
+        # Sanitize client_id to remove control characters that cause PostgreSQL errors
+        self.params["client_id"] = sanitize_client_id(client_id)
         self.params["client_secret"] = client_secret
         self.params["redirect_uri"] = self.request.POST.get("redirect_uri", "")
         self.params["grant_type"] = self.request.POST.get("grant_type", "")

oidc_provider/lib/utils/sanitization.py

@@ -0,0 +1,31 @@
+import re
+
+
+def sanitize_client_id(client_id):
+    """
+    Sanitize client_id according to OAuth 2.0 RFC 6749 specification.
+
+    Removes control characters that can cause database errors while preserving
+    all valid visible ASCII characters (VCHAR: 0x21-0x7E) as defined by the
+    OAuth 2.0 specification.
+
+    Args:
+        client_id (str): The client_id parameter from the request
+
+    Returns:
+        str: Sanitized client_id with control characters removed
+
+    Examples:
+        >>> sanitize_client_id("Hello\\x00World")
+        'HelloWorld'
+        >>> sanitize_client_id("valid-client-123")
+        'valid-client-123'
+        >>> sanitize_client_id("")
+        ''
+        >>> sanitize_client_id(None)
+        ''
+    """
+    if not client_id:
+        return ""
+
+    return re.sub(r"[^\x21-\x7E]", "", client_id)

oidc_provider/tests/cases/test_admin.py

@@ -0,0 +1,98 @@
+from django.contrib.auth import get_user_model
+from django.test import TestCase
+
+from oidc_provider.admin import ClientForm
+from oidc_provider.models import Client
+from oidc_provider.models import ResponseType
+from oidc_provider.tests.app.utils import create_fake_user
+
+User = get_user_model()
+
+
+class ClientFormTest(TestCase):
+    """
+    Test cases for ClientForm in admin.
+    """
+
+    def setUp(self):
+        self.user = create_fake_user()
+        self.code_response_type, _ = ResponseType.objects.get_or_create(
+            value="code", defaults={"description": "code (Authorization Code Flow)"}
+        )
+
+    def test_creates_client_without_client_id_generates_random_one(self):
+        """Test that creating a client without client_id generates a random 6-digit one."""
+        form_data = {
+            "name": "Test Client",
+            "owner": self.user.pk,
+            "client_type": "public",
+            "response_types": [self.code_response_type.pk],
+            "_redirect_uris": "http://example.com/callback",
+        }
+
+        form = ClientForm(data=form_data)
+        self.assertTrue(form.is_valid(), f"Form errors: {form.errors}")
+
+        # The form should generate a client_id
+        client_id = form.clean_client_id()
+        self.assertIsNotNone(client_id)
+        self.assertEqual(len(client_id), 6)
+        self.assertTrue(client_id.isdigit())
+        self.assertTrue(1 <= int(client_id) <= 999999)
+
+    def test_creates_client_with_custom_client_id_preserves_it(self):
+        """Test that providing a custom client_id preserves it for new clients."""
+        # Create and save a client first
+        client = Client.objects.create(
+            name="Existing Client",
+            owner=self.user,
+            client_type="public",
+            client_id="custom-client-123",
+        )
+        client.response_types.add(self.code_response_type)
+
+        form_data = {
+            "name": "Existing Client Updated",
+            "owner": self.user.pk,
+            "client_type": "public",
+            "response_types": [self.code_response_type.pk],
+            "_redirect_uris": "http://example.com/callback",
+            "client_id": "custom-client-123",
+        }
+
+        # Test updating existing client
+        form = ClientForm(data=form_data, instance=client)
+        self.assertTrue(form.is_valid(), f"Form errors: {form.errors}")
+
+        # Should return the sanitized version of existing client_id
+        client_id = form.clean_client_id()
+        self.assertEqual(client_id, "custom-client-123")
+
+    def test_sanitizes_existing_client_id_with_control_characters(self):
+        """Test that existing client_id with control characters gets sanitized."""
+        # Create a client with problematic client_id
+        client = Client.objects.create(
+            name="Problematic Client",
+            owner=self.user,
+            client_type="public",
+            client_id="normalclient",  # Start with normal client_id
+        )
+        client.response_types.add(self.code_response_type)
+
+        # Manually set problematic client_id to test sanitization
+        client.client_id = "client\x00\x01test"  # Contains null byte and control char
+
+        form_data = {
+            "name": "Problematic Client",
+            "owner": self.user.pk,
+            "client_type": "public",
+            "response_types": [self.code_response_type.pk],
+            "_redirect_uris": "http://example.com/callback",
+        }
+
+        form = ClientForm(data=form_data, instance=client)
+        self.assertTrue(form.is_valid(), f"Form errors: {form.errors}")
+
+        # Should return sanitized client_id
+        client_id = form.clean_client_id()
+        self.assertEqual(client_id, "clienttest")  # Control characters removed

oidc_provider/tests/cases/test_authorize_endpoint.py

@@ -583,6 +583,26 @@ def test_strip_prompt_login(self):
         self.assertIn("none", strip_prompt_login(path3))
         self.assertNotIn("login", strip_prompt_login(path3))
 
+    def test_client_id_with_null_char_are_rejected(self):
+        """
+        Test that client_id parameters containing unexpected characters and
+        are properly sanitized and to not cause database errors.
+        """
+        data = {
+            "client_id": "Hello\0World",
+            "response_type": next(self.client_code.response_type_values()),
+            "redirect_uri": self.client_code.default_redirect_uri,
+            "scope": "openid email",
+            "state": self.state,
+            "prompt": "none",
+        }
+
+        response = self._auth_request("get", data)
+
+        self.assertEqual(response.status_code, 200)
+
+        self.assertIn("Client ID Error", response.content.decode("utf-8"))
+
 
 class AuthorizationImplicitFlowTestCase(TestCase, AuthorizeEndpointMixin):
     """

oidc_provider/tests/cases/test_utils.py

@@ -11,6 +11,7 @@
 
 from oidc_provider.lib.utils.common import get_browser_state_or_default
 from oidc_provider.lib.utils.common import get_issuer
+from oidc_provider.lib.utils.sanitization import sanitize_client_id
 from oidc_provider.lib.utils.token import create_id_token
 from oidc_provider.lib.utils.token import create_token
 from oidc_provider.tests.app.utils import create_fake_client
@@ -155,3 +156,62 @@ def test_get_browser_state_uses_session_key_to_calculate_browser_state_if_availa
         request.session = Mock(session_key="my_session_key")
         state = get_browser_state_or_default(request)
         self.assertEqual(state, sha224("my_session_key".encode("utf-8")).hexdigest())
+
+
+class SanitizationTest(TestCase):
+    """
+    Test cases for sanitization utils.
+    """
+
+    def test_sanitize_client_id_removes_null_bytes(self):
+        """Test that null bytes are removed from client_id."""
+        client_id = "Hello\x00World"
+        result = sanitize_client_id(client_id)
+        self.assertEqual(result, "HelloWorld")
+
+    def test_sanitize_client_id_removes_control_characters(self):
+        """Test that various control characters are removed."""
+        client_id = "client\x01\x02\x03\x1f\x7fid"
+        result = sanitize_client_id(client_id)
+        self.assertEqual(result, "clientid")
+
+    def test_sanitize_client_id_preserves_valid_characters(self):
+        """Test that valid visible ASCII characters are preserved."""
+        client_id = "valid-client_123.abc!@#$%^&*()+={}[]|\\:;\"'<>?,./~`"
+        result = sanitize_client_id(client_id)
+        self.assertEqual(result, client_id)  # Should remain unchanged
+
+    def test_sanitize_client_id_handles_empty_string(self):
+        """Test that empty string returns empty string."""
+        result = sanitize_client_id("")
+        self.assertEqual(result, "")
+
+    def test_sanitize_client_id_handles_none(self):
+        """Test that None returns empty string."""
+        result = sanitize_client_id(None)
+        self.assertEqual(result, "")
+
+    def test_sanitize_client_id_removes_whitespace_characters(self):
+        """Test that whitespace characters are removed (not part of VCHAR)."""
+        client_id = "client\t\n\r id"
+        result = sanitize_client_id(client_id)
+        self.assertEqual(result, "clientid")
+
+    def test_sanitize_client_id_preserves_printable_ascii(self):
+        """Test preservation of all printable ASCII characters (0x21-0x7E)."""
+        # All VCHAR characters as per RFC 6749
+        vchar_string = "".join(chr(i) for i in range(0x21, 0x7F))
+        result = sanitize_client_id(vchar_string)
+        self.assertEqual(result, vchar_string)
+
+    def test_sanitize_client_id_removes_unicode_characters(self):
+        """Test that Unicode characters outside ASCII range are removed."""
+        client_id = "client-ñáéíóú-测试-🔥"
+        result = sanitize_client_id(client_id)
+        self.assertEqual(result, "client---")
+
+    def test_sanitize_client_id_mixed_valid_invalid(self):
+        """Test mixed valid and invalid characters."""
+        client_id = "valid\x00client\x01-\x7f123\tabc"
+        result = sanitize_client_id(client_id)
+        self.assertEqual(result, "validclient-123abc")