[Feat/documentation] docs and refactoring for sd_jwt, storage, tools, trust and x509

pyeudiw/jwk/__init__.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import json
 from typing import Union
 
@@ -6,14 +8,14 @@
 from cryptojwt.jwk.jwk import key_from_jwk_dict
 from cryptojwt.jwk.rsa import new_rsa_key
 
-from .exceptions import InvalidKid, KidNotFoundError
+from .exceptions import InvalidKid, KidNotFoundError, InvalidJwk
 
 KEY_TYPES_FUNC = dict(
     EC=new_ec_key,
     RSA=new_rsa_key
 )
 
-class JWK():
+class JWK:
     """
     The class representing a JWK istance
     """
@@ -117,6 +119,34 @@ def as_dict(self) -> dict:
     def __repr__(self):
         # private part!
         return self.as_json()
+
+class RSAJWK(JWK):
+    def __init__(self, key: dict | None = None, hash_func: str = "SHA-256") -> None:
+        super().__init__(key, "RSA", hash_func, None)
+
+class ECJWK(JWK):
+    def __init__(self, key: dict | None = None, hash_func: str = "SHA-256", ec_crv: str = "P-256") -> None:
+        super().__init__(key, "EC", hash_func, ec_crv)
+
+def jwk_form_dict(key: dict, hash_func: str = "SHA-256") -> RSAJWK | ECJWK:
+    """
+    Returns a JWK instance from a dict.
+
+    :param key: a dict that represents the key.
+    :type key: dict
+
+    :returns: a JWK instance.
+    :rtype: JWK
+    """
+    _kty = key.get('kty', None)
+
+    if _kty == None or _kty not in ['EC', 'RSA']:
+        raise InvalidJwk("Invalid JWK")
+    elif _kty == "RSA":
+        return RSAJWK(key, hash_func)
+    else:
+        ec_crv = key.get('crv', "P-256")
+        return ECJWK(key, hash_func, ec_crv)
 
 def find_jwk(kid: str, jwks: list[dict], as_dict: bool=True) -> dict | JWK:
     """

pyeudiw/jwk/exceptions.py

@@ -10,3 +10,6 @@ class InvalidKid(Exception):
 
 class JwkError(Exception):
     pass
+
+class InvalidJwk(Exception):
+    pass

pyeudiw/satosa/backend.py

@@ -41,7 +41,7 @@
 
 from .exceptions import HTTPError
 from .base_http_error_handler import BaseHTTPErrorHandler
-from .base_logger import BaseLogger
+from pyeudiw.tools.base_logger import BaseLogger
 
 class OpenID4VPBackend(BackendModule, BackendTrust, BackendDPoP, BaseHTTPErrorHandler, BaseLogger):
     """
@@ -152,8 +152,8 @@ def start_auth(self, context: Context, internal_request) -> Response:
 
     def pre_request_endpoint(self, context: Context, internal_request, **kwargs) -> Response:
         """
-        This endpoint is called by the frontend before calling the request endpoint.
-        It initializes the session and returns the request_uri to be used by the frontend.
+        This endpoint is called by the User-Agent/Wallet Instance before calling the request endpoint.
+        It initializes the session and returns the request_uri to be used by the User-Agent/Wallet Instance.
 
         :type context: the context of current request
         :param context: the request context
@@ -215,12 +215,12 @@ def pre_request_endpoint(self, context: Context, internal_request, **kwargs) ->
 
     def redirect_endpoint(self, context: Context, *args: tuple) -> Redirect | JsonResponse:
         """
-        This endpoint is called by the frontend after the user has been authenticated.
+        This endpoint is called by the User-Agent/Wallet Instance after the user has been authenticated.
 
         :type context: the context of current request
         :param context: the request context
 
-        :return: a redirect to the frontend, if is in same device flow, or a json response if is in cross device flow.
+        :return: a redirect to the User-Agent/Wallet Instance, if is in same device flow, or a json response if is in cross device flow.
         :rtype: Redirect | JsonResponse
         """
 
@@ -393,7 +393,7 @@ def redirect_endpoint(self, context: Context, *args: tuple) -> Redirect | JsonRe
 
     def request_endpoint(self, context: Context, *args) -> JsonResponse:
         """
-        This endpoint is called by the frontend to retrieve the signed signed Request Object.
+        This endpoint is called by the User-Agent/Wallet Instance to retrieve the signed signed Request Object.
 
         :type context: the context of current request
         :param context: the request context
@@ -479,7 +479,7 @@ def request_endpoint(self, context: Context, *args) -> JsonResponse:
 
     def get_response_endpoint(self, context: Context) -> Response:
         """
-        This endpoint is called by the frontend to retrieve the response of the authentication.
+        This endpoint is called by the User-Agent/Wallet Instance to retrieve the response of the authentication.
 
         :param context: the request context
         :type context: satosa.context.Context
@@ -529,7 +529,7 @@ def get_response_endpoint(self, context: Context) -> Response:
 
     def status_endpoint(self, context: Context) -> JsonResponse:
         """
