Merge pull request #255 from italia/dev

Userinfo encryption also with EC

spid_cie_oidc/__init__.py

@@ -1 +1 @@
-__version__ = "0.8.13"
+__version__ = "0.8.14"

spid_cie_oidc/entity/jwtse.py

@@ -3,8 +3,10 @@
 import json
 import logging
 
+import cryptojwt
 from cryptojwt.exception import UnsupportedAlgorithm, VerificationError
 from cryptojwt.jwe.jwe import factory
+from cryptojwt.jwe.jwe_ec import JWE_EC
 from cryptojwt.jwe.jwe_rsa import JWE_RSA
 from cryptojwt.jwk.jwk import key_from_jwk_dict
 from cryptojwt.jws.jws import JWS
@@ -39,14 +41,21 @@ def unpad_jwt_payload(jwt: str) -> dict:
 def create_jwe(plain_dict: Union[dict, None], jwk_dict: dict, **kwargs) -> str:
     logger.debug(f"Encrypting dict as JWE: " f"{plain_dict}")
     _key = key_from_jwk_dict(jwk_dict)
-    _rsa = JWE_RSA(
+
+    if isinstance(_key, cryptojwt.jwk.rsa.RSAKey):
+        JWE_CLASS = JWE_RSA
+    elif isinstance(_key, cryptojwt.jwk.ec.ECKey):
+        JWE_CLASS = JWE_EC
+
+    _keyobj = JWE_CLASS(
         json.dumps(plain_dict).encode(),
         alg=DEFAULT_JWE_ALG,
         enc=DEFAULT_JWE_ENC,
         kid=_key.kid,
         **kwargs
     )
-    jwe = _rsa.encrypt(_key.public_key())
+
+    jwe = _keyobj.encrypt(_key.public_key())
     logger.debug(f"Encrypted dict as JWE: {jwe}")
     return jwe
 

spid_cie_oidc/entity/statements.py

@@ -38,7 +38,7 @@ def get_http_url(urls: list, httpc_params: dict = {}) -> list:
     if getattr(settings, "HTTP_CLIENT_SYNC", False):
         responses = []
         for i in urls:
-            res = requests.get(i, **httpc_params)
+            res = requests.get(i, **httpc_params) # nosec - B113
             responses.append(res.content.decode())
     else:
         responses = asyncio.run(http_get(urls, httpc_params)) # pragma: no cover

spid_cie_oidc/provider/views/consent_page_view.py

@@ -1,6 +1,7 @@
 import logging
 from django.core.paginator import Paginator
 import urllib.parse
+from urllib.parse import urlparse
 
 from djagger.decorators import schema
 from django.contrib.auth import logout
@@ -95,7 +96,7 @@ def post(self, request, *args, **kwargs):
 
 
 def oidc_provider_not_consent(request):
-    redirect_uri = request.GET.get("redirect_uri")
+    redirect_uri = urlparse(request.GET.get("redirect_uri"))
     state = request.GET.get("state", "")
     logout(request)
     kwargs = dict(
@@ -105,7 +106,7 @@ def oidc_provider_not_consent(request):
         ),
         state = state
     )
-    url = f'{redirect_uri}?{urllib.parse.urlencode(kwargs)}'
+    url = f'{redirect_uri.path if redirect_uri.path else "/"}?{urllib.parse.urlencode(kwargs)}'
     return HttpResponseRedirect(url)
 
 

spid_cie_oidc/relying_party/oidc/__init__.py

@@ -34,7 +34,7 @@ def get_userinfo(
             verify=verify,
             timeout=getattr(
                 settings, "HTTPC_TIMEOUT", 8
-            )
+            ) # nosec - B113
         )
 
         if authz_userinfo.status_code != 200: # pragma: no cover

spid_cie_oidc/relying_party/views/rp_initiated_logout.py

@@ -87,7 +87,13 @@ def oidc_rpinitiated_logout(request):
             client_assertion_type = "urn:ietf:params:oauth:client-assertion-type:jwt-bearer"
         )
         try:
-            requests.post(revocation_endpoint_url, data = revocation_request)
+            requests.post(
+                revocation_endpoint_url, 
+                data = revocation_request,
+                timeout=getattr(
+                    settings, "HTTPC_TIMEOUT", 8
+                )
+            ) # nosec - B113
         except Exception as e: # pragma: no cover
             logger.warning(f"Token revocation failed: {e}")
         auth_tokens.update(revoked = timezone.localtime())