refactor serializers and enums into their own submodules

Author: Guillaume Pujol

requests_oauth2client/serializers.py

@@ -12,15 +12,17 @@
 
 from abc import ABC, abstractmethod
 from datetime import datetime, timezone
-from typing import TYPE_CHECKING, Any, Callable, Generic, TypeVar, override
+from typing import TYPE_CHECKING, Any, Callable, ClassVar, Generic, TypeVar, override
 
 import jwskate
 from attr import asdict, field, frozen
 from binapy import BinaPy
-from jwskate import Jwk
 
-from . import RequestParameterAuthorizationRequest, RequestUriParameterAuthorizationRequest
-from .authorization_request import AuthorizationRequest
+from .authorization_request import (
+    AuthorizationRequest,
+    RequestParameterAuthorizationRequest,
+    RequestUriParameterAuthorizationRequest,
+)
 from .dpop import DPoPKey, DPoPToken
 from .tokens import BearerToken
 
@@ -101,8 +103,8 @@ def get_class(self, args: Mapping[str, Any]) -> type[BearerToken]:
             "dpop": DPoPToken,
         }.get(token_type.lower(), BearerToken)
 
-    @staticmethod
-    def default_dumper(token: BearerToken) -> str:
+    @classmethod
+    def default_dumper(cls, token: BearerToken) -> str:
         """Serialize a token as JSON, then compress with deflate, then encodes as base64url.
 
         Args:
@@ -123,8 +125,10 @@ def default_dumper(token: BearerToken) -> str:
             BinaPy.serialize_to("json", {k: w for k, w in d.items() if w is not None}).to("deflate").to("b64u").ascii()
         )
 
-    @staticmethod
-    def default_loader(serialized: str, get_class: Callable[[Mapping[str, Any]], type[BearerToken]]) -> BearerToken:
+    @classmethod
+    def default_loader(
+        cls, serialized: str, get_class: Callable[[Mapping[str, Any]], type[BearerToken]]
+    ) -> BearerToken:
         """Deserialize a BearerToken.
 
         This does the opposite operations than `default_dumper`.
