Apply black to src/satosa

Author: Ric Janus Sapasap

src/satosa/attribute_mapping.py

@@ -14,9 +14,9 @@ def scope(s):
     :param s: string to extract scope from (filtered string in mako template)
     :return: the scope
     """
-    if '@' not in s:
+    if "@" not in s:
         raise ValueError("Unscoped string")
-    (local_part, _, domain_part) = s.partition('@')
+    (local_part, _, domain_part) = s.partition("@")
     return domain_part
 
 
@@ -31,16 +31,22 @@ def __init__(self, internal_attributes):
         :param internal_attributes: A map of how to convert the attributes
         (dict[internal_name, dict[attribute_profile, external_name]])
         """
-        self.separator = "."  # separator for nested attribute values, e.g. address.street_address
-        self.multivalue_separator = ";"  # separates multiple values, e.g. when using templates
+        self.separator = (
+            "."  # separator for nested attribute values, e.g. address.street_address
+        )
+        self.multivalue_separator = (
+            ";"  # separates multiple values, e.g. when using templates
+        )
         self.from_internal_attributes = internal_attributes["attributes"]
         self.template_attributes = internal_attributes.get("template_attributes", None)
 
         self.to_internal_attributes = defaultdict(dict)
         for internal_attribute_name, mappings in self.from_internal_attributes.items():
             for profile, external_attribute_names in mappings.items():
                 for external_attribute_name in external_attribute_names:
-                    self.to_internal_attributes[profile][external_attribute_name] = internal_attribute_name
+                    self.to_internal_attributes[profile][
+                        external_attribute_name
+                    ] = internal_attribute_name
 
     def to_internal_filter(self, attribute_profile, external_attribute_names):
         """
@@ -59,7 +65,11 @@ def to_internal_filter(self, attribute_profile, external_attribute_names):
         try:
             profile_mapping = self.to_internal_attributes[attribute_profile]
         except KeyError:
-            logline = "no attribute mapping found for the given attribute profile {}".format(attribute_profile)
+            logline = (
+                "no attribute mapping found for the given attribute profile {}".format(
+                    attribute_profile
+                )
+            )
             logger.warn(logline)
             # no attributes since the given profile is not configured
             return []
@@ -103,7 +113,9 @@ def to_internal(self, attribute_profile, external_dict):
             )
             if attribute_values:  # Only insert key if it has some values
                 logline = "backend attribute {external} mapped to {internal} ({value})".format(
-                    external=external_attribute_name, internal=internal_attribute_name, value=attribute_values
+                    external=external_attribute_name,
+                    internal=internal_attribute_name,
+                    value=attribute_values,
                 )
                 logger.debug(logline)
                 internal_dict[internal_attribute_name] = attribute_values
@@ -112,7 +124,9 @@ def to_internal(self, attribute_profile, external_dict):
                     external_attribute_name
                 )
                 logger.debug(logline)
-        internal_dict = self._handle_template_attributes(attribute_profile, internal_dict)
+        internal_dict = self._handle_template_attributes(
+            attribute_profile, internal_dict
+        )
         return internal_dict
 
     def _collate_attribute_values_by_priority_order(self, attribute_names, data):
@@ -128,7 +142,11 @@ def _collate_attribute_values_by_priority_order(self, attribute_names, data):
         return result
 
     def _render_attribute_template(self, template, data):
-        t = Template(template, cache_enabled=True, imports=["from satosa.attribute_mapping import scope"])
+        t = Template(
+            template,
+            cache_enabled=True,
+            imports=["from satosa.attribute_mapping import scope"],
+        )
         try:
             return t.render(**data).split(self.multivalue_separator)
         except (NameError, TypeError):
@@ -144,11 +162,19 @@ def _handle_template_attributes(self, attribute_profile, internal_dict):
                 continue
 
             external_attribute_name = mapping[attribute_profile]
-            templates = [t for t in external_attribute_name if "$" in t]  # these looks like templates...
-            template_attribute_values = [self._render_attribute_template(template, internal_dict) for template in
-                                         templates]
-            flattened_attribute_values = list(chain.from_iterable(template_attribute_values))
-            attribute_values = flattened_attribute_values or internal_dict.get(internal_attribute_name, None)
+            templates = [
+                t for t in external_attribute_name if "$" in t
+            ]  # these looks like templates...
+            template_attribute_values = [
+                self._render_attribute_template(template, internal_dict)
+                for template in templates
+            ]
+            flattened_attribute_values = list(
+                chain.from_iterable(template_attribute_values)
+            )
+            attribute_values = flattened_attribute_values or internal_dict.get(
+                internal_attribute_name, None
+            )
             if attribute_values:  # only insert key if it has some values
                 internal_dict[internal_attribute_name] = attribute_values
 
@@ -172,7 +198,9 @@ def _create_nested_attribute_value(self, nested_attribute_names, value):
             return {nested_attribute_names[0]: value}
 
         # keep digging further into the nested attribute names
-        child_dict = self._create_nested_attribute_value(nested_attribute_names[1:], value)
+        child_dict = self._create_nested_attribute_value(
+            nested_attribute_names[1:], value
+        )
         return {nested_attribute_names[0]: child_dict}
 
     def from_internal(self, attribute_profile, internal_dict):
@@ -190,10 +218,14 @@ def from_internal(self, attribute_profile, internal_dict):
         external_dict = {}
         for internal_attribute_name in internal_dict:
             try:
-                attribute_mapping = self.from_internal_attributes[internal_attribute_name]
-            except KeyError:
-                logline = "no attribute mapping found for the internal attribute {}".format(
+                attribute_mapping = self.from_internal_attributes[
                     internal_attribute_name
+                ]
+            except KeyError:
+                logline = (
+                    "no attribute mapping found for the internal attribute {}".format(
+                        internal_attribute_name
+                    )
                 )
                 logger.debug(logline)
                 continue
@@ -206,20 +238,29 @@ def from_internal(self, attribute_profile, internal_dict):
                 logger.debug(logline)
                 continue
 
-            external_attribute_names = self.from_internal_attributes[internal_attribute_name][attribute_profile]
+            external_attribute_names = self.from_internal_attributes[
+                internal_attribute_name
+            ][attribute_profile]
             # select the first attribute name
             external_attribute_name = external_attribute_names[0]
-            logline = "frontend attribute {external} mapped from {internal} ({value})".format(
-                external=external_attribute_name, internal=internal_attribute_name, value=internal_dict[internal_attribute_name]
+            logline = (
+                "frontend attribute {external} mapped from {internal} ({value})".format(
+                    external=external_attribute_name,
+                    internal=internal_attribute_name,
+                    value=internal_dict[internal_attribute_name],
+                )
             )
             logger.debug(logline)
 
             if self.separator in external_attribute_name:
                 nested_attribute_names = external_attribute_name.split(self.separator)
-                nested_dict = self._create_nested_attribute_value(nested_attribute_names[1:],
-                                                                  internal_dict[internal_attribute_name])
+                nested_dict = self._create_nested_attribute_value(
+                    nested_attribute_names[1:], internal_dict[internal_attribute_name]
+                )
                 external_dict[nested_attribute_names[0]] = nested_dict
             else:
-                external_dict[external_attribute_name] = internal_dict[internal_attribute_name]
+                external_dict[external_attribute_name] = internal_dict[
+                    internal_attribute_name
+                ]
 
         return external_dict

src/satosa/backends/apple.py

@@ -1,6 +1,7 @@
 """
 Apple backend module.
 """
+
 import logging
 from .openid_connect import OpenIDConnectBackend, STATE_KEY
 from oic.oauth2.message import Message

src/satosa/backends/bitbucket.py

@@ -1,6 +1,7 @@
 """
 OAuth backend for BitBucket
 """
+
 import json
 import logging
 import requests
@@ -37,10 +38,17 @@ def __init__(self, outgoing, internal_attributes, config, base_url, name):
         :type base_url: str
         :type name: str
         """
