Raise InvalidGrantError if no grant associated with auth code exists (#1476)

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)