Fix access token 500 (#1337)

* try/except when looking for an access token to avoid 500

* try/except when looking for an access token to avoid 500

* adding additionnal tests

* adding a test for using a deleted token

* returning an empty array, updating tests

* updating CHANGELOG and AUTHORS

Author: discobeta

AUTHORS

@@ -26,6 +26,7 @@ Antoine Laurent
 Anvesh Agarwal
 Aristóbulo Meneses
 Aryan Iyappan
+Asaf Klibansky
 Ash Christopher
 Asif Saif Uddin
 Bart Merenda

CHANGELOG.md

@@ -24,6 +24,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 * #1273 Add caching of loading of OIDC private key.
 * #1285 Add post_logout_redirect_uris field in application views.
 * #1311 Add option to disable client_secret hashing to allow verifying JWTs' signatures.
+* #1337 Gracefully handle expired or deleted refresh tokens, in `validate_user`.
 
 - ### Fixed
 * #1322 Instructions in documentation on how to create a code challenge and code verifier

oauth2_provider/oauth2_validators.py

@@ -725,8 +725,10 @@ def get_original_scopes(self, refresh_token, request, *args, **kwargs):
         # validate_refresh_token.
         rt = request.refresh_token_instance
         if not rt.access_token_id:
-            return AccessToken.objects.get(source_refresh_token_id=rt.id).scope
-
+            try:
+                return AccessToken.objects.get(source_refresh_token_id=rt.id).scope
+            except AccessToken.DoesNotExist:
+                return []
         return rt.access_token.scope
 
     def validate_refresh_token(self, refresh_token, client, request, *args, **kwargs):

tests/test_authorization_code.py

@@ -1002,6 +1002,39 @@ def test_refresh_repeating_requests_non_rotating_tokens(self):
         response = self.client.post(reverse("oauth2_provider:token"), data=token_request_data, **auth_headers)
         self.assertEqual(response.status_code, 200)
 
+    def test_refresh_with_deleted_token(self):
+        """
+        Ensure that using a deleted refresh token returns 400
+        """
+        self.client.login(username="test_user", password="123456")
+        authorization_code = self.get_auth()
+
+        token_request_data = {
+            "grant_type": "authorization_code",
+            "scope": "read write",
+            "code": authorization_code,
+            "redirect_uri": "http://example.org",
+        }
+        auth_headers = get_basic_auth_header(self.application.client_id, CLEARTEXT_SECRET)
+
+        # get a refresh token
+        response = self.client.post(reverse("oauth2_provider:token"), data=token_request_data, **auth_headers)
+
+        content = json.loads(response.content.decode("utf-8"))
+        rt = content["refresh_token"]
+
+        token_request_data = {
+            "grant_type": "refresh_token",
+            "refresh_token": rt,
+            "scope": "read write",
+        }
+
+        # delete the access token
+        AccessToken.objects.filter(token=content["access_token"]).delete()
+
+        response = self.client.post(reverse("oauth2_provider:token"), data=token_request_data, **auth_headers)
+        self.assertEqual(response.status_code, 400)
+
     def test_basic_auth_bad_authcode(self):
         """
         Request an access token using a bad authorization code