Clear expired ID tokens from database

AUTHORS

@@ -62,6 +62,7 @@ Julien Palard
 Jun Zhou
 Kaleb Porter
 Kristian Rune Larsen
+Ludwig Hähne
 Matias Seniquiel
 Michael Howitz
 Owen Gong

CHANGELOG.md

@@ -19,6 +19,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Changed
 * #1211 documentation improve on 'AUTHORIZATION_CODE_EXPIRE_SECONDS'.
 * #1218 Confim support for Python 3.11.
+* #1222 Remove expired ID tokens alongside access tokens in `cleartokens` management command
 
 ## [2.2.0] 2022-10-18
 

docs/management_commands.rst

@@ -22,6 +22,8 @@ problem since refresh tokens are long lived.
 To prevent the CPU and RAM high peaks during deletion process use ``CLEAR_EXPIRED_TOKENS_BATCH_SIZE`` and
 ``CLEAR_EXPIRED_TOKENS_BATCH_INTERVAL`` settings to adjust the process speed.
 
+The ``cleartokens`` management command will also delete expired access and ID tokens alongside expired refresh tokens.
+
 Note: Refresh tokens need to expire before AccessTokens can be removed from the
 database. Using ``cleartokens`` without ``REFRESH_TOKEN_EXPIRE_SECONDS`` has limited effect.
 

oauth2_provider/models.py

@@ -663,6 +663,7 @@ def batch_delete(queryset, query):
     refresh_expire_at = None
     access_token_model = get_access_token_model()
     refresh_token_model = get_refresh_token_model()
+    id_token_model = get_id_token_model()
     grant_model = get_grant_model()
     REFRESH_TOKEN_EXPIRE_SECONDS = oauth2_settings.REFRESH_TOKEN_EXPIRE_SECONDS
 
@@ -696,6 +697,12 @@ def batch_delete(queryset, query):
     access_tokens_delete_no = batch_delete(access_tokens, access_token_query)
     logger.info("%s Expired access tokens deleted", access_tokens_delete_no)
 
+    id_token_query = models.Q(access_token__isnull=True, expires__lt=now)
+    id_tokens = id_token_model.objects.filter(id_token_query)
+
+    id_tokens_delete_no = batch_delete(id_tokens, id_token_query)
+    logger.info("%s Expired ID tokens deleted", id_tokens_delete_no)
+
     grants_query = models.Q(expires__lt=now)
     grants = grant_model.objects.filter(grants_query)
 

tests/models.py

@@ -32,6 +32,13 @@ class SampleAccessToken(AbstractAccessToken):
         null=True,
         related_name="s_refreshed_access_token",
     )
+    id_token = models.OneToOneField(
+        oauth2_settings.ID_TOKEN_MODEL,
+        on_delete=models.CASCADE,
+        blank=True,
+        null=True,
+        related_name="s_access_token",
+    )
 
 
 class SampleRefreshToken(AbstractRefreshToken):

tests/presets.py

@@ -20,6 +20,7 @@
     },
     "DEFAULT_SCOPES": ["read", "write"],
     "PKCE_REQUIRED": False,
+    "REFRESH_TOKEN_EXPIRE_SECONDS": 3600,
 }
 OIDC_SETTINGS_RO = deepcopy(OIDC_SETTINGS_RW)
 OIDC_SETTINGS_RO["DEFAULT_SCOPES"] = ["read"]

tests/test_models.py

@@ -462,6 +462,45 @@ def test_id_token_methods(oidc_tokens, rf):
     assert IDToken.objects.filter(jti=id_token.jti).count() == 0
 
 
+@pytest.mark.django_db
+@pytest.mark.oauth2_settings(presets.OIDC_SETTINGS_RW)
+def test_clear_expired_id_tokens(oauth2_settings, oidc_tokens, rf):
+    id_token = IDToken.objects.get()
+    access_token = id_token.access_token
+
+    # All tokens still valid
+    clear_expired()
+
+    assert IDToken.objects.filter(jti=id_token.jti).exists()
+
+    earlier = timezone.now() - timedelta(minutes=1)
+    id_token.expires = earlier
+    id_token.save()
+
+    # ID token should be preserved until the access token is deleted
+    clear_expired()
+
+    assert IDToken.objects.filter(jti=id_token.jti).exists()
+
+    access_token.expires = earlier
+    access_token.save()
+
+    # ID and access tokens are expired but refresh token is still valid
+    clear_expired()
+
+    assert IDToken.objects.filter(jti=id_token.jti).exists()
+
+    # Mark refresh token as expired
+    delta = timedelta(seconds=oauth2_settings.REFRESH_TOKEN_EXPIRE_SECONDS + 60)
+    access_token.expires = timezone.now() - delta
+    access_token.save()
+
+    # With the refresh token expired, the ID token should be deleted
+    clear_expired()
+
+    assert not IDToken.objects.filter(jti=id_token.jti).exists()
+
+
 @pytest.mark.django_db
 @pytest.mark.oauth2_settings(presets.OIDC_SETTINGS_RW)
 def test_application_key(oauth2_settings, application):