Always hash application.client_secret.

Apply suggestions from code review

Co-authored-by: Andrew Chen Wang <60190294+Andrew-Chen-Wang@users.noreply.github.com>

Author: n2ygk

CHANGELOG.md

@@ -16,8 +16,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [unreleased]
 
-### Added
-* #729 Add support for [hashed client_secret values](https://django-oauth-toolkit.readthedocs.io/en/latest/settings.html#client-secret-hasher).
+## [2.0.0] unreleased
+
+### Changed
+* #1093 (**Breaking**) Changed to implement [hashed](https://docs.djangoproject.com/en/stable/topics/auth/passwords/)
+  client_secret values. This is a **breaking change** that will migrate all your existing
+  cleartext `application.client_secret` values to be hashed with Django's default password hashing algorithm
+  and can not be reversed. When adding or modifying an Application in the Admin console, you must copy the
+  auto-generated or manually-entered `client_secret` before hitting Save.
+
 
 ## [1.7.0] 2022-01-23
 

oauth2_provider/migrations/0006_alter_application_client_secret.py

@@ -1,10 +1,20 @@
-# Generated by Django 4.0.1 on 2022-01-07 22:40
-
 from django.db import migrations
+from django.contrib.auth.hashers import identify_hasher, make_password
+import logging
 import oauth2_provider.generators
 import oauth2_provider.models
 
 
+def forwards_func(apps, schema_editor):
+    """
+    Forward migration touches every application.client_secret which will cause it to be hashed if not already the case.
+    """
+    Application = apps.get_model('oauth2_provider', 'application')
+    applications = Application.objects.all()
+    for application in applications:
+        application.save(update_fields=['client_secret'])
+
+
 class Migration(migrations.Migration):
 
     dependencies = [
@@ -15,6 +25,7 @@ class Migration(migrations.Migration):
         migrations.AlterField(
             model_name='application',
             name='client_secret',
-            field=oauth2_provider.models.ClientSecretField(blank=True, db_index=True, default=oauth2_provider.generators.generate_client_secret, max_length=255),
+            field=oauth2_provider.models.ClientSecretField(blank=True, db_index=True, default=oauth2_provider.generators.generate_client_secret, help_text='Hashed on Save. Copy it now if this is a new secret.', max_length=255),
         ),
+        migrations.RunPython(forwards_func),
     ]

oauth2_provider/models.py

@@ -6,7 +6,7 @@
 
 from django.apps import apps
 from django.conf import settings
-from django.contrib.auth.hashers import make_password
+from django.contrib.auth.hashers import identify_hasher, make_password
 from django.core.exceptions import ImproperlyConfigured
 from django.db import models, transaction
 from django.urls import reverse
@@ -27,14 +27,15 @@
 
 class ClientSecretField(models.CharField):
     def pre_save(self, model_instance, add):
-        if oauth2_settings.CLIENT_SECRET_HASHER:
-            plain_secret = getattr(model_instance, self.attname)
-            if "$" not in plain_secret:  # not yet hashed
-                hashed_secret = make_password(
-                    plain_secret, salt=model_instance.client_id, hasher=oauth2_settings.CLIENT_SECRET_HASHER
-                )
-                setattr(model_instance, self.attname, hashed_secret)
-                return hashed_secret
+        secret = getattr(model_instance, self.attname)
+        try:
+            hasher = identify_hasher(secret)
+            logger.debug(f"{model_instance}: {self.attname} is already hashed with {hasher}.")
+        except ValueError:
+            logger.debug(f"{model_instance}: {self.attname} is not hashed; hashing it now.")
+            hashed_secret = make_password(secret)
+            setattr(model_instance, self.attname, hashed_secret)
+            return hashed_secret
         return super().pre_save(model_instance, add)
 
 
