feat: refresh tokens renewal based on user consent timeframe validity…

… with new settingslocal var OIDCFED_PROVIDER_MAX_CONSENT_TIMEFRAME #seconds

examples/provider/provider/settingslocal.py.example

@@ -19,7 +19,7 @@ ADMIN_PATH = 'admin/'
 OIDCFED_DEFAULT_TRUST_ANCHOR = "http://127.0.0.1:8000"
 OIDCFED_TRUST_ANCHORS = [OIDCFED_DEFAULT_TRUST_ANCHOR]
 OIDCFED_PROVIDER_PROFILE = "cie"
-
+ OIDCFED_PROVIDER_MAX_CONSENT_TIMEFRAME = 3600 #seconds
 OIDCFED_REQUIRED_TRUST_MARKS = []
 
 #OIDCFED_FEDERATION_TRUST_MARKS_PROFILES = {

spid_cie_oidc/provider/views/token_endpoint.py

@@ -1,6 +1,6 @@
-
 import base64
 import hashlib
+import json
 import logging
 
 from djagger.decorators import schema
@@ -23,30 +23,31 @@
     OIDCFED_PROVIDER_PROFILES
 )
 
+from spid_cie_oidc.entity.utils import datetime_from_timestamp, exp_from_now, iat_now
 from . import OpBase
-logger = logging.getLogger(__name__)
 