-        This endpoint is called by the frontend the url to the response endpoint to finalize the process.
+        This endpoint is called by the User-Agent/Wallet Instance the url to the response endpoint to finalize the process.
 
         :param context: the request context
         :type context: satosa.context.Context

pyeudiw/satosa/base_http_error_handler.py

@@ -1,5 +1,5 @@
 from satosa.context import Context
-from .base_logger import BaseLogger
+from pyeudiw.tools.base_logger import BaseLogger
 from .exceptions import EmptyHTTPError
 from pyeudiw.satosa.response import JsonResponse
 

pyeudiw/satosa/dpop.py

@@ -9,7 +9,7 @@
 from satosa.context import Context
 from pydantic import ValidationError
 
-from .base_logger import BaseLogger
+from pyeudiw.tools.base_logger import BaseLogger
 from .base_http_error_handler import BaseHTTPErrorHandler
 
 class BackendDPoP(BaseHTTPErrorHandler, BaseLogger):

pyeudiw/satosa/trust.py

@@ -14,7 +14,7 @@
 from pyeudiw.trust import TrustEvaluationHelper
 from pyeudiw.trust.trust_anchors import update_trust_anchors_ecs
 
-from .base_logger import BaseLogger
+from pyeudiw.tools.base_logger import BaseLogger
 
 class BackendTrust(BaseLogger):
     """

pyeudiw/sd_jwt/__init__.py

@@ -1,4 +1,3 @@
-import cryptojwt
 import json
 
 from jwcrypto.common import base64url_encode
@@ -23,12 +22,45 @@
 import jwcrypto
 
 from typing import Any
-
+from cryptojwt.jwk.rsa import RSAKey
+from cryptojwt.jwk.ec import ECKey
+from cryptography.hazmat.backends.openssl.rsa import _RSAPrivateKey
 
 class TrustChainSDJWTIssuer(SDJWTIssuer):
-    def __init__(self, user_claims: Dict, issuer_key, holder_key=None, sign_alg=None, add_decoy_claims: bool = True, serialization_format: str = "compact", additional_headers: dict = {}):
+    """
+    Class for issue SD-JWT of TrustChain.
+    """
+    def __init__(
+            self, 
+            user_claims: Dict[str, Any], 
+            issuer_key: dict, 
+            holder_key: dict | None = None, 
+            sign_alg: str | None = None, 
+            add_decoy_claims: bool = True, 
+            serialization_format: str = "compact", 
+            additional_headers: dict = {}
+        ) -> None:
+        """
+        Crate an istance of TrustChainSDJWTIssuer.
+
+        :param user_claims: the claims of the SD-JWT.
+        :type user_claims: dict
+        :param issuer_key: the issuer key.
+        :type issuer_key: dict
+        :param holder_key: the holder key.
+        :type holder_key: dict | None
+        :param sign_alg: the signing algorithm.
+        :type sign_alg: str | None
+        :param add_decoy_claims: if True add decoy claims.
+        :type add_decoy_claims: bool
+        :param serialization_format: the serialization format.
+        :type serialization_format: str
+        :param additional_headers: additional headers.
+        :type additional_headers: dict        
+        """
+
         self.additional_headers = additional_headers
-        sign_alg = DEFAULT_SIG_KTY_MAP[issuer_key.kty]
+        sign_alg = sign_alg if sign_alg else DEFAULT_SIG_KTY_MAP[issuer_key.kty]
 
         super().__init__(
             user_claims,
@@ -40,6 +72,9 @@ def __init__(self, user_claims: Dict, issuer_key, holder_key=None, sign_alg=None
         )
 
     def _create_signed_jws(self):
+        """
+        Creates the signed JWS.        
+        """
         self.sd_jwt = JWS(payload=dumps(self.sd_jwt_payload))
 
         _protected_headers = {"alg": self._sign_alg}
@@ -67,9 +102,19 @@ def _create_signed_jws(self):
             self.serialized_sd_jwt = dumps(jws_content)
 
 
-def _serialize_key(key, **kwargs):
+def _serialize_key(
+        key: RSAKey | ECKey | JWK | dict, 
+        **kwargs
+    ) -> dict:
+    """
+    Serialize a key into dict.
+
+    :param key: the key to serialize.
+    :type key: RSAKey | ECKey | JWK | dict
 