-        config.setdefault('response_type', 'code')
-        config['verify_accesstoken_state'] = False
-        super().__init__(outgoing, internal_attributes, config, base_url,
-                         name, 'bitbucket', 'account_id')
+        config.setdefault("response_type", "code")
+        config["verify_accesstoken_state"] = False
+        super().__init__(
+            outgoing,
+            internal_attributes,
+            config,
+            base_url,
+            name,
+            "bitbucket",
+            "account_id",
+        )
 
     def get_request_args(self, get_state=stateID):
         request_args = super().get_request_args(get_state=get_state)
@@ -58,28 +66,36 @@ def get_request_args(self, get_state=stateID):
 
     def auth_info(self, request):
         return AuthenticationInformation(
-            UNSPECIFIED, None,
-            self.config['server_info']['authorization_endpoint'])
+            UNSPECIFIED, None, self.config["server_info"]["authorization_endpoint"]
+        )
 
     def user_information(self, access_token):
-        url = self.config['server_info']['user_endpoint']
+        url = self.config["server_info"]["user_endpoint"]
         email_url = "{}/emails".format(url)
-        headers = {'Authorization': 'Bearer {}'.format(access_token)}
+        headers = {"Authorization": "Bearer {}".format(access_token)}
         resp = requests.get(url, headers=headers)
         data = json.loads(resp.text)
-        if 'email' in self.config['scope']:
+        if "email" in self.config["scope"]:
             resp = requests.get(email_url, headers=headers)
             emails = json.loads(resp.text)
