Fix formatting

Signed-off-by: Ivan Kanakarakis <[email protected]>

Author: c00kiemon5ter

src/saml2/client_base.py

@@ -171,19 +171,17 @@ def __init__(self, config=None, identity_cache=None, state_cache=None,
             "authn_requests_signed": False,
             "want_assertions_signed": False,
             "want_response_signed": True,
-            "want_assertions_or_response_signed" : False
+            "want_assertions_or_response_signed": False,
         }
-
         for attr, val_default in attribute_defaults.items():
             val_config = self.config.getattr(attr, "sp")
-            if val_config is None:
-                val = val_default
-            else:
-                val = val_config
-
+            val = (
+                val_config
+                if val_config is not None
+                else val_default
+            )
             if val == 'true':
                 val = True
-
             setattr(self, attr, val)
 
         # signing and digest algs
@@ -238,8 +236,7 @@ def _sso_location(self, entityid=None, binding=BINDING_HTTP_REDIRECT):
             raise IdpUnspecified("Too many IdPs to choose from: %s" % eids)
 
         try:
-            srvs = self.metadata.single_sign_on_service(list(eids.keys())[0], 
-                                                        binding)
+            srvs = self.metadata.single_sign_on_service(list(eids.keys())[0], binding)
             return next(locations(srvs), None)
 
         except IndexError:

src/saml2/entity.py

