feat: WIA Pydantic validation (#144)

* fix: WIA validation updated Rename `attested_security_context` into `aal`. Make wallet capabilities fields `Optional`. * chore: linting * fix: logging * refactor: validation refactored --------- Co-authored-by: Salvatore Laiso <[email protected]>
italia · Nov 7, 2023 · 77b5efd · 77b5efd
1 parent 799138c
commit 77b5efd
Show file tree

Hide file tree

Showing 6 changed files with 54 additions and 15 deletions.
diff --git a/pyeudiw/oauth2/dpop/__init__.py b/pyeudiw/oauth2/dpop/__init__.py
@@ -3,6 +3,7 @@
 import logging
 import uuid
 
+from pyeudiw.jwk.schema import JwkSchema
 from pyeudiw.oauth2.dpop.exceptions import (
     InvalidDPoP,
     InvalidDPoPAth,
@@ -61,6 +62,14 @@ def __init__(
             if self.dpop_header_prefix in http_header_authz
             else http_header_authz
         )
+        # If the jwk is invalid, raise an exception
+        try:
+            JwkSchema(**public_jwk)
+        except Exception as e:
+            logger.error(
+                "Jwk validation error, "
+                f"{e.__class__.__name__}: {e}"
+            )
         # If the jwt is invalid, this will raise an exception
         try:
             unpad_jwt_header(http_header_dpop)

diff --git a/pyeudiw/openid4vp/schemas/wallet_instance_attestation.py b/pyeudiw/openid4vp/schemas/wallet_instance_attestation.py
@@ -28,7 +28,6 @@ class WalletInstanceAttestationHeader(BaseModel):
     alg: str
     typ: Literal["wallet-attestation+jwt"]
     kid: str
-    x5c: Optional[List[str]] = None
     trust_chain: Optional[List[str]] = None
 
     @field_validator("alg")
@@ -42,14 +41,15 @@ class WalletInstanceAttestationPayload(BaseModel):
     sub: str
     iat: int
     exp: int
-    type: Literal["WalletInstanceAttestation"]
-    policy_uri: HttpUrl
-    tos_uri: HttpUrl
-    logo_uri: HttpUrl
-    attested_security_context: HttpUrl
+    aal: HttpUrl
     cnf: CNFSchema
-    authorization_endpoint: str
-    response_types_supported: List[str]
-    vp_formats_supported: VPFormatSchema
-    request_object_signing_alg_values_supported: List[str]
-    presentation_definition_uri_supported: bool
+    # Wallet Capabilities
+    type: Optional[Literal["WalletInstanceAttestation"]]
+    policy_uri: Optional[HttpUrl]
+    tos_uri: Optional[HttpUrl]
+    logo_uri: Optional[HttpUrl]
+    authorization_endpoint: Optional[str]
+    response_types_supported: Optional[List[str]]
+    vp_formats_supported: Optional[VPFormatSchema]
+    request_object_signing_alg_values_supported: Optional[List[str]]
+    presentation_definition_uri_supported: Optional[bool]
diff --git a/pyeudiw/satosa/dpop.py b/pyeudiw/satosa/dpop.py
@@ -5,8 +5,12 @@
 
 from pyeudiw.jwt.utils import unpad_jwt_header, unpad_jwt_payload
 from pyeudiw.oauth2.dpop import DPoPVerifier
+from pyeudiw.openid4vp.schemas.wallet_instance_attestation import WalletInstanceAttestationPayload, \
+    WalletInstanceAttestationHeader
 from pyeudiw.satosa.response import JsonResponse
 
+import satosa.logging_util as lu
+from satosa.context import Context
 
 logger = logging.getLogger(__name__)
 
@@ -32,6 +36,24 @@ def _request_endpoint_dpop(self, context, *args) -> Union[JsonResponse, None]:
                 )
             )
 
+            try:
+                WalletInstanceAttestationHeader(**_head)
+            except Exception as e:
+                self._log(
+                    context,
+                    level='warning',
+                    message=f"[FOUND WIA] Invalid Headers: {_head}! \nValidation error: {e}"
+                )
+
+            try:
+                WalletInstanceAttestationPayload(**wia)
+            except Exception as e:
+                self._log(
+                    context,
+                    level='warning',
+                    message=f"[FOUND WIA] Invalid WIA: {wia}! \nValidation error: {e}"
+                )
+
             try:
                 self._validate_trust(context, dpop_jws)
             except Exception as e:
@@ -44,7 +66,6 @@ def _request_endpoint_dpop(self, context, *args) -> Union[JsonResponse, None]:
                     err=f"{e}"
                 )
 
-            # TODO: validate wia scheme using pydantic
             try:
                 dpop = DPoPVerifier(
                     public_jwk=wia['cnf']['jwk'],
@@ -82,3 +103,12 @@ def _request_endpoint_dpop(self, context, *args) -> Union[JsonResponse, None]:
                 "a default set of capabilities and a low security level are applied."
             )
             self._log(context, level='warning', message=_msg)
+
+    def _log(self, context: Context, level: str, message: str) -> None:
+        log_level = getattr(logger, level)
+        log_level(
+            lu.LOG_FMT.format(
+                id=lu.get_session_id(context.state),
+                message=message
+            )
+        )
diff --git a/pyeudiw/tests/oauth2/test_dpop.py b/pyeudiw/tests/oauth2/test_dpop.py
@@ -20,7 +20,7 @@
     "policy_uri": "https://wallet-provider.example.org/privacy_policy",
     "tos_uri": "https://wallet-provider.example.org/info_policy",
     "logo_uri": "https://wallet-provider.example.org/logo.svg",
-    "asc": "https://wallet-provider.example.org/LoA/basic",
+    "aal": "https://wallet-provider.example.org/LoA/basic",
     "cnf":
     {
         "jwk": PUBLIC_JWK

diff --git a/pyeudiw/tests/openid4vp/schemas/test_wallet_instance_attestation.py b/pyeudiw/tests/openid4vp/schemas/test_wallet_instance_attestation.py
@@ -23,7 +23,7 @@
         "policy_uri": "https://wallet-provider.example.org/privacy_policy",
         "tos_uri": "https://wallet-provider.example.org/info_policy",
         "logo_uri": "https://wallet-provider.example.org/logo.svg",
-        "attested_security_context": "https://wallet-provider.example.org/LoA/basic",
+        "aal": "https://wallet-provider.example.org/LoA/basic",
         "cnf": {
             "jwk": {
                 "alg": "RS256",

diff --git a/pyeudiw/tests/settings.py b/pyeudiw/tests/settings.py
@@ -350,7 +350,7 @@
     "policy_uri": "https://wallet-provider.example.org/privacy_policy",
     "tos_uri": "https://wallet-provider.example.org/info_policy",
     "logo_uri": "https://wallet-provider.example.org/logo.svg",
-    "asc": "https://wallet-provider.example.org/LoA/basic",
+    "aal": "https://wallet-provider.example.org/LoA/basic",
     "cnf":
     {
         "jwk": PUBLIC_JWK