@@ -151,6 +155,48 @@ def default_loader(serialized: str, get_class: Callable[[Mapping[str, Any]], typ
         return token_class(**args)
 
 
+@frozen
+class DPoPKeySerializer(Serializer[DPoPKey]):
+    """A (de)serializer for `DPoPKey` instances."""
+
+    dumper: Callable[[DPoPKey], str] = field(factory=lambda: DPoPKeySerializer.default_dumper)
+    loader: Callable[[str, Callable[[Mapping[str, Any]], type[DPoPKey]]], DPoPKey] = field(
+        factory=lambda: DPoPKeySerializer.default_loader
+    )
+
+    @override
+    def get_class(self, args: Mapping[str, Any]) -> type[DPoPKey]:
+        return DPoPKey
+
+    @classmethod
+    def default_dumper(cls, dpop_key: DPoPKey) -> str:
+        """Provide a default dumper implementation.
+
+        This will not serialize jti_generator, iat_generator, and dpop_token_class!
+
+        """
+        d = dpop_key.private_key.to_dict()
+        d.pop("jti_generator", None)
+        d.pop("iat_generator", None)
+        d.pop("dpop_token_class", None)
+        return BinaPy.serialize_to("json", d).to("deflate").to("b64u").ascii()
+
+    @classmethod
+    def default_loader(
+        cls,
+        serialized: str,
+        get_class: Callable[[Mapping[str, Any]], type[DPoPKey]],
+    ) -> DPoPKey:
+        """Provide a default deserializer implementation.
+
+        This will not deserialize iat_generator, iat_generator, and dpop_token_class!
+
+        """
+        private_key = BinaPy(serialized).decode_from("b64u").decode_from("deflate").parse_from("json")
+        dpop_class = get_class({})
+        return dpop_class(private_key=private_key)
+
+
 @frozen
 class AuthorizationRequestSerializer(
     Serializer[AuthorizationRequest | RequestParameterAuthorizationRequest | RequestUriParameterAuthorizationRequest]
@@ -180,6 +226,8 @@ class AuthorizationRequestSerializer(
         AuthorizationRequest | RequestParameterAuthorizationRequest | RequestUriParameterAuthorizationRequest,
     ] = field(factory=lambda: AuthorizationRequestSerializer.default_loader)
 
+    dpop_key_serializer: ClassVar[Serializer[DPoPKey]] = DPoPKeySerializer()
+
     @override
     def get_class(
         self, args: Mapping[str, Any]
@@ -190,8 +238,9 @@ def get_class(
             return RequestUriParameterAuthorizationRequest
         return AuthorizationRequest
 
-    @staticmethod
+    @classmethod
     def default_dumper(
+        cls,
         azr: AuthorizationRequest | RequestParameterAuthorizationRequest | RequestUriParameterAuthorizationRequest,
     ) -> str:
         """Provide a default dumper implementation.
@@ -208,12 +257,13 @@ def default_dumper(
         """
         d = asdict(azr)
         if azr.dpop_key:
-            d["dpop_key"]["private_key"] = azr.dpop_key.private_key.to_dict()
+            d["dpop_key"] = cls.dpop_key_serializer.dumps(azr.dpop_key)
         d.update(**d.pop("kwargs", {}))
         return BinaPy.serialize_to("json", d).to("deflate").to("b64u").ascii()
 
-    @staticmethod
+    @classmethod
     def default_loader(
+        cls,
         serialized: str,
         get_class: Callable[
             [Mapping[str, Any]],
@@ -234,55 +284,9 @@ def default_loader(
         """
         args = BinaPy(serialized).decode_from("b64u").decode_from("deflate").parse_from("json")
 
-        if dpop_key := args.get("dpop_key"):
-            dpop_key["private_key"] = Jwk(dpop_key["private_key"])
-            dpop_key.pop("jti_generator", None)
-            dpop_key.pop("iat_generator", None)
-            dpop_key.pop("dpop_token_class", None)
-            args["dpop_key"] = DPoPKey(**dpop_key)
+        if args["dpop_key"]:
+            args["dpop_key"] = cls.dpop_key_serializer.loads(args["dpop_key"])
 
         azr_class = get_class(args)
 
         return azr_class(**args)
-
-
-@frozen
-class DPoPKeySerializer(Serializer[DPoPKey]):
-    """A (de)serializer for `DPoPKey` instances."""
-
-    dumper: Callable[[DPoPKey], str] = field(factory=lambda: DPoPKeySerializer.default_dumper)
-    loader: Callable[[str, Callable[[Mapping[str, Any]], type[DPoPKey]]], DPoPKey] = field(
-        factory=lambda: DPoPKeySerializer.default_loader
-    )
-
-    @override
-    def get_class(self, args: Mapping[str, Any]) -> type[DPoPKey]:
-        return DPoPKey
-
-    @staticmethod
-    def default_dumper(dpop_key: DPoPKey) -> str:
-        """Provide a default dumper implementation.
-
-        This will not serialize jti_generator, iat_generator, and dpop_token_class!
-
-        """
-        d = dpop_key.private_key.to_dict()
-        d.pop("jti_generator", None)
-        d.pop("iat_generator", None)
-        d.pop("dpop_token_class", None)
-        return BinaPy.serialize_to("json", d).to("deflate").to("b64u").ascii()
-
-    @staticmethod
-    def default_loader(
-        serialized: str,
-        get_class: Callable[[Mapping[str, Any]], type[DPoPKey]],
-    ) -> DPoPKey:
-        """Provide a default deserializer implementation.
-
-        This will not deserialize iat_generator, iat_generator, and dpop_token_class!
-
-        """
-        args = BinaPy(serialized).decode_from("b64u").decode_from("deflate").parse_from("json")
-        args["private_key"] = Jwk(args["private_key"])
-        cls = get_class(args)
-        return cls(**args)

tests/unit_tests/conftest.py

@@ -18,10 +18,10 @@
     ClientSecretJwt,
     ClientSecretPost,
     DPoPKey,
-    DPoPToken,
     OAuth2Client,
     PrivateKeyJwt,
     PublicApp,
+    RequestParameterAuthorizationRequest,
 )
 
 if TYPE_CHECKING:
@@ -52,14 +52,11 @@ def bearer_auth(access_token: str) -> BearerToken:
     return BearerToken(access_token)
 
 
-@pytest.fixture(scope="session")
-def dpop_key() -> DPoPKey:
-    return DPoPKey.generate()
-
-
-@pytest.fixture(scope="session")
-def dpop_token(access_token: str, dpop_key: DPoPKey) -> DPoPToken:
-    return DPoPToken(access_token=access_token, _dpop_key=dpop_key)
+@pytest.fixture(scope="session", params=[None, "ES256"])
+def dpop_key(request: FixtureRequest) -> DPoPKey | None:
+    if request.param is None:
+        return None
+    return DPoPKey.generate(alg=request.param)
 
 
 @pytest.fixture(scope="session")
@@ -391,6 +388,7 @@ def authorization_request(  # noqa: C901
     code_challenge_method: str,
     expected_issuer: str | None,
     auth_request_kwargs: dict[str, Any],
+    dpop_key: DPoPKey,
 ) -> AuthorizationRequest:
     authorization_response_iss_parameter_supported = bool(expected_issuer)
 
@@ -405,6 +403,7 @@ def authorization_request(  # noqa: C901
         code_challenge_method=code_challenge_method,
         authorization_response_iss_parameter_supported=authorization_response_iss_parameter_supported,
         issuer=expected_issuer,
+        dpop_key=dpop_key,
         **auth_request_kwargs,
     )
 
@@ -416,6 +415,7 @@ def authorization_request(  # noqa: C901
     assert azr.redirect_uri == redirect_uri
     assert azr.issuer == expected_issuer
     assert azr.kwargs == auth_request_kwargs
+    assert azr.dpop_key == dpop_key
 
     args = dict(url.args)
     expected_args = dict(
@@ -499,6 +499,9 @@ def authorization_request(  # noqa: C901
             assert generated_code_challenge == code_verifier
             assert azr.code_verifier == code_verifier
 
+    if dpop_key:
+        expected_args["dpop_jkt"] = dpop_key.dpop_jkt
+
     assert args == expected_args
 
     return azr
@@ -535,3 +538,16 @@ def authorization_response(
     assert auth_response.code_verifier == authorization_request.code_verifier
 
     return auth_response
+
+
+@pytest.fixture(scope="session")
+def request_parameter_signing_key() -> Jwk:
+    return Jwk.generate(alg="ES256")
+
+
+@pytest.fixture
+def request_parameter_authorization_request(
+    authorization_request: AuthorizationRequest,
+    request_parameter_signing_key: Jwk,
+) -> RequestParameterAuthorizationRequest:
+    return authorization_request.sign(request_parameter_signing_key)

tests/unit_tests/test_serializers.py

@@ -10,6 +10,8 @@
     BearerTokenSerializer,
     DPoPKey,
     DPoPToken,
+    RequestParameterAuthorizationRequest,
+    RequestUriParameterAuthorizationRequest,
 )
 
 
@@ -33,11 +35,42 @@ def test_token_serializer(token: BearerToken, freezer: FrozenDateTimeFactory) ->
     assert serializer.loads(candidate) == token
 
 
-def test_authorization_request_serializer(authorization_request: AuthorizationRequest) -> None:
+def test_authorization_request_serializer(
+    authorization_request: AuthorizationRequest,
+    request_parameter_authorization_request: RequestParameterAuthorizationRequest,
+) -> None:
     serializer = AuthorizationRequestSerializer()
     serialized = serializer.dumps(authorization_request)
     assert serializer.loads(serialized) == authorization_request
 
+    request_parameter_serialized = serializer.dumps(request_parameter_authorization_request)
+    assert serializer.loads(request_parameter_serialized) == request_parameter_authorization_request
+
+
+@pytest.fixture(
+    scope="module", params=["this_is_a_request_uri", "urn:this:is:a:request_uri", "https://foo.bar/request_uri"]
+)
+def request_uri_authorization_request(
+    authorization_endpoint: str, client_id: str, request: pytest.FixtureRequest
+) -> RequestUriParameterAuthorizationRequest:
+    request_uri = request.param
+    return RequestUriParameterAuthorizationRequest(
+        authorization_endpoint=authorization_endpoint,
+        client_id=client_id,
+        request_uri=request_uri,
+        custom_param="custom_value",
+    )
+
+
+def test_request_uri_authorization_request_serializer(
+    request_uri_authorization_request: RequestUriParameterAuthorizationRequest,
+) -> None:
+    serializer = AuthorizationRequestSerializer()
+    serialized = serializer.dumps(request_uri_authorization_request)
+    deserialized = serializer.loads(serialized)
+    assert isinstance(deserialized, RequestUriParameterAuthorizationRequest)
+    assert deserialized == request_uri_authorization_request
+
 
 def test_authorization_request_serializer_with_dpop_key() -> None:
     dpop_key = DPoPKey.generate()