-    if isinstance(key, cryptojwt.jwk.rsa.RSAKey):
+    :returns: the serialized key into a dict.
+    """
+    if isinstance(key, RSAKey) or isinstance(key, ECKey):
         key = key.serialize()
     elif isinstance(key, JWK):
         key = key.as_dict()
@@ -80,7 +125,19 @@ def _serialize_key(key, **kwargs):
     return key
 
 
-def pk_encode_int(i, bit_size=None):
+def pk_encode_int(i: str, bit_size: int = None) -> str:
+    """
+    Encode an integer as a base64url string with padding.
+
+    :param i: the integer to encode.
+    :type i: str
+    :param bit_size: the bit size of the integer.
+    :type bit_size: int
+
+    :returns: the encoded integer.
+    :rtype: str   
+    """
+
     extend = 0
     if bit_size is not None:
         extend = ((bit_size + 7) // 8) * 2
@@ -93,7 +150,22 @@ def pk_encode_int(i, bit_size=None):
     return base64url_encode(unhexlify(extend * '0' + hexi))
 
 
-def import_pyca_pri_rsa(key, **params):
+def import_pyca_pri_rsa(key: _RSAPrivateKey, **params) -> jwcrypto.jwk.JWK:
+    """
+    Import a private RSA key from a PyCA object.
+
+    :param key: the key to import.
+    :type key: RSAKey | ECKey
+
+    :raises ValueError: if the key is not a PyCA RSAKey object.
+
+    :returns: the imported key.
+    :rtype: RSAKey
+    """
+
+    if not isinstance(key, _RSAPrivateKey):
+        raise ValueError("key must be a ssl RSAPrivateKey object")
+
     pn = key.private_numbers()
     params.update(
         kty='RSA',
@@ -129,7 +201,19 @@ def import_ec(key, **params):
     )
     return jwcrypto.jwk.JWK(**params)
 
-def _adapt_keys(issuer_key: JWK, holder_key: JWK):
+def _adapt_keys(issuer_key: JWK, holder_key: JWK) -> dict:
+    """
+    Adapt the keys to the SD-JWT library.
+
+    :param issuer_key: the issuer key.
+    :type issuer_key: JWK
+    :param holder_key: the holder key.
+    :type holder_key: JWK
+
+    :returns: the adapted keys as a dict.
+    :rtype: dict
+    """
+
     # _iss_key = issuer_key.key.serialize(private=True)
     # _iss_key['key_ops'] = 'sign'
 
@@ -152,11 +236,45 @@ def _adapt_keys(issuer_key: JWK, holder_key: JWK):
     )
 
 
-def load_specification_from_yaml_string(yaml_specification: str):
+def load_specification_from_yaml_string(yaml_specification: str) -> dict:
+    """
+    Load a specification from a yaml string.
+
+    :param yaml_specification: the yaml string.
+    :type yaml_specification: str
+
+    :returns: the specification as a dict.
+    :rtype: dict
+    """
+
     return _yaml_load_specification(StringIO(yaml_specification))
 
 
-def issue_sd_jwt(specification: dict, settings: dict, issuer_key: JWK, holder_key: JWK, trust_chain: list[str] | None = None) -> str:
+def issue_sd_jwt(
+        specification: Dict[str, Any], 
+        settings: dict, 
+        issuer_key: JWK, 
+        holder_key: JWK, 
+        trust_chain: list[str] | None = None
+    ) -> str:
+    """
+    Issue a SD-JWT.
+
+    :param specification: the specification of the SD-JWT.
+    :type specification: Dict[str, Any]
+    :param settings: the settings of the SD-JWT.
+    :type settings: dict
+    :param issuer_key: the issuer key.
+    :type issuer_key: JWK
+    :param holder_key: the holder key.
+    :type holder_key: JWK
+    :param trust_chain: the trust chain.
+    :type trust_chain: list[str] | None
+
+    :returns: the issued SD-JWT.
+    :rtype: str
+    """
+
     claims = {
         "iss": settings["issuer"],
         "iat": iat_now(),
@@ -180,7 +298,23 @@ def issue_sd_jwt(specification: dict, settings: dict, issuer_key: JWK, holder_ke
     return {"jws": sdjwt_at_issuer.serialized_sd_jwt, "issuance": sdjwt_at_issuer.sd_jwt_issuance}
 
 
-def _cb_get_issuer_key(issuer: str, settings: dict, adapted_keys: dict, *args, **kwargs):
+def _cb_get_issuer_key(issuer: str, settings: dict, adapted_keys: dict, *args, **kwargs) -> JWK:
+    """
+    Helper function for get the issuer key.
+
+    :param issuer: the issuer.
+    :type issuer: str
+    :param settings: the settings of SD-JWT.
+    :type settings: dict
+    :param adapted_keys: the adapted keys.
+    :type adapted_keys: dict
+
+    :raises Exception: if the issuer is unknown.
+
+    :returns: the issuer key.
+    :rtype: JWK
+    """
+
     if issuer == settings["issuer"]:
         return adapted_keys["issuer_public_key"]
     else:
@@ -193,6 +327,20 @@ def verify_sd_jwt(
     holder_key: JWK,
     settings: dict = {'key_binding': True}
 ) -> (list | dict | Any):
+    """
+    Verify a SD-JWT.
+
+    :param sd_jwt_presentation: the SD-JWT to verify.
+    :type sd_jwt_presentation: str
+    :param issuer_key: the issuer key.
+    :type issuer_key: JWK
+    :param holder_key: the holder key.
+    :type holder_key: JWK
+    :param settings: the settings of SD-JWT.
+
+    :returns: the verified payload.
+    :rtype: list | dict | Any
+    """
 
     settings.update(
         {