Raise InvalidGrantError if no grant associated with auth code exists

Previously, when invalidating an authorization code after it has been used, if for whatever reason the associated grant object no longer exists, an uncaught exception would be raised - Grant.DoesNotExist.

This could be caused by concurrent requests being made using the same authorization token.

We now handle this scenario gracefully by catching Grant.DoesNotExist and returning an InvalidGrantError.

Author: mforner13

AUTHORS

@@ -119,3 +119,4 @@ pySilver
 Wouter Klein Heerenbrink
 Yaroslav Halchenko
 Yuri Savin
+Miriam Forner

CHANGELOG.md

@@ -24,6 +24,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 * Update middleware, validators, and views to use token checksums instead of token for token retrieval and validation.
 * #1446 use generic models pk instead of id.
 * Bump oauthlib version to 3.2.0 and above
+* Update the OAuth2Validator's invalidate_authorization_code method to return an InvalidGrantError if the associated grant does not exist.
 
 ### Deprecated
 ### Removed

oauth2_provider/oauth2_validators.py

@@ -24,7 +24,7 @@
 from jwcrypto import jws, jwt
 from jwcrypto.common import JWException
 from jwcrypto.jwt import JWTExpired
-from oauthlib.oauth2.rfc6749 import utils
+from oauthlib.oauth2.rfc6749 import errors, utils
 from oauthlib.openid import RequestValidator
 
 from .exceptions import FatalClientError
@@ -318,10 +318,15 @@ def confirm_redirect_uri(self, client_id, code, redirect_uri, client, *args, **k
 
     def invalidate_authorization_code(self, client_id, code, request, *args, **kwargs):
         """
-        Remove the temporary grant used to swap the authorization token
+        Remove the temporary grant used to swap the authorization token.
+
+        :raises: InvalidGrantError if the grant does not exist.
         """
-        grant = Grant.objects.get(code=code, application=request.client)
-        grant.delete()
+        try:
+            grant = Grant.objects.get(code=code, application=request.client)
+            grant.delete()
+        except Grant.DoesNotExist:
+            raise errors.InvalidGrantError(request=request)
 
     def validate_client_id(self, client_id, request, *args, **kwargs):
         """

tests/test_oauth2_validators.py

@@ -9,9 +9,15 @@
 from django.utils import timezone
 from jwcrypto import jwt
 from oauthlib.common import Request
+from oauthlib.oauth2.rfc6749 import errors as rfc6749_errors
 
 from oauth2_provider.exceptions import FatalClientError
-from oauth2_provider.models import get_access_token_model, get_application_model, get_refresh_token_model
+from oauth2_provider.models import (
+    get_access_token_model,
+    get_application_model,
+    get_grant_model,
+    get_refresh_token_model,
+)
 from oauth2_provider.oauth2_backends import get_oauthlib_core
 from oauth2_provider.oauth2_validators import OAuth2Validator
 
@@ -28,6 +34,7 @@
 UserModel = get_user_model()
 Application = get_application_model()
 AccessToken = get_access_token_model()
+Grant = get_grant_model()
 RefreshToken = get_refresh_token_model()
 
 CLEARTEXT_SECRET = "1234567890abcdefghijklmnopqrstuvwxyz"
@@ -578,3 +585,14 @@ def test_validate_id_token_bad_token_no_aud(oauth2_settings, mocker, oidc_key):
     validator = OAuth2Validator()
     status = validator.validate_id_token(token.serialize(), ["openid"], mocker.sentinel.request)
     assert status is False
+
+
+@pytest.mark.django_db
+def test_invalidate_authorization_token_returns_invalid_grant_error_when_grant_does_not_exist():
+    client_id = "123"
+    code = "12345"
+    request = Request("/")
+    assert Grant.objects.all().count() == 0
+    with pytest.raises(rfc6749_errors.InvalidGrantError):
+        validator = OAuth2Validator()
+        validator.invalidate_authorization_code(client_id=client_id, code=code, request=request)