Merge pull request #15 from IdentityPython/develop

Updated _unwrap_identity.

Author: rohe

pyproject.toml

@@ -7,7 +7,7 @@ build-backend = "setuptools.build_meta"
 
 [metadata]
 name = "idpyoidc"
-version = "1.0.10"
+version = "1.0.11"
 author = "Roland Hedberg"
 author_email = "[email protected]"
 description = "Everything OAuth2 and OIDC"

src/idpyoidc/__init__.py

@@ -1,5 +1,5 @@
 __author__ = "Roland Hedberg"
-__version__ = "1.0.10"
+__version__ = "1.0.11"
 
 import os
 from typing import Dict

src/idpyoidc/server/oauth2/authorization.py

@@ -7,7 +7,6 @@
 from urllib.parse import urlencode
 from urllib.parse import urlparse
 
-from cryptojwt import BadSyntax
 from cryptojwt import as_unicode
 from cryptojwt import b64d
 from cryptojwt.jwe.exception import JWEException
@@ -90,10 +89,10 @@ def max_age(request):
 
 
 def verify_uri(
-    endpoint_context: EndpointContext,
-    request: Union[dict, Message],
-    uri_type: str,
-    client_id: Optional[str] = None,
+        endpoint_context: EndpointContext,
+        request: Union[dict, Message],
+        uri_type: str,
+        client_id: Optional[str] = None,
 ):
     """
     A redirect URI
@@ -223,10 +222,10 @@ def get_uri(endpoint_context, request, uri_type):
 
 
 def authn_args_gather(
-    request: Union[AuthorizationRequest, dict],
-    authn_class_ref: str,
-    cinfo: dict,
-    **kwargs,
+        request: Union[AuthorizationRequest, dict],
+        authn_class_ref: str,
+        cinfo: dict,
+        **kwargs,
 ):
     """
     Gather information to be used by the authentication method
@@ -522,27 +521,33 @@ def _login_required_error(self, redirect_uri, request):
         return _res
 
     def _unwrap_identity(self, identity):
+        # identity is a dict or a json object
+        # the value of 'uid' in the dictionary might be a base64 encoded (b64e) json object
         if isinstance(identity, dict):
+            _uid = as_unicode(identity['uid'])
             try:
-                _id = b64d(as_bytes(identity["uid"]))
-            except BadSyntax:
+                _id = b64d(as_bytes(_uid))
+            except Exception as err:
                 return identity
         else:
             try:
                 _id = b64d(as_bytes(identity))
-            except BadSyntax:
+            except Exception as err:
                 return identity
 
-        return json.loads(as_unicode(_id))
+        try:
+            return json.loads(as_unicode(_id))
+        except UnicodeDecodeError:
+            return identity
 
     def setup_auth(
-        self,
-        request: Optional[Union[Message, dict]],
-        redirect_uri: str,
-        cinfo: dict,
-        cookie: List[dict] = None,
-        acr: str = None,
-        **kwargs,
+            self,
+            request: Optional[Union[Message, dict]],
+            redirect_uri: str,
+            cinfo: dict,
+            cookie: List[dict] = None,
+            acr: str = None,
+            **kwargs,
     ) -> dict:
         """
 
@@ -665,12 +670,12 @@ def aresp_check(self, aresp, request):
         return ""
 
     def response_mode(
-        self,
-        request: Union[dict, AuthorizationRequest],
-        response_args: Optional[Union[dict, AuthorizationResponse]] = None,
-        return_uri: Optional[str] = "",
-        fragment_enc: Optional[bool] = None,
-        **kwargs,
+            self,
+            request: Union[dict, AuthorizationRequest],
+            response_args: Optional[Union[dict, AuthorizationResponse]] = None,
+            return_uri: Optional[str] = "",
+            fragment_enc: Optional[bool] = None,
+            **kwargs,
     ) -> dict:
         resp_mode = request["response_mode"]
         if resp_mode == "form_post":
@@ -969,10 +974,10 @@ def do_request_user(self, request_info, **kwargs):
         return kwargs
 
     def process_request(
-        self,
-        request: Optional[Union[Message, dict]] = None,
-        http_info: Optional[dict] = None,
-        **kwargs,
+            self,
+            request: Optional[Union[Message, dict]] = None,
+            http_info: Optional[dict] = None,
+            **kwargs,
     ):
         """The AuthorizationRequest endpoint
 

tests/test_server_24_oauth2_authorization_endpoint.py

@@ -692,6 +692,17 @@ def test_req_user_no_prompt(self):
         res = self.endpoint.setup_auth(request, redirect_uri, cinfo, None, req_user="adam")
         assert "error" in res
 
+    def test_unwrap_identity(self):
+        identity = {
+            'sid':
+                'Z0FBQUFBQmlZQXFBeDlvSjRENVVYSDBFeTZ6YzVQWTRGVy1laFk2ZmJIbWdPeUhzbVJYbWo5clVPQ045MXpiSVYwS0pfZkREaVUwX2VaVU9HMk9hUktxaGR0R0dQMlRLOXVWQWVTYWJMdDFsVWZJUEItWS1NVi1WQXllNEVlYm9KMDJsSmFYU0pLYWVJeVRKZkJCYmE1T2RpWXRPM3ZmanRlMThfLUNvcnd4ZXVxcFBWdDY0M18tbXNzbjFvbGl4OFdJRTF6YTcwQ3dqNjdsRHdUa1V4ZTlZMjU3SVlXaXdSSTVJSFJJNENwand3a2pOdmV2WGFPRGZhSnZma2NkZ01ZZk1iS3hma1phcQ==',
+            'state': '80ec120d9a322e70e02503e9a99e734174c1e6cb',
+            'timestamp': 1650461312,
+            'uid': '6260077f56d8970e543aa380',
+            'grant_id': 'c636b820c0ad11ecbdd1acde48001122'}
+        _id = self.endpoint._unwrap_identity(identity)
+        assert _id["uid"] == '6260077f56d8970e543aa380'
+
     # def test_sso(self):
     #     _pr_resp = self.endpoint.parse_request(AUTH_REQ_DICT)
     #     _resp = self.endpoint.process_request(_pr_resp)