+logger = logging.getLogger(__name__)
 
 schema_profile = OIDCFED_PROVIDER_PROFILES[OIDCFED_DEFAULT_PROVIDER_PROFILE]
 
 
 @schema(
-    methods=['GET','POST'],
-    post_request_schema = {
+    methods=['GET', 'POST'],
+    post_request_schema={
         "authn_request": schema_profile["authorization_code"],
         "refresh_request": schema_profile["refresh_token"],
 
     },
-    post_response_schema = {
-            "200": schema_profile["authorization_code_response"],
-            # TODO
-            # "200": schema_profile["refresh_token_response"],
-            "400": schema_profile["token_error_response"],
+    post_response_schema={
+        "200": schema_profile["authorization_code_response"],
+        # TODO
+        # "200": schema_profile["refresh_token_response"],
+        "400": schema_profile["token_error_response"],
     },
-    get_response_schema = {
-            "400": BaseModel
+    get_response_schema={
+        "400": BaseModel
     },
-    tags = ['Provider']
+    tags=['Provider']
 )
 @method_decorator(csrf_exempt, name="dispatch")
 class TokenEndpoint(OpBase, View):
@@ -77,35 +78,46 @@ def grant_auth_code(self, request, *args, **kwargs):
         #
 
         issued_token = IssuedToken.objects.filter(
-            session= self.authz,
-            revoked = False
+            session=self.authz,
+            revoked=False
         ).first()
 
         jwk_at = unpad_jwt_payload(issued_token.access_token)
         expires_in = self.get_expires_in(jwk_at['iat'], jwk_at['exp'])
 
-        iss_token_data = dict( # nosec B106
-            access_token = issued_token.access_token,
-            id_token = issued_token.id_token,
-            token_type = "Bearer", # nosec B106
-            expires_in = expires_in,
+        iss_token_data = dict(  # nosec B106
+            access_token=issued_token.access_token,
+            id_token=issued_token.id_token,
+            token_type="Bearer",  # nosec B106
+            expires_in=expires_in,
             # TODO: remove unsupported scope
-            scope = self.authz.authz_request["scope"],
+            scope=self.authz.authz_request["scope"],
         )
         if issued_token.refresh_token:
             iss_token_data['refresh_token'] = issued_token.refresh_token
         return JsonResponse(iss_token_data)
 
     def is_token_renewable(self, session) -> bool:
         issuedToken = IssuedToken.objects.filter(
-            session = session
-        )
-        # TODO: check also ACR
-        return (
-            (issuedToken.count() - 1) < getattr(
-                settings, "OIDCFED_PROVIDER_MAX_REFRESH", 1
-            )
-        )
+            session=session
+        ).first()
+
+        id_token = unpad_jwt_payload(issuedToken.id_token)
+
+        consent_expiration = id_token['iat'] + getattr(settings, "OIDCFED_PROVIDER_MAX_CONSENT_TIMEFRAME")
+
+        delta = consent_expiration - iat_now()
+
+        if delta > 0:
+            return True
+        return False
+
+        # # TODO: check also ACR
+        # return (
+        #     (issuedToken.count() - 1) < getattr(
+        #         settings, "OIDCFED_PROVIDER_MAX_REFRESH", 1
+        #     )
+        # )
 
     def grant_refresh_token(self, request, *args, **kwargs):
         """
@@ -119,8 +131,8 @@ def grant_refresh_token(self, request, *args, **kwargs):
         # 2. create a new instance of issuedtoken linked to the same sessions and revoke the older
         # 3. response with a new refresh, access and id_token
         issued_token = IssuedToken.objects.filter(
-            refresh_token = request.POST['refresh_token'],
-            revoked = False
+            refresh_token=request.POST['refresh_token'],
+            revoked=False
         ).first()
 
         if not issued_token:
@@ -130,17 +142,17 @@ def grant_refresh_token(self, request, *args, **kwargs):
                     "error_description": "Refresh token not found",
 
                 },
-                status = 400
+                status=400
             )
 
         session = issued_token.session
-        if not self.is_token_renewable(session): # pragma: no cover
+        if not self.is_token_renewable(session):  # pragma: no cover
             return JsonResponse(
-                    {
-                        "error": "invalid_request",
-                        "error_description": "Refresh Token can no longer be updated",
+                {
+                    "error": "invalid_request",
+                    "error_description": "Refresh Token can no longer be updated",
 
-                    }, status = 400
+                }, status=400
             )
         iss_token_data = self.get_iss_token_data(session, self.get_issuer())
         IssuedToken.objects.create(**iss_token_data)
@@ -150,12 +162,12 @@ def grant_refresh_token(self, request, *args, **kwargs):
         jwk_at = unpad_jwt_payload(iss_token_data['access_token'])
         expires_in = self.get_expires_in(jwk_at['iat'], jwk_at['exp'])
 
-        data = dict( # nosec B106
-            access_token = iss_token_data['access_token'],
-            id_token = iss_token_data['id_token'],
-            refresh_token = iss_token_data['refresh_token'],
-            token_type = "Bearer", # nosec B106
-            expires_in = expires_in,
+        data = dict(  # nosec B106
+            access_token=iss_token_data['access_token'],
+            id_token=iss_token_data['id_token'],
+            refresh_token=iss_token_data['refresh_token'],
+            token_type="Bearer",  # nosec B106
+            expires_in=expires_in,
         )
 
         return JsonResponse(data)
@@ -174,7 +186,7 @@ def post(self, request, *args, **kwargs):
                     "error": "invalid_request",
                     "error_description": "Token request object validation failed",
                 },
-                status = 400
+                status=400
             )
 
         self.commons = self.get_jwt_common_data()
@@ -186,7 +198,7 @@ def post(self, request, *args, **kwargs):
                 request.POST['client_id'],
                 request.POST['client_assertion']
             )
-        except Exception as e: # pragma: no cover
+        except Exception as e:  # pragma: no cover
             logger.warning(
                 "Client authentication failed for "
                 f"{request.POST.get('client_id', 'unknown')}: {e}"
@@ -197,7 +209,7 @@ def post(self, request, *args, **kwargs):
                     'error': "unauthorized_client",
                     'error_description': ""
 
-                }, status = 403
+                }, status=403
             )
 
         if request.POST.get("grant_type") == 'authorization_code':

spid_cie_oidc/relying_party/views/__init__.py

@@ -9,6 +9,8 @@
 from ..oidc import *
 from ..oauth2 import *
 
+from enum import Enum
+
 from spid_cie_oidc.entity.exceptions import InvalidTrustchain
 from spid_cie_oidc.entity.models import TrustChain
 from spid_cie_oidc.entity.trust_chain_operations import get_or_create_trust_chain
@@ -21,6 +23,12 @@
 logger = logging.getLogger(__name__)
 
 
+class TokenRequestType(str, Enum):
+    refresh = "refresh"
+    revocation = "revocation"
+    introspection = "introspection"
+
+
 class SpidCieOidcRp:
     """
     Baseclass with common methods for RPs
@@ -110,16 +118,16 @@ def get_token_request(self, auth_token, request, token_type):
             client_assertion_type="urn:ietf:params:oauth:client-assertion-type:jwt-bearer"
         )
 
-        if token_type == 'refresh': # nosec - B105
+        if token_type == TokenRequestType.refresh: #'refresh':  # nosec - B105
             token_request_data["grant_type"] = "refresh_token"
             token_request_data["refresh_token"] = auth_token.refresh_token
             audience = authz.provider_configuration["token_endpoint"]
 
-        elif token_type == 'revocation': # nosec - B105
+        elif token_type == TokenRequestType.revocation: #'revocation':  # nosec - B105
             token_request_data["token"] = auth_token.access_token
             audience = authz.provider_configuration["revocation_endpoint"]
 
-        elif token_type == 'introspection': # nosec - B105
+        elif token_type == TokenRequestType.introspection: #'introspection':  # nosec - B105
             token_request_data["token"] = auth_token.access_token
             audience = authz.provider_configuration["introspection_endpoint"]
 

spid_cie_oidc/relying_party/views/rp_extend_session.py

@@ -8,7 +8,7 @@
 from ..oauth2 import *
 from ..oidc import *
 
-from . import SpidCieOidcRp
+from . import SpidCieOidcRp, TokenRequestType
 from django.views import View
 
 from spid_cie_oidc.entity.jwtse import (
@@ -49,7 +49,7 @@ def get(self, request, *args, **kwargs):
         auth_token = auth_tokens.last()
 
         try:
-            token_response = self.get_token_request(auth_token, request, "refresh")
+            token_response = self.get_token_request(auth_token, request, TokenRequestType.refresh) #"refresh")
             if token_response.status_code == 400:
                 return HttpResponseRedirect(reverse("spid_cie_rp_landing"))
 

spid_cie_oidc/relying_party/views/rp_initiated_logout.py

@@ -9,7 +9,7 @@
 from django.utils import timezone
 from ..models import OidcAuthenticationToken
 
-from . import SpidCieOidcRp
+from . import SpidCieOidcRp, TokenRequestType
 from django.views import View
 
 logger = logging.getLogger(__name__)
@@ -47,7 +47,7 @@ def get(self, request):
         logout(request)
 
         try:
-            self.get_token_request(auth_token, request, "revocation")
+            self.get_token_request(auth_token, request, TokenRequestType.revocation) #"revocation")
             auth_token.logged_out = timezone.localtime()
             auth_token.save()
         except Exception as e:  # pragma: no cover

spid_cie_oidc/relying_party/views/rp_introspection.py

@@ -7,7 +7,7 @@
 from ..oauth2 import *
 from ..oidc import *
 
-from . import SpidCieOidcRp
+from . import SpidCieOidcRp, TokenRequestType
 from django.views import View
 
 
@@ -48,7 +48,7 @@ def get(self, request, *args, **kwargs):
         auth_token = auth_tokens.last()
 
         try:
-            token_response = self.get_token_request(auth_token, request, "introspection")
+            token_response = self.get_token_request(auth_token, request, TokenRequestType.introspection) # "introspection")
             introspection_token_response = json.loads(token_response.content.decode())
             data = {"introspection": introspection_token_response}
             return render(request, self.template, data)