@@ -105,7 +106,11 @@ class AbstractApplication(models.Model):
     client_type = models.CharField(max_length=32, choices=CLIENT_TYPES)
     authorization_grant_type = models.CharField(max_length=32, choices=GRANT_TYPES)
     client_secret = ClientSecretField(
-        max_length=255, blank=True, default=generate_client_secret, db_index=True
+        max_length=255,
+        blank=True,
+        default=generate_client_secret,
+        db_index=True,
+        help_text=_("Hashed on Save. Copy it now if this is a new secret."),
     )
     name = models.CharField(max_length=255, blank=True)
     skip_authorization = models.BooleanField(default=False)

oauth2_provider/oauth2_validators.py

@@ -124,18 +124,7 @@ def _authenticate_basic_auth(self, request):
         elif request.client.client_id != client_id:
             log.debug("Failed basic auth: wrong client id %s" % client_id)
             return False
-        # we use the "$" as a sentinel character to determine
-        # whether a secret has been hashed like a Django password or not.
-        # We can do this because the default oauthlib.common.UNICODE_ASCII_CHARACTER_SET
-        # used by our default generator does not include the "$" character.
-        # However, if a different character set was used to generate the secret, this sentinel
-        # might be a false positive.
-        elif "$" in request.client.client_secret and request.client.client_secret != client_secret:
-            if not check_password(client_secret, request.client.client_secret):
-                log.debug("Failed basic auth: wrong hashed client secret %s" % client_secret)
-                return False
-            return True
-        elif request.client.client_secret != client_secret:
+        elif not check_password(client_secret, request.client.client_secret):
             log.debug("Failed basic auth: wrong client secret %s" % client_secret)
             return False
         else:
@@ -160,7 +149,7 @@ def _authenticate_request_body(self, request):
         if self._load_application(client_id, request) is None:
             log.debug("Failed body auth: Application %s does not exists" % client_id)
             return False
-        elif request.client.client_secret != client_secret:
+        elif not check_password(client_secret, request.client.client_secret):
             log.debug("Failed body auth: wrong client secret %s" % client_secret)
             return False
         else:

oauth2_provider/settings.py

@@ -37,7 +37,6 @@
     "CLIENT_ID_GENERATOR_CLASS": "oauth2_provider.generators.ClientIdGenerator",
     "CLIENT_SECRET_GENERATOR_CLASS": "oauth2_provider.generators.ClientSecretGenerator",
     "CLIENT_SECRET_GENERATOR_LENGTH": 128,
-    "CLIENT_SECRET_HASHER": None,
     "ACCESS_TOKEN_GENERATOR": None,
     "REFRESH_TOKEN_GENERATOR": None,
     "EXTRA_SERVER_KWARGS": {},

tests/conftest.py

@@ -16,6 +16,8 @@
 Application = get_application_model()
 UserModel = get_user_model()
 
