feat: client assertion payload check

.gitignore

@@ -143,3 +143,4 @@ examples/wallet_trust_anchor/wallet_trust_anchor/settingslocal.py
 
 examples-docker/
 docker-compose-externalrp.yml
+*.old

spid_cie_oidc/__init__.py

@@ -1 +1 @@
-__version__ = "1.2.2"
+__version__ = "1.3.0"

spid_cie_oidc/provider/schemas/client_assertion.py

@@ -0,0 +1,32 @@
+from spid_cie_oidc.entity.utils import iat_now
+
+from pydantic import BaseModel, AnyHttpUrl, constr, validator
+from typing import Literal, Optional, List
+
+
+class ClientAssertion(BaseModel):
+    iss: AnyHttpUrl
+    sub: AnyHttpUrl
+    iat: int
+    exp: int
+    jti: Optional[str]
+    aud: str | List[AnyHttpUrl]
+
+    @validator("sub")
+    def iss_and_sub_must_match(cls, sub, values):
+        if values['iss'] != sub:
+            raise ValueError(
+                'Client Assertion: iss and sub must have the same value'
+            )
+        return sub
+
+    @validator("exp")
+    def not_expired(cls, exp, values):
+        _now = iat_now()
+        if not (values['iat'] <= _now < exp):
+            raise ValueError(
+                'Client Assertion: exp must be greater than '
+                'iat and less than the current time.'
+                f'{values["iat"]} <= {_now} < {exp}'
+            )
+        return exp

spid_cie_oidc/provider/views/__init__.py

@@ -22,6 +22,7 @@
     ValidationException
 )
 from spid_cie_oidc.provider.models import OidcSession
+from spid_cie_oidc.provider.schemas.client_assertion import ClientAssertion
 
 from spid_cie_oidc.provider.settings import (
     OIDCFED_ATTRNAME_I18N,
@@ -198,7 +199,12 @@ def check_client_assertion(self, client_id: str, client_assertion: str) -> bool:
         _op = self.get_issuer()
         _op_eid = _op.sub
         _op_eid_authz_endpoint = [_op.metadata['openid_provider']['authorization_endpoint']]
-
+
+        try:
+            ClientAssertion(**payload)
+        except Exception as e:
+            raise Exception(f"Client Assertion: json schema validation error: {e}")
+
         if isinstance(_aud, str):
             _aud = [_aud]
         _allowed_auds = _aud + _op_eid_authz_endpoint
@@ -208,7 +214,7 @@ def check_client_assertion(self, client_id: str, client_assertion: str) -> bool:
                 f"Client assertion failed: {_sub} != {client_id}"
             )
             # TODO Specialize exceptions
-            raise Exception()
+            raise Exception("Client Assertion: sub != client_id")
 
         if _op_eid:
             _allowed_auds.append(_op_eid)
@@ -219,7 +225,7 @@ def check_client_assertion(self, client_id: str, client_assertion: str) -> bool:
                 f"{self.request.build_absolute_uri()} not in {_allowed_auds}"
             )
             # TODO Specialize exceptions
-            raise Exception()
+            raise Exception("Client Assertion: fake audience")
 
         tc = TrustChain.objects.get(sub=client_id, is_active=True)
         jwk = self.find_jwk(head, tc.metadata['openid_relying_party']['jwks']['keys'])

spid_cie_oidc/provider/views/token_endpoint.py

@@ -105,9 +105,7 @@ def is_token_renewable(self, session) -> bool:
             ).first()
 
             id_token = unpad_jwt_payload(issuedToken.id_token)
-
             consent_expiration = id_token['iat'] + OIDCFED_PROVIDER_MAX_CONSENT_TIMEFRAME
-
             delta = consent_expiration - iat_now()
             if delta > 0:
                 return True
@@ -138,7 +136,7 @@ def grant_refresh_token(self, request, *args, **kwargs):
             refresh_token=request.POST['refresh_token'],
             revoked=False
         ).first()
-
+        
         if not issued_token:
             return JsonResponse(
                 {
@@ -148,7 +146,6 @@ def grant_refresh_token(self, request, *args, **kwargs):
                 },
                 status=400
             )
-
         session = issued_token.session
         if not self.is_token_renewable(session):  # pragma: no cover
             return JsonResponse(
@@ -171,7 +168,7 @@ def grant_refresh_token(self, request, *args, **kwargs):
             id_token=iss_token_data['id_token'],
             refresh_token=iss_token_data['refresh_token'],
             token_type="Bearer",  # nosec B106
-            expires_in=expires_in,
+            expires_in=expires_in
         )
 
         return JsonResponse(data)
@@ -192,10 +189,8 @@ def post(self, request, *args, **kwargs):
                 },
                 status=400
             )
-
         self.commons = self.get_jwt_common_data()
         self.issuer = self.get_issuer()
-
         # check client_assertion and client ownership
         try:
             self.check_client_assertion(

spid_cie_oidc/relying_party/views/__init__.py

@@ -144,7 +144,7 @@ def get_token_request(self, auth_token, request, token_type):
                 "aud": [audience],
                 "iat": iat_now(),
                 "exp": exp_from_now(),
-                "jti": str(uuid.uuid4()),
+                "jti": str(uuid.uuid4())
             },
             jwk_dict=rp_conf.jwks_core[0],
         )