-            data.update({
-                'email': [e for e in [d.get('email')
-                                      for d in emails.get('values')
-                                      if d.get('is_primary')
-                                      ]
-                          ],
-                'email_confirmed': [e for e in [d.get('email')
-                                                for d in emails.get('values')
-                                                if d.get('is_confirmed')
-                                                ]
-                                    ]
-                })
+            data.update(
+                {
+                    "email": [
+                        e
+                        for e in [
+                            d.get("email")
+                            for d in emails.get("values")
+                            if d.get("is_primary")
+                        ]
+                    ],
+                    "email_confirmed": [
+                        e
+                        for e in [
+                            d.get("email")
+                            for d in emails.get("values")
+                            if d.get("is_confirmed")
+                        ]
+                    ],
+                }
+            )
         return data

src/satosa/backends/github.py

@@ -1,6 +1,7 @@
 """
 OAuth backend for LinkedIn
 """
+
 import json
 import logging
 import requests
@@ -39,11 +40,11 @@ def __init__(self, outgoing, internal_attributes, config, base_url, name):
         :type base_url: str
         :type name: str
         """
-        config.setdefault('response_type', 'code')
-        config['verify_accesstoken_state'] = False
+        config.setdefault("response_type", "code")
+        config["verify_accesstoken_state"] = False
         super().__init__(
-            outgoing, internal_attributes, config, base_url, name, 'github',
-            'id')
+            outgoing, internal_attributes, config, base_url, name, "github", "id"
+        )
 
     def start_auth(self, context, internal_request, get_state=stateID):
         """
@@ -58,53 +59,56 @@ def start_auth(self, context, internal_request, get_state=stateID):
         context.state[self.name] = dict(state=oauth_state)
 
         request_args = dict(
-            client_id=self.config['client_config']['client_id'],
+            client_id=self.config["client_config"]["client_id"],
             redirect_uri=self.redirect_url,
             state=oauth_state,
-            allow_signup=self.config.get('allow_signup', False))
-        scope = ' '.join(self.config['scope'])
+            allow_signup=self.config.get("allow_signup", False),
+        )
+        scope = " ".join(self.config["scope"])
         if scope:
-            request_args['scope'] = scope
+            request_args["scope"] = scope
 
-        cis = self.consumer.construct_AuthorizationRequest(
-            request_args=request_args)
+        cis = self.consumer.construct_AuthorizationRequest(request_args=request_args)
         return Redirect(cis.request(self.consumer.authorization_endpoint))
 
     def auth_info(self, requrest):
         return AuthenticationInformation(
-            UNSPECIFIED, None,
-            self.config['server_info']['authorization_endpoint'])
+            UNSPECIFIED, None, self.config["server_info"]["authorization_endpoint"]
+        )
 
     def _authn_response(self, context):
         state_data = context.state[self.name]
         aresp = self.consumer.parse_response(
-            AuthorizationResponse, info=json.dumps(context.request))
+            AuthorizationResponse, info=json.dumps(context.request)
+        )
         self._verify_state(aresp, state_data, context.state)
-        url = self.config['server_info']['token_endpoint']
+        url = self.config["server_info"]["token_endpoint"]
         data = dict(
-            code=aresp['code'],
+            code=aresp["code"],
             redirect_uri=self.redirect_url,
-            client_id=self.config['client_config']['client_id'],
-            client_secret=self.config['client_secret'], )
-        headers = {'Accept': 'application/json'}
+            client_id=self.config["client_config"]["client_id"],
+            client_secret=self.config["client_secret"],
+        )
+        headers = {"Accept": "application/json"}
 
         r = requests.post(url, data=data, headers=headers)
         response = r.json()
-        if self.config.get('verify_accesstoken_state', True):
+        if self.config.get("verify_accesstoken_state", True):
             self._verify_state(response, state_data, context.state)
 
         user_info = self.user_information(response["access_token"])
         auth_info = self.auth_info(context.request)
         internal_response = InternalData(auth_info=auth_info)
         internal_response.attributes = self.converter.to_internal(
-            self.external_type, user_info)
+            self.external_type, user_info
+        )
         internal_response.subject_id = str(user_info[self.user_id_attr])
         return self.auth_callback_func(context, internal_response)
 
     def user_information(self, access_token):
-        url = self.config['server_info']['user_info']
-        headers = {'Authorization': 'token {}'.format(access_token)}
+        url = self.config["server_info"]["user_info"]
+        headers = {"Authorization": "token {}".format(access_token)}
         r = requests.get(url, headers=headers)
         ret = r.json()
-        ret['id'] = str(ret['id'])
+        ret["id"] = str(ret["id"])
         return ret