+CLEARTEXT_SECRET = "1234567890abcdefghijklmnopqrstuvwxyz"
+
 
 class OAuthSettingsWrapper:
     """
@@ -101,12 +103,14 @@ def application():
         client_type=Application.CLIENT_CONFIDENTIAL,
         authorization_grant_type=Application.GRANT_AUTHORIZATION_CODE,
         algorithm=Application.RS256_ALGORITHM,
+        client_secret=CLEARTEXT_SECRET,
     )
 
 
 @pytest.fixture
 def hybrid_application(application):
     application.authorization_grant_type = application.GRANT_OPENID_HYBRID
+    application.client_secret = CLEARTEXT_SECRET
     application.save()
     return application
 
@@ -141,7 +145,7 @@ def oidc_tokens(oauth2_settings, application, test_user, client):
             "code": code,
             "redirect_uri": "http://example.org",
             "client_id": application.client_id,
-            "client_secret": application.client_secret,
+            "client_secret": CLEARTEXT_SECRET,
             "scope": "openid",
         },
     )

tests/presets.py

@@ -44,6 +44,3 @@
     "READ_SCOPE": "read",
     "WRITE_SCOPE": "write",
 }
-
-# default django auth hasher as of version 3.2
-CLIENT_SECRET_HASHER = {"CLIENT_SECRET_HASHER": "pbkdf2_sha256"}

tests/test_authorization_code.py

@@ -34,6 +34,7 @@
 
 URI_OOB = "urn:ietf:wg:oauth:2.0:oob"
 URI_OOB_AUTO = "urn:ietf:wg:oauth:2.0:oob:auto"
+CLEARTEXT_SECRET = "1234567890abcdefghijklmnopqrstuvwxyz"
 
 
 # mocking a protected resource view
@@ -60,6 +61,7 @@ def setUp(self):
             user=self.dev_user,
             client_type=Application.CLIENT_CONFIDENTIAL,
             authorization_grant_type=Application.GRANT_AUTHORIZATION_CODE,
+            client_secret=CLEARTEXT_SECRET,
         )
 
     def tearDown(self):
@@ -677,7 +679,7 @@ def test_basic_auth(self):
             "code": authorization_code,
             "redirect_uri": "http://example.org",
         }
-        auth_headers = get_basic_auth_header(self.application.client_id, self.application.client_secret)
+        auth_headers = get_basic_auth_header(self.application.client_id, CLEARTEXT_SECRET)
 
         response = self.client.post(reverse("oauth2_provider:token"), data=token_request_data, **auth_headers)
         self.assertEqual(response.status_code, 200)
@@ -699,7 +701,7 @@ def test_refresh(self):
             "code": authorization_code,
             "redirect_uri": "http://example.org",
         }
-        auth_headers = get_basic_auth_header(self.application.client_id, self.application.client_secret)
+        auth_headers = get_basic_auth_header(self.application.client_id, CLEARTEXT_SECRET)
 
         response = self.client.post(reverse("oauth2_provider:token"), data=token_request_data, **auth_headers)
         content = json.loads(response.content.decode("utf-8"))
@@ -744,7 +746,7 @@ def test_refresh_with_grace_period(self):
             "code": authorization_code,
             "redirect_uri": "http://example.org",
         }
-        auth_headers = get_basic_auth_header(self.application.client_id, self.application.client_secret)
+        auth_headers = get_basic_auth_header(self.application.client_id, CLEARTEXT_SECRET)
 
         response = self.client.post(reverse("oauth2_provider:token"), data=token_request_data, **auth_headers)
         content = json.loads(response.content.decode("utf-8"))
@@ -795,7 +797,7 @@ def test_refresh_invalidates_old_tokens(self):
             "code": authorization_code,
             "redirect_uri": "http://example.org",
         }
-        auth_headers = get_basic_auth_header(self.application.client_id, self.application.client_secret)
+        auth_headers = get_basic_auth_header(self.application.client_id, CLEARTEXT_SECRET)
 
         response = self.client.post(reverse("oauth2_provider:token"), data=token_request_data, **auth_headers)
         content = json.loads(response.content.decode("utf-8"))
@@ -827,7 +829,7 @@ def test_refresh_no_scopes(self):
             "code": authorization_code,
             "redirect_uri": "http://example.org",
         }
-        auth_headers = get_basic_auth_header(self.application.client_id, self.application.client_secret)
+        auth_headers = get_basic_auth_header(self.application.client_id, CLEARTEXT_SECRET)
 
         response = self.client.post(reverse("oauth2_provider:token"), data=token_request_data, **auth_headers)
         content = json.loads(response.content.decode("utf-8"))
@@ -855,7 +857,7 @@ def test_refresh_bad_scopes(self):
             "code": authorization_code,
             "redirect_uri": "http://example.org",
         }
-        auth_headers = get_basic_auth_header(self.application.client_id, self.application.client_secret)
+        auth_headers = get_basic_auth_header(self.application.client_id, CLEARTEXT_SECRET)
 
         response = self.client.post(reverse("oauth2_provider:token"), data=token_request_data, **auth_headers)
         content = json.loads(response.content.decode("utf-8"))
@@ -881,7 +883,7 @@ def test_refresh_fail_repeating_requests(self):
             "code": authorization_code,
             "redirect_uri": "http://example.org",
         }
-        auth_headers = get_basic_auth_header(self.application.client_id, self.application.client_secret)
+        auth_headers = get_basic_auth_header(self.application.client_id, CLEARTEXT_SECRET)
 
         response = self.client.post(reverse("oauth2_provider:token"), data=token_request_data, **auth_headers)
         content = json.loads(response.content.decode("utf-8"))
@@ -911,7 +913,7 @@ def test_refresh_repeating_requests(self):
             "code": authorization_code,
             "redirect_uri": "http://example.org",
         }
-        auth_headers = get_basic_auth_header(self.application.client_id, self.application.client_secret)
+        auth_headers = get_basic_auth_header(self.application.client_id, CLEARTEXT_SECRET)
 
         response = self.client.post(reverse("oauth2_provider:token"), data=token_request_data, **auth_headers)
         content = json.loads(response.content.decode("utf-8"))
@@ -948,7 +950,7 @@ def test_refresh_repeating_requests_non_rotating_tokens(self):
             "code": authorization_code,
             "redirect_uri": "http://example.org",
         }
-        auth_headers = get_basic_auth_header(self.application.client_id, self.application.client_secret)
+        auth_headers = get_basic_auth_header(self.application.client_id, CLEARTEXT_SECRET)
 
         response = self.client.post(reverse("oauth2_provider:token"), data=token_request_data, **auth_headers)
         content = json.loads(response.content.decode("utf-8"))
@@ -977,7 +979,7 @@ def test_basic_auth_bad_authcode(self):
             "code": "BLAH",
             "redirect_uri": "http://example.org",
         }
-        auth_headers = get_basic_auth_header(self.application.client_id, self.application.client_secret)
+        auth_headers = get_basic_auth_header(self.application.client_id, CLEARTEXT_SECRET)
 
         response = self.client.post(reverse("oauth2_provider:token"), data=token_request_data, **auth_headers)
         self.assertEqual(response.status_code, 400)
@@ -989,7 +991,7 @@ def test_basic_auth_bad_granttype(self):
         self.client.login(username="test_user", password="123456")
 
         token_request_data = {"grant_type": "UNKNOWN", "code": "BLAH", "redirect_uri": "http://example.org"}
-        auth_headers = get_basic_auth_header(self.application.client_id, self.application.client_secret)
+        auth_headers = get_basic_auth_header(self.application.client_id, CLEARTEXT_SECRET)
 
         response = self.client.post(reverse("oauth2_provider:token"), data=token_request_data, **auth_headers)
         self.assertEqual(response.status_code, 400)
@@ -1014,7 +1016,7 @@ def test_basic_auth_grant_expired(self):
             "code": "BLAH",
             "redirect_uri": "http://example.org",
         }
-        auth_headers = get_basic_auth_header(self.application.client_id, self.application.client_secret)
+        auth_headers = get_basic_auth_header(self.application.client_id, CLEARTEXT_SECRET)
 
         response = self.client.post(reverse("oauth2_provider:token"), data=token_request_data, **auth_headers)
         self.assertEqual(response.status_code, 400)
@@ -1049,7 +1051,7 @@ def test_basic_auth_wrong_auth_type(self):
             "redirect_uri": "http://example.org",
         }
 
-        user_pass = "{0}:{1}".format(self.application.client_id, self.application.client_secret)
+        user_pass = "{0}:{1}".format(self.application.client_id, CLEARTEXT_SECRET)
         auth_string = base64.b64encode(user_pass.encode("utf-8"))
         auth_headers = {
             "HTTP_AUTHORIZATION": "Wrong " + auth_string.decode("utf-8"),
@@ -1070,7 +1072,7 @@ def test_request_body_params(self):
             "code": authorization_code,
             "redirect_uri": "http://example.org",
             "client_id": self.application.client_id,
-            "client_secret": self.application.client_secret,
+            "client_secret": CLEARTEXT_SECRET,
         }
 
         response = self.client.post(reverse("oauth2_provider:token"), data=token_request_data)
@@ -1445,7 +1447,7 @@ def test_code_exchange_succeed_when_redirect_uri_match(self):
             "code": authorization_code,
             "redirect_uri": "http://example.org?foo=bar",
         }
-        auth_headers = get_basic_auth_header(self.application.client_id, self.application.client_secret)
+        auth_headers = get_basic_auth_header(self.application.client_id, CLEARTEXT_SECRET)
 
         response = self.client.post(reverse("oauth2_provider:token"), data=token_request_data, **auth_headers)
         self.assertEqual(response.status_code, 200)
@@ -1480,7 +1482,7 @@ def test_code_exchange_fails_when_redirect_uri_does_not_match(self):
             "code": authorization_code,
             "redirect_uri": "http://example.org?foo=baraa",
         }
-        auth_headers = get_basic_auth_header(self.application.client_id, self.application.client_secret)
+        auth_headers = get_basic_auth_header(self.application.client_id, CLEARTEXT_SECRET)
 
         response = self.client.post(reverse("oauth2_provider:token"), data=token_request_data, **auth_headers)
         self.assertEqual(response.status_code, 400)
@@ -1520,7 +1522,7 @@ def test_code_exchange_succeed_when_redirect_uri_match_with_multiple_query_param
             "code": authorization_code,
             "redirect_uri": "http://example.com?bar=baz&foo=bar",
         }
-        auth_headers = get_basic_auth_header(self.application.client_id, self.application.client_secret)
+        auth_headers = get_basic_auth_header(self.application.client_id, CLEARTEXT_SECRET)
 
         response = self.client.post(reverse("oauth2_provider:token"), data=token_request_data, **auth_headers)
         self.assertEqual(response.status_code, 200)
@@ -1565,7 +1567,7 @@ def test_oob_as_html(self):
             "code": authorization_code,
             "redirect_uri": URI_OOB,
             "client_id": self.application.client_id,
-            "client_secret": self.application.client_secret,
+            "client_secret": CLEARTEXT_SECRET,
         }
 
         response = self.client.post(reverse("oauth2_provider:token"), data=token_request_data)
@@ -1605,7 +1607,7 @@ def test_oob_as_json(self):
             "code": authorization_code,
             "redirect_uri": URI_OOB_AUTO,
             "client_id": self.application.client_id,
-            "client_secret": self.application.client_secret,
+            "client_secret": CLEARTEXT_SECRET,
         }
 
         response = self.client.post(reverse("oauth2_provider:token"), data=token_request_data)
@@ -1681,7 +1683,7 @@ def test_id_token_code_exchange_succeed_when_redirect_uri_match_with_multiple_qu
             "code": authorization_code,
             "redirect_uri": "http://example.com?bar=baz&foo=bar",
         }
-        auth_headers = get_basic_auth_header(self.application.client_id, self.application.client_secret)
+        auth_headers = get_basic_auth_header(self.application.client_id, CLEARTEXT_SECRET)
 
         response = self.client.post(reverse("oauth2_provider:token"), data=token_request_data, **auth_headers)
         self.assertEqual(response.status_code, 200)
@@ -1715,7 +1717,7 @@ def test_id_token(self):
             "code": authorization_code,
             "redirect_uri": "http://example.org",
             "client_id": self.application.client_id,
-            "client_secret": self.application.client_secret,
+            "client_secret": CLEARTEXT_SECRET,
             "scope": "openid",
         }
 
@@ -1761,7 +1763,7 @@ def test_resource_access_allowed(self):
             "code": authorization_code,
             "redirect_uri": "http://example.org",
         }
-        auth_headers = get_basic_auth_header(self.application.client_id, self.application.client_secret)
+        auth_headers = get_basic_auth_header(self.application.client_id, CLEARTEXT_SECRET)
 
         response = self.client.post(reverse("oauth2_provider:token"), data=token_request_data, **auth_headers)
         content = json.loads(response.content.decode("utf-8"))
@@ -1819,7 +1821,7 @@ def test_id_token_resource_access_allowed(self):
             "code": authorization_code,
             "redirect_uri": "http://example.org",
         }
-        auth_headers = get_basic_auth_header(self.application.client_id, self.application.client_secret)
+        auth_headers = get_basic_auth_header(self.application.client_id, CLEARTEXT_SECRET)
 
         response = self.client.post(reverse("oauth2_provider:token"), data=token_request_data, **auth_headers)
         content = json.loads(response.content.decode("utf-8"))