# keystone.federation.utils

# Source code for keystone.federation.utils

    # Licensed under the Apache License, Version 2.0 (the "License"); you may
    # not use this file except in compliance with the License. You may obtain
    # a copy of the License at
    #
    #      http://www.apache.org/licenses/LICENSE-2.0
    #
    # Unless required by applicable law or agreed to in writing, software
    # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
    # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
    # License for the specific language governing permissions and limitations
    # under the License.

    """Utilities for Federation Extension."""

    import ast
    import copy
    import json
    import re

    import flask
    import jsonschema
    from oslo_config import cfg
    from oslo_log import log
    from oslo_serialization import jsonutils
    from oslo_utils import timeutils

    from keystone.common import provider_api
    import keystone.conf
    from keystone import exception
    from keystone.i18n import _

    CONF = keystone.conf.CONF
    LOG = log.getLogger(__name__)
    PROVIDERS = provider_api.ProviderAPIs

    class UserType:
        """User mapping type."""

        EPHEMERAL = 'ephemeral'
        LOCAL = 'local'

    ROLE_PROPERTIES = {
        "type": "array",
        "items": {
            "type": "object",
            "required": ["name"],
            "properties": {"name": {"type": "string"}},
            "additionalProperties": False,
        },
    }

    PROJECTS_SCHEMA = {
        "type": "array",
        "items": {
            "type": "object",
            "required": ["name", "roles"],
            "additionalProperties": False,
            "properties": {"name": {"type": "string"}, "roles": ROLE_PROPERTIES},
        },
    }

    DOMAIN_SCHEMA = {
        "type": "object",
        "properties": {"id": {"type": "string"}, "name": {"type": "string"}},
        "additionalProperties": False,
    }
    DEFAULT_SCHEMA_VERSION = "1.0"
    SCHEMA_VERSION_HONOR_DOMAIN = "2.0"
    SCHEMA_VERSION_SYNC_PROJECT_ASSIGNMENTS_WITH_IDP_STATE = "3.0"

    IDP_ATTRIBUTE_MAPPING_SCHEMA_1_0 = {
        "type": "object",
        "required": ['rules'],
        "properties": {
            "rules": {
                "minItems": 1,
                "type": "array",
                "items": {
                    "type": "object",
                    "required": ['local', 'remote'],
                    "additionalProperties": False,
                    "properties": {
                        "local": {
                            "type": "array",
                            "items": {
                                "type": "object",
                                "additionalProperties": False,
                                "properties": {
                                    "user": {
                                        "type": "object",
                                        "properties": {
                                            "id": {"type": "string"},
                                            "name": {"type": "string"},
                                            "email": {"type": "string"},
                                            "domain": {
                                                "$ref": "#/definitions/domain"
                                            },
                                            "type": {
                                                "type": "string",
                                                "enum": [
                                                    UserType.EPHEMERAL,
                                                    UserType.LOCAL,
                                                ],
                                            },
                                        },
                                        "additionalProperties": False,
                                    },
                                    "projects": PROJECTS_SCHEMA,
                                    "group": {
                                        "type": "object",
                                        "oneOf": [
                                            {"$ref": "#/definitions/group_by_id"},
                                            {
                                                "$ref": "#/definitions/group_by_name"
                                            },
                                        ],
                                    },
                                    "groups": {"type": "string"},
                                    "group_ids": {"type": "string"},
                                    "domain": {"$ref": "#/definitions/domain"},
                                },
                            },
                        },
                        "remote": {
                            "minItems": 1,
                            "type": "array",
                            "items": {
                                "type": "object",
                                "oneOf": [
                                    {"$ref": "#/definitions/empty"},
                                    {"$ref": "#/definitions/any_one_of"},
                                    {"$ref": "#/definitions/not_any_of"},
                                    {"$ref": "#/definitions/blacklist"},
                                    {"$ref": "#/definitions/whitelist"},
                                ],
                            },
                        },
                    },
                },
            },
            "schema_version": {"type": "string"},
        },
        "definitions": {
            "empty": {
                "type": "object",
                "required": ['type'],
                "properties": {"type": {"type": "string"}},
                "additionalProperties": False,
            },
            "any_one_of": {
                "type": "object",
                "additionalProperties": False,
                "required": ['type', 'any_one_of'],
                "properties": {
                    "type": {"type": "string"},
                    "any_one_of": {"type": "array"},
                    "regex": {"type": "boolean"},
                },
            },
            "not_any_of": {
                "type": "object",
                "additionalProperties": False,
                "required": ['type', 'not_any_of'],
                "properties": {
                    "type": {"type": "string"},
                    "not_any_of": {"type": "array"},
                    "regex": {"type": "boolean"},
                },
            },
            "blacklist": {
                "type": "object",
                "additionalProperties": False,
                "required": ['type', 'blacklist'],
                "properties": {
                    "type": {"type": "string"},
                    "blacklist": {"type": "array"},
                    "regex": {"type": "boolean"},
                },
            },
            "whitelist": {
                "type": "object",
                "additionalProperties": False,
                "required": ['type', 'whitelist'],
                "properties": {
                    "type": {"type": "string"},
                    "whitelist": {"type": "array"},
                    "regex": {"type": "boolean"},
                },
            },
            "domain": DOMAIN_SCHEMA,
            "group_by_id": {
                "type": "object",
                "properties": {"id": {"type": "string"}},
                "additionalProperties": False,
                "required": ["id"],
            },
            "group_by_name": {
                "type": "object",
                "properties": {
                    "name": {"type": "string"},
                    "domain": {"$ref": "#/definitions/domain"},
                },
                "additionalProperties": False,
                "required": ["name", "domain"],
            },
        },
    }

    # `IDP_ATTRIBUTE_MAPPING_SCHEMA_2_0` adds the domain option for projects,
    # the goal is to work in a similar fashion as `user` and `groups` properties
    IDP_ATTRIBUTE_MAPPING_SCHEMA_2_0: dict = copy.deepcopy(
        IDP_ATTRIBUTE_MAPPING_SCHEMA_1_0
    )

    PROJECTS_SCHEMA_2_0: dict = copy.deepcopy(PROJECTS_SCHEMA)
    PROJECTS_SCHEMA_2_0["items"]["properties"]["domain"] = DOMAIN_SCHEMA

    IDP_ATTRIBUTE_MAPPING_SCHEMA_2_0['properties']['rules']['items']['properties'][
        'local'
    ]['items']['properties']['projects'] = PROJECTS_SCHEMA_2_0

    IDP_ATTRIBUTE_MAPPING_SCHEMA_3_0: dict = copy.deepcopy(
        IDP_ATTRIBUTE_MAPPING_SCHEMA_2_0
    )

    IDP_ATTRIBUTE_MAPPING_SCHEMA_3_0['properties']['rules']['items']['properties'][
        'local'
    ]['items']['properties']['projects_json'] = {"type": "string"}

    def get_default_attribute_mapping_schema_version():
        return CONF.federation.attribute_mapping_default_schema_version

    class DirectMaps:
        """An abstraction around the remote matches.

        Each match is treated internally as a list.
        """

        def __init__(self):
            self._matches = []

        def __str__(self):
            """Return the direct map array as a string."""
            return f'{self._matches}'

        def add(self, values):
            """Add a matched value to the list of matches.

            :param list value: the match to save

            """
            self._matches.append(values)

        def __getitem__(self, idx):
            """Used by Python when executing ``''.format(*DirectMaps())``."""
            value = self._matches[idx]
            if isinstance(value, list) and len(value) == 1:
                return value[0]
            else:
                return value

    def validate_mapping_structure(ref):
        version = ref.get(
            'schema_version', get_default_attribute_mapping_schema_version()
        )

        LOG.debug(
            "Validating mapping [%s] using validator from version [%s].",
            ref,
            version,
        )

        v = jsonschema.Draft4Validator(
            IDP_ATTRIBUTE_MAPPING_SCHEMAS[version]['schema']
        )

        messages = ''
        for error in sorted(v.iter_errors(ref), key=str):
            messages = messages + error.message + "\n"

        if messages:
            raise exception.ValidationError(messages)

    def validate_expiration(token):
        token_expiration_datetime = timeutils.normalize_time(
            timeutils.parse_isotime(token.expires_at)
        )
        if timeutils.utcnow() > token_expiration_datetime:
            raise exception.Unauthorized(_('Federation token is expired'))

    def get_remote_id_parameter(idp, protocol):
        # NOTE(marco-fargetta): Since we support any protocol ID, we attempt to
        # retrieve the remote_id_attribute of the protocol ID. It will look up
        # first if the remote_id_attribute exists.
        protocol_ref = PROVIDERS.federation_api.get_protocol(idp['id'], protocol)
        remote_id_parameter = protocol_ref.get('remote_id_attribute')
        if remote_id_parameter:
            return remote_id_parameter
        else:
            # If it's not registered in the config, then register the option and
            # try again. This allows the user to register protocols other than
            # oidc and saml2.
            try:
                remote_id_parameter = CONF[protocol]['remote_id_attribute']
            except AttributeError:
                # TODO(dolph): Move configuration registration to keystone.conf
                CONF.register_opt(
                    cfg.StrOpt('remote_id_attribute'), group=protocol
                )
                try:
                    remote_id_parameter = CONF[protocol]['remote_id_attribute']
                except AttributeError:  # nosec
                    # No remote ID attr, will be logged and use the default
                    # instead.
                    pass
        if not remote_id_parameter:
            LOG.debug(
                'Cannot find "remote_id_attribute" in configuration '
                'group %s. Trying default location in '
                'group federation.',
                protocol,
            )
            remote_id_parameter = CONF.federation.remote_id_attribute

        return remote_id_parameter

    def validate_idp(idp, protocol, assertion):
        """The IdP providing the assertion should be registered for the mapping."""
        remote_id_parameter = get_remote_id_parameter(idp, protocol)
        if not remote_id_parameter or not idp['remote_ids']:
            LOG.debug('Impossible to identify the IdP %s ', idp['id'])
            # If nothing is defined, the administrator may want to
            # allow the mapping of every IdP
            return
        try:
            idp_remote_identifier = assertion[remote_id_parameter]
        except KeyError:
            msg = _('Could not find Identity Provider identifier in environment')
            raise exception.ValidationError(msg)
        if idp_remote_identifier not in idp['remote_ids']:
            msg = _(
                'Incoming identity provider identifier not included '
                'among the accepted identifiers.'
            )
            raise exception.Forbidden(msg)

    def validate_mapped_group_ids(group_ids, mapping_id, identity_api):
        """Iterate over group ids and make sure they are present in the backend.

        This call is not transactional.
        :param group_ids: IDs of the groups to be checked
        :type group_ids: list of str

        :param mapping_id: id of the mapping used for this operation
        :type mapping_id: str

        :param identity_api: Identity Manager object used for communication with
                             backend
        :type identity_api: identity.Manager

        :raises keystone.exception.MappedGroupNotFound: If the group returned by
            mapping was not found in the backend.

        """
        for group_id in group_ids:
            try:
                identity_api.get_group(group_id)
            except exception.GroupNotFound:
                raise exception.MappedGroupNotFound(
                    group_id=group_id, mapping_id=mapping_id
                )

    # TODO(marek-denis): Optimize this function, so the number of calls to the
    # backend are minimized.

    def transform_to_group_ids(
        group_names, mapping_id, identity_api, resource_api
    ):
        """Transform groups identified by name/domain to their ids.

        Function accepts list of groups identified by a name and domain giving
        a list of group ids in return. A message is logged if the group doesn't
        exist in the backend.

        Example of group_names parameter::

            [
                {"name": "group_name", "domain": {"id": "domain_id"}},
                {"name": "group_name_2", "domain": {"name": "domain_name"}},
            ]

        :param group_names: list of group identified by name and its domain.
        :type group_names: list

        :param mapping_id: id of the mapping used for mapping assertion into
            local credentials
        :type mapping_id: str

        :param identity_api: identity_api object
        :param resource_api: resource manager object

        :returns: generator object with group ids

        """

        def resolve_domain(domain):
            """Return domain id.

            Input is a dictionary with a domain identified either by a ``id`` or a
            ``name``. In the latter case system will attempt to fetch domain object
            from the backend.

            :returns: domain's id
            :rtype: str

            """
            domain_id = domain.get('id') or resource_api.get_domain_by_name(
                domain.get('name')
            ).get('id')
            return domain_id

        for group in group_names:
            try:
                group_dict = identity_api.get_group_by_name(
                    group['name'], resolve_domain(group['domain'])
                )
                yield group_dict['id']
            except exception.GroupNotFound:
                LOG.debug('Group %s has no entry in the backend', group['name'])

    def get_assertion_params_from_env():
        LOG.debug('Environment variables: %s', flask.request.environ)
        prefix = CONF.federation.assertion_prefix
        for k, v in list(flask.request.environ.items()):
            if not k.startswith(prefix):
                continue
            # These bytes may be decodable as ISO-8859-1 according to Section
            # 3.2.4 of RFC 7230. Let's assume that our web server plugins are
            # correctly encoding the data.
            if not isinstance(v, str) and getattr(v, 'decode', False):
                v = v.decode('ISO-8859-1')
            yield (k, v)

    class RuleProcessor:
        """A class to process assertions and mapping rules."""

        class _EvalType:
            """Mapping rule evaluation types."""

            ANY_ONE_OF = 'any_one_of'
            NOT_ANY_OF = 'not_any_of'
            BLACKLIST = 'blacklist'
            WHITELIST = 'whitelist'

        def __init__(self, mapping_id, rules):
            """Initialize RuleProcessor.

            Example rules can be found at:
            :class:`keystone.tests.mapping_fixtures`

            :param mapping_id: id for the mapping
            :type mapping_id: string
            :param rules: rules from a mapping
            :type rules: dict

            """
            self.mapping_id = mapping_id
            self.rules = rules

        def process(self, assertion_data):
            """Transform assertion to a dictionary.

            The dictionary contains mapping of user name and group ids
            based on mapping rules.

            This function will iterate through the mapping rules to find
            assertions that are valid.

            :param assertion_data: an assertion containing values from an IdP
            :type assertion_data: dict

            Example assertion_data::

                {
                    'Email': 'testacct@example.com',
                    'UserName': 'testacct',
                    'FirstName': 'Test',
                    'LastName': 'Account',
                    'orgPersonType': 'Tester',
                }

            :returns: dictionary with user and group_ids

            The expected return structure is::

                {
                    'name': 'foobar',
                    'group_ids': ['abc123', 'def456'],
                    'group_names': [
                        {'name': 'group_name_1', 'domain': {'name': 'domain1'}},
                        {'name': 'group_name_1_1', 'domain': {'name': 'domain1'}},
                        {'name': 'group_name_2', 'domain': {'id': 'xyz132'}},
                    ],
                }

            """
            # Assertions will come in as string key-value pairs, and will use a
            # semi-colon to indicate multiple values, i.e. groups.
            # This will create a new dictionary where the values are arrays, and
            # any multiple values are stored in the arrays.
            LOG.debug('assertion data: %s', assertion_data)
            assertion = {
                n: v.split(';')
                for n, v in assertion_data.items()
                if isinstance(v, str)
            }
            LOG.debug('assertion: %s', assertion)
            identity_values = []

            LOG.debug('rules: %s', self.rules)
            for rule in self.rules:
                direct_maps = self._verify_all_requirements(
                    rule['remote'], assertion
                )

                # If the compare comes back as None, then the rule did not apply
                # to the assertion data, go on to the next rule
                if direct_maps is None:
                    continue

                # If there are no direct mappings, then add the local mapping
                # directly to the array of saved values. However, if there is
                # a direct mapping, then perform variable replacement.
                if not direct_maps:
                    identity_values += rule['local']
                else:
                    for local in rule['local']:
                        new_local = self._update_local_mapping(local, direct_maps)
                        identity_values.append(new_local)

            LOG.debug('identity_values: %s', identity_values)
            mapped_properties = self._transform(identity_values)
            LOG.debug('mapped_properties: %s', mapped_properties)
            return mapped_properties

        @staticmethod
        def _ast_literal_eval(value):
            try:
                values = ast.literal_eval(value)
                # We expect a list here, but literal_eval will successfully
                # parse and return a single value. We raise ValueError if so.
                if not isinstance(values, list):
                    raise ValueError
            except (ValueError, SyntaxError):
                values = [value]
            return values

        def _normalize_groups(self, identity_value):
            # In this case, identity_value['groups'] is a string
            # representation of a list, and we want a real list.  This is
            # due to the way we do direct mapping substitutions today (see
            # function _update_local_mapping() )
            if 'name' in identity_value['groups']:
                group_names_list = self._ast_literal_eval(identity_value['groups'])

                def convert_json(group):
                    if group.startswith('JSON:'):
                        return jsonutils.loads(group.lstrip('JSON:'))
                    return group

                group_dicts = [convert_json(g) for g in group_names_list]
                for g in group_dicts:
                    if 'domain' not in g:
                        msg = _(
                            "Invalid rule: %(identity_value)s. Both "
                            "'groups' and 'domain' keywords must be "
                            "specified."
                        )
                        msg = msg % {'identity_value': identity_value}
                        raise exception.ValidationError(msg)
            else:
                if 'domain' not in identity_value:
                    msg = _(
                        "Invalid rule: %(identity_value)s. Both "
                        "'groups' and 'domain' keywords must be "
                        "specified."
                    )
                    msg = msg % {'identity_value': identity_value}
                    raise exception.ValidationError(msg)
                group_names_list = self._ast_literal_eval(identity_value['groups'])
                domain = identity_value['domain']
                group_dicts = [
                    {'name': name, 'domain': domain} for name in group_names_list
                ]
            return group_dicts

        def normalize_user(self, user, default_mapping_domain):
            """Parse and validate user mapping."""
            if user.get('type') is None:
                user['type'] = UserType.EPHEMERAL
            if user.get('type') not in (UserType.EPHEMERAL, UserType.LOCAL):
                msg = _("User type %s not supported") % user.get('type')
                raise exception.ValidationError(msg)

        def extract_groups(self, groups_by_domain):
            for groups in list(groups_by_domain.values()):
                yield from list({g['name']: g for g in groups}.values())

        def _transform(self, identity_values):
            """Transform local mappings, to an easier to understand format.

            Transform the incoming array to generate the return value for
            the process function. Generating content for Keystone tokens will
            be easier if some pre-processing is done at this level.

            :param identity_values: local mapping from valid evaluations
            :type identity_values: array of dict

            Example identity_values::

                [
                    {
                        'group': {'id': '0cd5e9'},
                        'user': {
                            'email': 'bob@example.com'
                        },
                    },
                    {
                        'groups': ['member', 'admin', tester'],
                        'domain': {
                            'name': 'default_domain'
                        }
                    },
                    {
                        'group_ids': ['abc123', 'def456', '0cd5e9']
                    }
                ]

            :returns: dictionary with user name, group_ids and group_names.
            :rtype: dict

            """
            # initialize the group_ids as a set to eliminate duplicates
            user = {}
            group_ids = set()
            group_names = []
            groups_by_domain = {}
            projects = []

            # if mapping yield no valid identity values, we should bail right away
            # instead of continuing on with a normalized bogus user
            if not identity_values:
                msg = (
                    "Could not map any federated user properties to identity "
                    "values. Check debug logs or the mapping used for "
                    "additional details."
                )
                tr_msg = _(
                    "Could not map any federated user properties to "
                    "identity values. Check debug logs or the mapping "
                    "used for additional details."
                )
                LOG.warning(msg)
                raise exception.ValidationError(tr_msg)

            for identity_value in identity_values:
                if 'user' in identity_value:
                    # if a mapping outputs more than one user name, log it
                    if user:
                        LOG.warning(
                            'Ignoring user [%s]', identity_value.get('user')
                        )
                    else:
                        user = identity_value.get('user')

                if 'group' in identity_value:
                    group = identity_value['group']
                    if 'id' in group:
                        group_ids.add(group['id'])
                    elif 'name' in group:
                        groups = self.process_group_by_name(
                            group, groups_by_domain
                        )
                        group_names.extend(groups)
                if 'groups' in identity_value:
                    group_dicts = self._normalize_groups(identity_value)
                    group_names.extend(group_dicts)
                if 'group_ids' in identity_value:
                    # If identity_values['group_ids'] is a string representation
                    # of a list, parse it to a real list. Also, if the provided
                    # group_ids parameter contains only one element, it will be
                    # parsed as a simple string, and not a list or the
                    # representation of a list.
                    group_ids.update(
                        self._ast_literal_eval(identity_value['group_ids'])
                    )

                if (
                    'projects' in identity_value
                    or 'projects_json' in identity_value
                ):
                    projects = self.extract_projects(identity_value)

            self.normalize_user(user, identity_value.get('domain'))

            return {
                'user': user,
                'group_ids': list(group_ids),
                'group_names': group_names,
                'projects': projects,
            }

        def process_group_by_name(self, group, groups_by_domain):
            domain = group['domain'].get('name') or group['domain'].get('id')
            groups_by_domain.setdefault(domain, []).append(group)
            return self.extract_groups(groups_by_domain)

        def extract_projects(self, identity_value):
            return identity_value.get('projects', [])

        def _update_local_mapping(self, local, direct_maps):
            """Replace any {0}, {1} ... values with data from the assertion.

            :param local: local mapping reference that needs to be updated
            :type local: dict
            :param direct_maps: identity values used to update local
            :type direct_maps: keystone.federation.utils.DirectMaps

            Example local::

                {'user': {'name': '{0} {1}', 'email': '{2}'}}

            Example direct_maps::

                [['Bob'], ['Thompson'], ['bob@example.com']]

            :returns: new local mapping reference with replaced values.

            The expected return structure is::

                {'user': {'name': 'Bob Thompson', 'email': 'bob@example.org'}}

            :raises keystone.exception.DirectMappingError: when referring to a
                remote match from a local section of a rule

            """
            LOG.debug('direct_maps: %s', direct_maps)
            LOG.debug('local: %s', local)
            new = {}
            for k, v in local.items():
                LOG.debug(
                    "Processing rule [%s] for key [%s] with parameters [%s]",
                    v,
                    k,
                    direct_maps,
                )
                new_value = self.process_rule_mapping(direct_maps, k, v)
                LOG.debug(
                    "The rule [%s] for key [%s] was mapped to [%s] using "
                    "parameters [%s].",
                    v,
                    k,
                    new_value,
                    direct_maps,
                )

                new[k] = new_value
            return new

        def process_rule_mapping(self, direct_maps, key, rule):
            if isinstance(rule, dict):
                new_value = self._update_local_mapping(rule, direct_maps)
            elif isinstance(rule, list):
                new_value = [
                    self._update_local_mapping(item, direct_maps) for item in rule
                ]
            else:
                try:
                    new_value = rule.format(*direct_maps)
                except IndexError:
                    LOG.debug(
                        "Error while processing rule [%s] for key [%s] "
                        "with parameters [%s].",
                        rule,
                        key,
                        direct_maps.__dict__,
                    )
                    raise exception.DirectMappingError(mapping_id=self.mapping_id)
            return new_value

        def _verify_all_requirements(self, requirements, assertion):
            """Compare remote requirements of a rule against the assertion.

            If a value of ``None`` is returned, the rule with this assertion
            doesn't apply.
            If an array of zero length is returned, then there are no direct
            mappings to be performed, but the rule is valid.
            Otherwise, then it will first attempt to filter the values according
            to blacklist or whitelist rules and finally return the values in
            order, to be directly mapped.

            :param requirements: list of remote requirements from rules
            :type requirements: list

            Example requirements::

                [
                    {"type": "UserName"},
                    {"type": "orgPersonType", "any_one_of": ["Customer"]},
                    {"type": "ADFS_GROUPS", "whitelist": ["g1", "g2", "g3", "g4"]},
                ]

            :param assertion: dict of attributes from an IdP
            :type assertion: dict

            Example assertion::

                {
                    'UserName': ['testacct'],
                    'LastName': ['Account'],
                    'orgPersonType': ['Tester'],
                    'Email': ['testacct@example.com'],
                    'FirstName': ['Test'],
                    'ADFS_GROUPS': ['g1', 'g2'],
                }

            :returns: identity values used to update local
            :rtype: keystone.federation.utils.DirectMaps or None

            """
            direct_maps = DirectMaps()

            for requirement in requirements:
                requirement_type = requirement['type']
                direct_map_values = assertion.get(requirement_type)
                regex = requirement.get('regex', False)

                if not direct_map_values:
                    return None

                any_one_values = requirement.get(self._EvalType.ANY_ONE_OF)
                if any_one_values is not None:
                    if self._evaluate_requirement(
                        any_one_values,
                        direct_map_values,
                        self._EvalType.ANY_ONE_OF,
                        regex,
                    ):
                        continue
                    else:
                        return None

                not_any_values = requirement.get(self._EvalType.NOT_ANY_OF)
                if not_any_values is not None:
                    if self._evaluate_requirement(
                        not_any_values,
                        direct_map_values,
                        self._EvalType.NOT_ANY_OF,
                        regex,
                    ):
                        continue
                    else:
                        return None

                # If 'any_one_of' or 'not_any_of' are not found, then values are
                # within 'type'. Attempt to find that 'type' within the assertion,
                # and filter these values if 'whitelist' or 'blacklist' is set.
                blacklisted_values = requirement.get(self._EvalType.BLACKLIST)
                whitelisted_values = requirement.get(self._EvalType.WHITELIST)

                # If a blacklist or whitelist is used, we want to map to the
                # whole list instead of just its values separately.
                if blacklisted_values is not None:
                    direct_map_values = self._evaluate_requirement(
                        blacklisted_values,
                        direct_map_values,
                        self._EvalType.BLACKLIST,
                        regex,
                    )
                elif whitelisted_values is not None:
                    direct_map_values = self._evaluate_requirement(
                        whitelisted_values,
                        direct_map_values,
                        self._EvalType.WHITELIST,
                        regex,
                    )

                direct_maps.add(direct_map_values)

                LOG.debug('updating a direct mapping: %s', direct_map_values)

            return direct_maps

        def _evaluate_values_by_regex(self, values, assertion_values):
            return [
                assertion
                for assertion in assertion_values
                if any(re.search(regex, assertion) for regex in values)
            ]

        def _evaluate_requirement(
            self, values, assertion_values, eval_type, regex
        ):
            """Evaluate the incoming requirement and assertion.

            Filter the incoming assertions against the requirement values. If regex
            is specified, the assertion list is filtered by checking if any of the
            requirement regexes matches. Otherwise, the list is filtered by string
            equality with any of the allowed values.

            Once the assertion values are filtered, the output is determined by the
            evaluation type:
                any_one_of: return True if there are any matches, False otherwise
                not_any_of: return True if there are no matches, False otherwise
                blacklist: return the incoming values minus any matches
                whitelist: return only the matched values

            :param values: list of allowed values, defined in the requirement
            :type values: list
            :param assertion_values: The values from the assertion to evaluate
            :type assertion_values: list/string
            :param eval_type: determine how to evaluate requirements
            :type eval_type: string
            :param regex: perform evaluation with regex
            :type regex: boolean

            :returns: list of filtered assertion values (if evaluation type is
                      'blacklist' or 'whitelist'), or boolean indicating if the
                      assertion values fulfill the requirement (if evaluation type
                      is 'any_one_of' or 'not_any_of')

            """
            if regex:
                matches = self._evaluate_values_by_regex(values, assertion_values)
            else:
                matches = set(values).intersection(set(assertion_values))

            if eval_type == self._EvalType.ANY_ONE_OF:
                return bool(matches)
            elif eval_type == self._EvalType.NOT_ANY_OF:
                return not bool(matches)
            elif eval_type == self._EvalType.BLACKLIST:
                return list(set(assertion_values).difference(set(matches)))
            elif eval_type == self._EvalType.WHITELIST:
                return list(matches)
            else:
                raise exception.UnexpectedError(
                    _('Unexpected evaluation type "%(eval_type)s"')
                    % {'eval_type': eval_type}
                )

    def assert_enabled_identity_provider(federation_api, idp_id):
        identity_provider = federation_api.get_idp(idp_id)
        if identity_provider.get('enabled') is not True:
            msg = f'Identity Provider {idp_id} is disabled'
            tr_msg = _('Identity Provider %(idp)s is disabled') % {'idp': idp_id}
            LOG.debug(msg)
            raise exception.Forbidden(tr_msg)

    def assert_enabled_service_provider_object(service_provider):
        if service_provider.get('enabled') is not True:
            sp_id = service_provider['id']
            msg = f'Service Provider {sp_id} is disabled'
            tr_msg = _('Service Provider %(sp)s is disabled') % {'sp': sp_id}
            LOG.debug(msg)
            raise exception.Forbidden(tr_msg)

    class RuleProcessorToHonorDomainOption(RuleProcessor):
        """Handles the default domain configured in the attribute mapping.

        This rule processor is designed to handle the `domain` attribute
        configured at the root of the attribute mapping. When this attribute is
        configured, we should take it as the default one for the attribute
        mapping, instead of the domain of the IdP. Moreover, we should respect
        the override to it that can take place at the `groups`, `user`,
        and `projects` attributes definition.
        """

        def __init__(self, mapping_id, rules):
            super().__init__(mapping_id, rules)

        def extract_projects(self, identity_value):
            projects = super().extract_projects(identity_value)
            return self.normalize_projects(identity_value, projects)

        def normalize_projects(self, identity_value, projects):
            default_mapping_domain = identity_value.get("domain")
            for project in projects:
                if not project.get("domain"):
                    LOG.debug(
                        "Configuring the domain [%s] for project [%s].",
                        default_mapping_domain,
                        project,
                    )
                    project["domain"] = default_mapping_domain
            return projects

        def normalize_user(self, user, default_mapping_domain):
            super().normalize_user(user, default_mapping_domain)
            if not user.get("domain"):
                LOG.debug(
                    "Configuring the domain [%s] for user [%s].",
                    default_mapping_domain,
                    user,
                )
                user["domain"] = default_mapping_domain
            else:
                LOG.debug(
                    "The user [%s] was configured with a domain. "
                    "Therefore, we do not need to define.",
                    user,
                )

    class RuleProcessorForProjectJsonAttribute(RuleProcessorToHonorDomainOption):
        """Handles the 'project_json' property in the attribute mapping

        This rule processor is designed to handle the `project_json` attribute
        configured at the root of the attribute mapping. When this attribute is
        configured, we load it (the JSON that it contains), and append to the
        possible set of hard-coded projects that we find at `projects` property.
        """

        PATTERN_PROJECT_JSON_INDEX = r"\{?(\d+)\}?"

        def __init__(self, mapping_id, rules):
            super().__init__(mapping_id, rules)

            self.regular_expression_index_of_project_json = re.compile(
                RuleProcessorForProjectJsonAttribute.PATTERN_PROJECT_JSON_INDEX
            )

        def extract_projects(self, identity_value):
            projects = super().extract_projects(identity_value)
            projects += identity_value.get('projects_json', [])
            return self.normalize_projects(identity_value, projects)

        def process_rule_mapping(self, direct_maps, key, rule):
            # I left the handling of both 'projects' and 'projects_json'.
            # This can help people to combine the old method with the new approach as well.
            if key != "projects_json" and key != "projects":
                return super().process_rule_mapping(direct_maps, key, rule)

            if key == "projects" and not isinstance(rule, str):
                return super().process_rule_mapping(direct_maps, key, rule)

            matcher = self.regular_expression_index_of_project_json.match(rule)

            if matcher:
                project_json_index = matcher.group(1)
                projects_json = direct_maps[int(project_json_index)]
                if isinstance(projects_json, str):
                    projects_json = json.loads(projects_json)
                LOG.debug(
                    "%s key was solved to value [%s] at index [%s] of "
                    "parameters [%s].",
                    key,
                    projects_json,
                    project_json_index,
                    direct_maps,
                )
                RuleProcessorForProjectJsonAttribute.validate_projects_json(
                    projects_json
                )

                return projects_json
            else:
                raise exception.SchemaValidationError(
                    f"The rule [{rule}] for key [{key}] does not match the expected"
                    f" pattern [{RuleProcessorForProjectJsonAttribute.PATTERN_PROJECT_JSON_INDEX}]."
                )

        @staticmethod
        def validate_projects_json(projects_json):
            json_validator = jsonschema.Draft4Validator(PROJECTS_SCHEMA_2_0)
            messages = ''
            for error in sorted(
                json_validator.iter_errors(projects_json), key=str
            ):
                messages = messages + error.message + "\n"
            if messages:
                LOG.error("Error validating 'projects_json': [%s].", messages)
                raise exception.ValidationError(messages)

    IDP_ATTRIBUTE_MAPPING_SCHEMAS = {
        DEFAULT_SCHEMA_VERSION: {
            "schema": IDP_ATTRIBUTE_MAPPING_SCHEMA_1_0,
            "processor": RuleProcessor,
        },
        SCHEMA_VERSION_HONOR_DOMAIN: {
            "schema": IDP_ATTRIBUTE_MAPPING_SCHEMA_2_0,
            "processor": RuleProcessorToHonorDomainOption,
        },
        SCHEMA_VERSION_SYNC_PROJECT_ASSIGNMENTS_WITH_IDP_STATE: {
            "schema": IDP_ATTRIBUTE_MAPPING_SCHEMA_3_0,
            "processor": RuleProcessorForProjectJsonAttribute,
        },
    }

    def create_attribute_mapping_rules_processor(mapping):
        version = mapping.get(
            'schema_version', get_default_attribute_mapping_schema_version()
        )

        return IDP_ATTRIBUTE_MAPPING_SCHEMAS[version]['processor'](
            mapping['id'], mapping['rules']
        )

this page last updated: 2026-06-11 21:52:50

Except where otherwise noted, this document is licensed under Creative Commons Attribution 3.0 License. See all OpenStack Legal Documents.

found an error? report a bug