@@ -800,8 +800,9 @@ def _response(self, in_response_to, consumer_url=None, status=None,
                 return response
 
         if sign:
-            return self.sign(response, to_sign=to_sign, sign_alg=sign_alg,
-                             digest_alg=digest_alg)
+            return self.sign(
+                response, to_sign=to_sign, sign_alg=sign_alg, digest_alg=digest_alg
+            )
         else:
             return response
 
@@ -835,8 +836,7 @@ def _status_response(self, response_class, issuer, status, sign=False,
                                   status=status, **kwargs)
 
         if sign:
-            return self.sign(response, mid, sign_alg=sign_alg,
-                             digest_alg=digest_alg)
+            return self.sign(response, mid, sign_alg=sign_alg, digest_alg=digest_alg)
         else:
             return response
 
@@ -1121,9 +1121,15 @@ def create_manage_name_id_response(self, request, bindings=None,
 
         rinfo = self.response_args(request, bindings)
 
-        response = self._status_response(samlp.ManageNameIDResponse, issuer,
-                                         status, sign, sign_alg=sign_alg,
-                                         digest_alg=digest_alg, **rinfo)
+        response = self._status_response(
+            samlp.ManageNameIDResponse,
+            issuer,
+            status,
+            sign,
+            sign_alg=sign_alg,
+            digest_alg=digest_alg,
+            **rinfo,
+        )
 
         logger.info("Response: %s", response)
 

src/saml2/server.py

@@ -384,17 +384,32 @@ def setup_assertion(self, authn, sp_entity_id, in_response_to, consumer_url,
                 **kwargs)
         return assertion
 
-    def _authn_response(self, in_response_to, consumer_url,
-                        sp_entity_id, identity=None, name_id=None,
-                        status=None, authn=None, issuer=None, policy=None,
-                        sign_assertion=False, sign_response=False,
-                        best_effort=False, encrypt_assertion=False,
-                        encrypt_cert_advice=None, encrypt_cert_assertion=None,
-                        authn_statement=None,
-                        encrypt_assertion_self_contained=False,
-                        encrypted_advice_attributes=False,
-                        pefim=False, sign_alg=None, digest_alg=None,
-                        farg=None, session_not_on_or_after=None):
+    def _authn_response(
+        self,
+        in_response_to,
+        consumer_url,
+        sp_entity_id,
+        identity=None,
+        name_id=None,
+        status=None,
+        authn=None,
+        issuer=None,
+        policy=None,
+        sign_assertion=False,
+        sign_response=False,
+        best_effort=False,
+        encrypt_assertion=False,
+        encrypt_cert_advice=None,
+        encrypt_cert_assertion=None,
+        authn_statement=None,
+        encrypt_assertion_self_contained=False,
+        encrypted_advice_attributes=False,
+        pefim=False,
+        sign_alg=None,
+        digest_alg=None,
+        farg=None,
+        session_not_on_or_after=None,
+    ):
         """ Create a response. A layer of indirection.
 
         :param in_response_to: The session identifier of the request
@@ -524,8 +539,7 @@ def create_attribute_response(self, identity, in_response_to, destination,
 
         if not name_id and userid:
             try:
-                name_id = self.ident.construct_nameid(userid, policy,
-                                                      sp_entity_id)
+                name_id = self.ident.construct_nameid(userid, policy, sp_entity_id)
                 logger.warning("Unspecified NameID format")
             except Exception:
                 pass
@@ -565,56 +579,53 @@ def create_attribute_response(self, identity, in_response_to, destination,
         if sp_entity_id:
             kwargs['sp_entity_id'] = sp_entity_id
 
-        return self._response(in_response_to, destination, status, issuer,
-                              sign_response, to_sign, sign_alg=sign_alg,
-                              digest_alg=digest_alg, **kwargs)
+        return self._response(
+            in_response_to,
+            destination,
+            status,
+            issuer,
+            sign_response,
+            to_sign,
+            sign_alg=sign_alg,
+            digest_alg=digest_alg,
+            **kwargs,
+        )
+
+    def gather_authn_response_args(
+        self, sp_entity_id, name_id_policy, userid, **kwargs
+    ):
+        # collect args and return them
+        args = {}
 
-    # ------------------------------------------------------------------------
+        args["policy"] = kwargs.get(
+            "release_policy", self.config.getattr("policy", "idp")
+        )
+        args['best_effort'] = kwargs.get("best_effort", False)
 
-    def gather_authn_response_args(self, sp_entity_id, name_id_policy, userid,
-                                   **kwargs):
-        param_default = {
+        param_defaults = {
             'sign_assertion': False,
             'sign_response': False,
             'encrypt_assertion': False,
             'encrypt_assertion_self_contained': True,
             'encrypted_advice_attributes': False,
             'encrypt_cert_advice': None,
-            'encrypt_cert_assertion': None
+            'encrypt_cert_assertion': None,
         }
 
-        args = {}
-
-        try:
-            args["policy"] = kwargs["release_policy"]
-        except KeyError:
-            args["policy"] = self.config.getattr("policy", "idp")
-
-        try:
-            args['best_effort'] = kwargs["best_effort"]
-        except KeyError:
-            args['best_effort'] = False
-        
         # signing and digest algs
         self.signing_algorithm = self.config.getattr('signing_algorithm', "idp")
         self.digest_algorithm = self.config.getattr('digest_algorithm', "idp")
-        
-        for param in ['sign_assertion', 'sign_response', 'encrypt_assertion',
-                      'encrypt_assertion_self_contained',
-                      'encrypted_advice_attributes', 'encrypt_cert_advice',
-                      'encrypt_cert_assertion']:
-            try:
-                _val = kwargs[param]
-            except KeyError:
-                _val = None
-
-            if _val is None:
-                _val = self.config.getattr(param, "idp")
 
-            if _val is None:
-                args[param] = param_default[param]
-            else:
-                args[param] = _val
+        for param, val_default in param_defaults.items():
+            val_kw = kwargs.get(param)
+            val_config = self.config.getattr(param, "idp")
+            args[param] = (
+                val_kw
+                if val_kw is not None
+                else val_config
+                if val_config is not None
+                else val_default
+            )
 
         for arg, attr, eca, pefim in [
             ('encrypted_advice_attributes', 'verify_encrypt_cert_advice',
@@ -698,7 +709,7 @@ def create_authn_response(
         sign_alg=None,
         digest_alg=None,
         session_not_on_or_after=None,
-        **kwargs
+        **kwargs,
     ):
         """ Constructs an AuthenticationResponse
 
@@ -733,21 +744,24 @@ def create_authn_response(
 
         try:
             args = self.gather_authn_response_args(
-                sp_entity_id, name_id_policy=name_id_policy, userid=userid,
-                name_id=name_id, sign_response=sign_response,
+                sp_entity_id,
+                name_id_policy=name_id_policy,
+                userid=userid,
+                name_id=name_id,
+                sign_response=sign_response,
                 sign_assertion=sign_assertion,
                 encrypt_cert_advice=encrypt_cert_advice,
                 encrypt_cert_assertion=encrypt_cert_assertion,
                 encrypt_assertion=encrypt_assertion,
-                encrypt_assertion_self_contained
-                =encrypt_assertion_self_contained,
+                encrypt_assertion_self_contained=encrypt_assertion_self_contained,
                 encrypted_advice_attributes=encrypted_advice_attributes,
-                pefim=pefim, **kwargs)
+                pefim=pefim,
+                **kwargs,
+            )
         except IOError as exc:
-            response = self.create_error_response(in_response_to,
-                                                  destination,
-                                                  sp_entity_id,
-                                                  exc, name_id)
+            response = self.create_error_response(
+                in_response_to, destination, sp_entity_id, exc, name_id
+            )
             return ("%s" % response).split("\n")
 
         try: