Merge pull request #427 from melanger/patch-6

reuse the generic OpenID-Connect backend for the Apple backend

Author: c00kiemon5ter

src/satosa/attribute_mapping.py

@@ -1,6 +1,7 @@
 import logging
 from collections import defaultdict
 from itertools import chain
+from typing import Mapping
 
 from mako.template import Template
 
@@ -97,8 +98,9 @@ def to_internal(self, attribute_profile, external_dict):
                 continue
 
             external_attribute_name = mapping[attribute_profile]
-            attribute_values = self._collate_attribute_values_by_priority_order(external_attribute_name,
-                                                                                external_dict)
+            attribute_values = self._collate_attribute_values_by_priority_order(
+                external_attribute_name, 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
@@ -157,6 +159,8 @@ def _get_nested_attribute_value(self, nested_key, data):
 
         d = data
         for key in keys:
+            if not isinstance(d, Mapping):
+                return None
             d = d.get(key)
             if d is None:
                 return None

src/satosa/backends/apple.py

@@ -2,135 +2,22 @@
 Apple backend module.
 """
 import logging
-from datetime import datetime
-from urllib.parse import urlparse
-
+from .openid_connect import OpenIDConnectBackend, STATE_KEY
 from oic.oauth2.message import Message
-from oic import oic
-from oic import rndstr
 from oic.oic.message import AuthorizationResponse
-from oic.oic.message import ProviderConfigurationResponse
-from oic.oic.message import RegistrationRequest
-from oic.utils.authn.authn_context import UNSPECIFIED
-from oic.utils.authn.client import CLIENT_AUTHN_METHOD
-
 import satosa.logging_util as lu
-from satosa.internal import AuthenticationInformation
-from satosa.internal import InternalData
-from .base import BackendModule
-from .oauth import get_metadata_desc_for_oauth_backend
-from ..exception import SATOSAAuthenticationError, SATOSAError
-from ..response import Redirect
-
+from ..exception import SATOSAAuthenticationError
 import json
 import requests
 
 
 logger = logging.getLogger(__name__)
 
-NONCE_KEY = "oidc_nonce"
-STATE_KEY = "oidc_state"
 
 # https://developer.okta.com/blog/2019/06/04/what-the-heck-is-sign-in-with-apple
-class AppleBackend(BackendModule):
+class AppleBackend(OpenIDConnectBackend):
     """Sign in with Apple backend"""
 
-    def __init__(self, auth_callback_func, internal_attributes, config, base_url, name):
-        """
-        Sign in with Apple backend module.
-        :param auth_callback_func: Callback should be called by the module after the authorization
-        in the backend is done.
-        :param internal_attributes: Mapping dictionary between SATOSA internal attribute names and
-        the names returned by underlying IdP's/OP's as well as what attributes the calling SP's and
-        RP's expects namevice.
-        :param config: Configuration parameters for the module.
-        :param base_url: base url of the service
-        :param name: name of the plugin
-
-        :type auth_callback_func:
-        (satosa.context.Context, satosa.internal.InternalData) -> satosa.response.Response
-        :type internal_attributes: dict[string, dict[str, str | list[str]]]
-        :type config: dict[str, dict[str, str] | list[str]]
-        :type base_url: str
-        :type name: str
-        """
-        super().__init__(auth_callback_func, internal_attributes, base_url, name)
-        self.auth_callback_func = auth_callback_func
-        self.config = config
-        self.client = _create_client(
-            config["provider_metadata"],
-            config["client"]["client_metadata"],
-            config["client"].get("verify_ssl", True),
-        )
-        if "scope" not in config["client"]["auth_req_params"]:
-            config["auth_req_params"]["scope"] = "openid"
-        if "response_type" not in config["client"]["auth_req_params"]:
-            config["auth_req_params"]["response_type"] = "code"
-
-    def start_auth(self, context, request_info):
-        """
-        See super class method satosa.backends.base#start_auth
-        :type context: satosa.context.Context
-        :type request_info: satosa.internal.InternalData
-        """
-        oidc_nonce = rndstr()
-        oidc_state = rndstr()
-        state_data = {NONCE_KEY: oidc_nonce, STATE_KEY: oidc_state}
-        context.state[self.name] = state_data
-
-        args = {
-            "scope": self.config["client"]["auth_req_params"]["scope"],
-            "response_type": self.config["client"]["auth_req_params"]["response_type"],
-            "client_id": self.client.client_id,
-            "redirect_uri": self.client.registration_response["redirect_uris"][0],
-            "state": oidc_state,
-            "nonce": oidc_nonce,
-        }
-        args.update(self.config["client"]["auth_req_params"])
-        auth_req = self.client.construct_AuthorizationRequest(request_args=args)
-        login_url = auth_req.request(self.client.authorization_endpoint)
-        return Redirect(login_url)
-
-    def register_endpoints(self):
-        """
-        Creates a list of all the endpoints this backend module needs to listen to. In this case
-        it's the authentication response from the underlying OP that is redirected from the OP to
-        the proxy.
-        :rtype: Sequence[(str, Callable[[satosa.context.Context], satosa.response.Response]]
-        :return: A list that can be used to map the request to SATOSA to this endpoint.
-        """
-        url_map = []
-        redirect_path = urlparse(
-            self.config["client"]["client_metadata"]["redirect_uris"][0]
-        ).path
-        if not redirect_path:
-            raise SATOSAError("Missing path in redirect uri")
-
-        url_map.append(("^%s$" % redirect_path.lstrip("/"), self.response_endpoint))
-        return url_map
-
-    def _verify_nonce(self, nonce, context):
-        """
-        Verify the received OIDC 'nonce' from the ID Token.
-        :param nonce: OIDC nonce
-        :type nonce: str
-        :param context: current request context
-        :type context: satosa.context.Context
-        :raise SATOSAAuthenticationError: if the nonce is incorrect
-        """
-        backend_state = context.state[self.name]
-        if nonce != backend_state[NONCE_KEY]:
-            msg = "Missing or invalid nonce in authn response for state: {}".format(
-                backend_state
-            )
-            logline = lu.LOG_FMT.format(
-                id=lu.get_session_id(context.state), message=msg
-            )
-            logger.debug(logline)
-            raise SATOSAAuthenticationError(
-                context.state, "Missing or invalid nonce in authn response"
-            )
-
     def _get_tokens(self, authn_response, context):
         """
         :param authn_response: authentication response from OP
@@ -169,25 +56,6 @@ def _get_tokens(self, authn_response, context):
 
         return authn_response.get("access_token"), authn_response.get("id_token")
 
-    def _check_error_response(self, response, context):
-        """
-        Check if the response is an OAuth error response.
-        :param response: the OIDC response
-        :type response: oic.oic.message
-        :raise SATOSAAuthenticationError: if the response is an OAuth error response
-        """
-        if "error" in response:
-            msg = "{name} error: {error} {description}".format(
-                name=type(response).__name__,
-                error=response["error"],
-                description=response.get("error_description", ""),
-            )
-            logline = lu.LOG_FMT.format(
-                id=lu.get_session_id(context.state), message=msg
-            )
-            logger.debug(logline)
-            raise SATOSAAuthenticationError(context.state, "Access denied")
-
     def response_endpoint(self, context, *args):
         """
         Handles the authentication response from the OP.
@@ -209,8 +77,8 @@ def response_endpoint(self, context, *args):
         # - https://developer.apple.com/documentation/sign_in_with_apple/namei
         try:
             userdata = context.request.get("user", "{}")
-            userinfo = json.load(userdata)
-        except Exception:
+            userinfo = json.loads(userdata)
+        except json.JSONDecodeError:
             userinfo = {}
 
         authn_resp = self.client.parse_response(
@@ -242,78 +110,19 @@ def response_endpoint(self, context, *args):
             raise SATOSAAuthenticationError(context.state, "No user info available.")
 
         all_user_claims = dict(list(userinfo.items()) + list(id_token_claims.items()))
+
+        # convert "string or Boolean" claims to actual booleans
+        for bool_claim_name in ["email_verified", "is_private_email"]:
+            userinfo[bool_claim_name] = (
+                True
+                if userinfo[bool_claim_name] == "true"
+                else False
+            )
+
         msg = "UserInfo: {}".format(all_user_claims)
         logline = lu.LOG_FMT.format(id=lu.get_session_id(context.state), message=msg)
         logger.debug(logline)
         internal_resp = self._translate_response(
             all_user_claims, self.client.authorization_endpoint
         )
         return self.auth_callback_func(context, internal_resp)
-
-    def _translate_response(self, response, issuer):
-        """
-        Translates oidc response to SATOSA internal response.
-        :type response: dict[str, str]
-        :type issuer: str
-        :type subject_type: str
-        :rtype: InternalData
-
-        :param response: Dictioary with attribute name as key.
-        :param issuer: The oidc op that gave the repsonse.
-        :param subject_type: public or pairwise according to oidc standard.
-        :return: A SATOSA internal response.
-        """
-        auth_info = AuthenticationInformation(UNSPECIFIED, str(datetime.now()), issuer)
-        internal_resp = InternalData(auth_info=auth_info)
-        internal_resp.attributes = self.converter.to_internal("openid", response)
-        internal_resp.subject_id = response["sub"]
-        return internal_resp
-
-    def get_metadata_desc(self):
-        """
-        See satosa.backends.oauth.get_metadata_desc
-        :rtype: satosa.metadata_creation.description.MetadataDescription
-        """
-        return get_metadata_desc_for_oauth_backend(
-            self.config["provider_metadata"]["issuer"], self.config
-        )
-
-
-def _create_client(provider_metadata, client_metadata, verify_ssl=True):
-    """
-    Create a pyoidc client instance.
-    :param provider_metadata: provider configuration information
-    :type provider_metadata: Mapping[str, Union[str, Sequence[str]]]
-    :param client_metadata: client metadata
-    :type client_metadata: Mapping[str, Union[str, Sequence[str]]]
-    :return: client instance to use for communicating with the configured provider
-    :rtype: oic.oic.Client
-    """
-    client = oic.Client(client_authn_method=CLIENT_AUTHN_METHOD, verify_ssl=verify_ssl)
-
-    # Provider configuration information
-    if "authorization_endpoint" in provider_metadata:
-        # no dynamic discovery necessary
-        client.handle_provider_config(
-            ProviderConfigurationResponse(**provider_metadata),
-            provider_metadata["issuer"],
-        )
-    else:
-        # do dynamic discovery
-        client.provider_config(provider_metadata["issuer"])
-
-    # Client information
-    if "client_id" in client_metadata:
-        # static client info provided
-        client.store_registration_info(RegistrationRequest(**client_metadata))
-    else:
-        # do dynamic registration
-        client.register(
-            client.provider_info["registration_endpoint"], **client_metadata
-        )
-
-    client.subject_type = (
-        client.registration_response.get("subject_type")
-        or client.provider_info["subject_types_supported"][0]
-    )
-    return client

tests/satosa/test_attribute_mapping.py

@@ -5,6 +5,56 @@
 from satosa.attribute_mapping import AttributeMapper
 
 
+class TestAttributeMapperNestedDataDifferentAttrProfile:
+    def test_nested_mapping_nested_data_to_internal(self):
+        mapping = {
+            "attributes": {
+                "name": {
+                    "openid": ["name"]
+                },
+                "givenname": {
+                    "openid": ["given_name", "name.firstName"]
+                },
+            },
+        }
+
+        data = {
+            "name": {
+                "firstName": "value-first",
+                "lastName": "value-last",
+            },
+            "email": "[email protected]",
+        }
+
+        converter = AttributeMapper(mapping)
+        internal_repr = converter.to_internal("openid", data)
+        assert internal_repr["name"] == [data["name"]]
+        assert internal_repr["givenname"] == [data["name"]["firstName"]]
+
+
+    def test_nested_mapping_simple_data_to_internal(self):
+        mapping = {
+            "attributes": {
+                "name": {
+                    "openid": ["name"]
+                },
+                "givenname": {
+                    "openid": ["given_name", "name.firstName"]
+                },
+            },
+        }
+
+        data = {
+            "name": "value-first",
+            "email": "[email protected]",
+        }
+
+        converter = AttributeMapper(mapping)
+        internal_repr = converter.to_internal("openid", data)
+        assert internal_repr["name"] == [data["name"]]
+        assert internal_repr.get("givenname") is None
+
+
 class TestAttributeMapper:
     def test_nested_attribute_to_internal(self):
         mapping = {