[AAP-48713] Alter authenticator maps to work with attribute expansion (ansible#768)

## Description
<!-- Mandatory: Provide a clear, concise description of the changes and
their purpose -->
- What is being changed?
If you add the syntax `{% for_attr_value(<attribute name>) %}` to an
authenticator map we will expand that value based on the values returned
from an attribute on login.

- Why is this change needed?
This will make gateway backwards compatible with an obscure SAML setting
from AWX.

- How does this change address the issue?
This allows authenticator maps to behave like the SAML authenticator
can.


## Type of Change
<!-- Mandatory: Check one or more boxes that apply -->
- [ ] Bug fix (non-breaking change which fixes an issue)
- [X] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to not work as expected)
- [ ] Documentation update
- [ ] Test update
- [X] Refactoring (no functional changes)
- [ ] Development environment change
- [ ] Configuration change

## Self-Review Checklist
<!-- These items help ensure quality - they complement our automated CI
checks -->
- [X] I have performed a self-review of my code
- [X] I have added relevant comments to complex code sections
- [X] I have updated documentation where needed
- [X] I have considered the security impact of these changes
- [X] I have considered performance implications
- [X] I have thought about error handling and edge cases
- [X] I have tested the changes in my local environment

## Testing Instructions
<!-- Optional for test-only changes. Mandatory for all other changes -->
<!-- Must be detailed enough for reviewers to reproduce -->
### Prerequisites
<!-- List any specific setup required -->

### Steps to Test
1. Setup an authenticator which return multiple attribute values
2. Create authenticator maps that expand values
3. Login through the authenticator
4. Assert that the values were expanded as expected

### Expected Results
<!-- Describe what should happen after following the steps -->

## Additional Context
<!-- Optional but helpful information -->

### Required Actions
<!-- Check if changes require work in other areas -->
<!-- Remove section if no external actions needed -->
- [ ] Requires documentation updates
  <!-- API docs, feature docs, deployment guides -->
- [ ] Requires downstream repository changes
  <!-- Specify repos: django-ansible-base, eda-server, etc. -->
- [ ] Requires infrastructure/deployment changes
  <!-- CI/CD, installer updates, new services -->
- [ ] Requires coordination with other teams
  <!-- UI team, platform services, infrastructure -->
- [ ] Blocked by PR/MR: #XXX
  <!-- Reference blocking PRs/MRs with brief context -->

### Screenshots/Logs
<!-- Add if relevant to demonstrate the changes -->

Author: john-westcott-iv

ansible_base/authentication/serializers/authenticator_map.py

@@ -1,16 +1,15 @@
 import logging
+from typing import Optional
 
-from django.conf import settings
-from django.core.exceptions import ObjectDoesNotExist
 from django.utils.translation import gettext_lazy as _
 from rest_framework.serializers import ValidationError
 
 from ansible_base.authentication.models import AuthenticatorMap
-from ansible_base.authentication.utils.authenticator_map import _EXPANSION_FIELDS, check_expansion_syntax, has_expansion
+from ansible_base.authentication.utils.authenticator_map import _EXPANSION_FIELDS, check_expansion_syntax, check_role_type, has_expansion
 from ansible_base.authentication.utils.trigger_definition import TRIGGER_DEFINITION
 from ansible_base.lib.serializers.common import NamedCommonModelSerializer
-from ansible_base.lib.utils.auth import get_organization_model, get_team_model
 from ansible_base.lib.utils.string import is_empty
+from ansible_base.lib.utils.typing import TranslatedString
 
 logger = logging.getLogger('ansible_base.authentication.serializers.authenticator_map')
 
@@ -51,57 +50,12 @@ def validate(self, data) -> dict:
             raise ValidationError(errors)
         return data
 
-    @staticmethod
-    def _is_rbac_installed():
-        return 'ansible_base.rbac' in settings.INSTALLED_APPS
-
-    def validate_role_data(self, map_type, role, org, team):
-        errors = {}
-
-        # Validation is possible only if RBAC is installed
-        if not self._is_rbac_installed():
-            logger.warning(_("You specified a role without RBAC installed "))
-            return errors
-
-        from ansible_base.rbac.models import RoleDefinition
-
-        # If this role is dynamically expanded we can't check it now, only at run time.
+    def validate_role_data(self, map_type: Optional[str], role: Optional[str], org: Optional[str], team: Optional[str]) -> dict[str, TranslatedString]:
+        # If the role field has an expansion in it we can only check this role at runtime
         if has_expansion(role):
-            return errors
-
-        try:
-            rbac_role = RoleDefinition.objects.get(name=role)
-            is_system_role = rbac_role.content_type is None
-
-            # system role is allowed for map type == role without further conditions
-            if is_system_role and map_type == 'role':
-                return errors
+            return {}
 
-            is_org_role, is_team_role = False, False
-            if not is_system_role:
-                model_class = rbac_role.content_type.model_class()
-                is_org_role = issubclass(model_class, get_organization_model())
-                is_team_role = issubclass(model_class, get_team_model())
-
-            # role type and map type must correspond
-            if map_type == 'organization' and not is_org_role:
-                errors['role'] = _("For an organization map type you must specify an organization based role")
-
-            if map_type == 'team' and not is_team_role:
-                errors['role'] = _("For a team map type you must specify a team based role")
-
-            # org/team role needs organization field
-            if (is_org_role or is_team_role) and is_empty(org):
-                errors["organization"] = _("You must specify an organization with the selected role")
-
-            # team role needs team field
-            if is_team_role and is_empty(team):
-                errors["team"] = _("You must specify a team with the selected role")
-
-        except ObjectDoesNotExist:
-            errors['role'] = _("RoleDefinition {role} doesn't exist").format(role=role)
-
-        return errors
+        return check_role_type(map_type, role, org, team)
 
     def validate_trigger_data(self, data):
         errors = {}

ansible_base/authentication/utils/authenticator_map.py

@@ -1,9 +1,21 @@
+import logging
 import re
-from typing import Any, Optional
+from typing import Any, Dict, Optional
 
+from django.conf import settings
+from django.core.exceptions import ObjectDoesNotExist
 from django.utils.translation import gettext_lazy as _
 
+from ansible_base.authentication.models import AuthenticatorMap
+from ansible_base.lib.utils.auth import get_organization_model, get_team_model
+from ansible_base.lib.utils.collection import dict_cartesian_product
+from ansible_base.lib.utils.string import is_empty
+from ansible_base.lib.utils.typing import TranslatedString
+
+logger = logging.getLogger('ansible_base.authentication.utils.authenticator_map')
+
 _EXPANSION_FIELDS = ['organization', 'role', 'team']
+_EXPANSION_RE = re.compile(r'{%\s*for_attr_value\(([^(%})]+)\)\s*%}')
 
 
 def has_expansion(value: Optional[str]) -> bool:
@@ -18,12 +30,201 @@ def has_expansion(value: Optional[str]) -> bool:
         return False
 
 
-def check_expansion_syntax(value: Optional[str]) -> Optional[Any]:
+def check_expansion_syntax(value: Optional[str]) -> Optional[TranslatedString]:
     """
     Check a given field to see if it contains the proper syntax for {% for_attr_value(user_orgs) %}
-
-    Raises a ValidationError if its incorrect
     """
 
-    if has_expansion(value) and not re.search(r'{%\s*for_attr_value\(.+\)\s*%}', value):
+    if has_expansion(value) and not _EXPANSION_RE.search(value):
         return _("Expansion only supports the format {% for_attr_value(attribute) %}")
+
+
+def expand_syntax(attributes: dict, auth_map: AuthenticatorMap) -> list[Dict[str, str]]:
+    """
+    Given attributes and a map, look for the fields that can be expanded and do the expansion
+    If no fields require expansion, do nothing
+
+    returns:
+    list of named tuples of expanded value
+    """
+    expanded_strings = {}
+    for field in _EXPANSION_FIELDS:
+        field_value = getattr(auth_map, field, None)
+
+        if field_value is None:
+            # If this particular field is None we can move on.
+            # This might be true for the team value of an authenticator map type of organization for example.
+            continue
+
+        # Get a list of all of the attributes specified in {% for_attr_value(something) %}
+        # In the above example, we would get a list like ['something'].
+        # If there were more than one in a given string we would have an array like ['something', 'else']
+        attrs_used = _EXPANSION_RE.findall(field_value)
+
+        if has_expansion(field_value) and attrs_used == []:
+            # This field had an invalid expansion so we can't expand anything, just return []
+            logger.info(
+                f"Authenticator Map {auth_map.name} on {auth_map.authenticator.name} has a bad expansion in {field} unable to process map ({field_value})"
+            )
+            continue
+
+        # Make sure that the attributes we got contain all the ones we want to expand
+        if missing_attributes := set(attrs_used).difference(set(attributes.keys())):
+            logger.info(
+                f"Authenticator Map {auth_map.name} on {auth_map.authenticator.name} tried to expand attribute(s) "
+                f"{', '.join(missing_attributes)} but they were not in the users attributes!"
+            )
+            continue
+
+        # Normalize and check the attribute values we are going to use
+        try:
+            normalized_attributes = _normalize_attributes(auth_map, attrs_used, attributes)
+        except ValueError:
+            # If we had issues normalizing the data we can just move on because if can't replace
+            # even one of the items in the string there is no point in trying at all
+            continue
+
+        # Create an entry in our expanded_string dictionary with the initial value of the field
+        expanded_strings[field] = [field_value]
+
+        # For each value in the attribute, expand the strings
+        # Lets say we have a field like "Organization {% for_attr_value(org) %} - Department {% for attr_value(dept) %}"
+        # And in our attributes we had:
+        #    org = ["a", "b"]
+        #    dept = ['Z', 'Y']
+        # First our string list will be:
+        #     ['Organization {% for_attr_value(org) %} - Department {% for attr_value(dept) %}']
+        # After the first pass of this loop, our string list will be:
+        #     ['Organization a - Department {% for attr_value(dept) %}', 'Organization b Department {% for attr_value(dept) %}']
+        # After the second pass the string list will be:
+        #     [
+        #         ['Organization a - Department Z'
+        #         ['Organization a - Department Y'
+        #         ['Organization b - Department Z'
+        #         ['Organization b - Department Y'
+        #     ]
+        expanded_strings[field] = _expand_strings(attrs_used, normalized_attributes, expanded_strings[field])
+
+    # Return the cartesian product of the expanded strings
+    return dict_cartesian_product(expanded_strings)
+
+
+def _expand_strings(attrs_used: list[str], normalized_attributes: dict[str, list[str]], expanded_strings: list[str]) -> list[str]:
+    for attr_name in attrs_used:
+        # Make a list of new strings.
+        new_strings = []
+
+        for value in normalized_attributes[attr_name]:
+            for string in expanded_strings:
+                new_strings.append(_EXPANSION_RE.sub(value, string, 1))
+
+        expanded_strings = new_strings
+
+    return expanded_strings
+
+
+def _normalize_attributes(auth_map: AuthenticatorMap, attrs_used: list[str], attributes: dict) -> dict[str, list[str]]:
+    """
+    Normalize the attributes we are going to use
+
+    Given a list of attributes that are used (i.e. ["first_name", "last_name"]) create and return a dictionary of attributes lke:
+        {
+            "first_name": ["all", "values", "of", "first_name"],
+            "last_name": ["value"],
+        }
+
+    If the list contains non-strings we will raise errors
+
+    Raises ValueError if we got one or more issues when normalizing values
+    """
+    normalized_attributes = {}
+    errors = []
+    for attr_name in attrs_used:
+        attr_errors = []
+        # Make sure the attribute is in the list of attributes we have.
+        attr_value = _make_list(attributes.get(attr_name, []))
+
+        if len(attr_value) == 0:
+            attr_errors.append(
+                f"Authenticator Map {auth_map.name} on {auth_map.authenticator.name} tried to expand attribute {attr_name} "
+                f"but there were not values in that attribute, ignoring"
+            )
+
+        # If any of the values are not a string then we can't expand it
+        for value in attr_value:
+            if type(value) is not str:
+                attr_errors.append(
+                    f"Authenticator Map {auth_map.name} on {auth_map.authenticator.name} tried to expand attribute {attr_name} "
+                    f"but that was not a list of string, instead got {value}, ignoring"
+                )
+
+        if attr_errors == []:
+            normalized_attributes[attr_name] = attr_value
+        else:
+            # If we got errors for this item add them to the overall errors for logging
+            errors.extend(attr_errors)
+            for error in attr_errors:
+                logger.info(error)
+
+    if errors:
+        raise ValueError()
+
+    return normalized_attributes
+
+
+def _make_list(value: Any) -> list[Any]:
+    if type(value) is list:
+        return value
+    return [value]
+
+
+def _is_rbac_installed():
+    """
+    Determine if RBAC is installed.
+    Separated out for easy mocking.
+    """
+    return 'ansible_base.rbac' in settings.INSTALLED_APPS
+
+
+def check_role_type(map_type: Optional[str], role: Optional[str], org: Optional[str], team: Optional[str]) -> dict[str, TranslatedString]:
+    errors = {}
+
+    if not _is_rbac_installed():
+        errors['role'] = _("You specified a role without RBAC installed ")
+        return errors
+
+    from ansible_base.rbac.models import RoleDefinition
+
+    try:
+        rbac_role = RoleDefinition.objects.get(name=role)
+        is_system_role = rbac_role.content_type is None
+
+        # system role is allowed for map type == role without further conditions
+        if is_system_role and map_type == 'role':
+            return errors
+
+        is_org_role, is_team_role = False, False
+        if not is_system_role:
+            model_class = rbac_role.content_type.model_class()
+            is_org_role = issubclass(model_class, get_organization_model())
+            is_team_role = issubclass(model_class, get_team_model())
+
+        # role type and map type must correspond
+        if map_type == 'organization' and not is_org_role:
+            errors['role'] = _("For an organization map type you must specify an organization based role")
+
+        if map_type == 'team' and not is_team_role:
+            errors['role'] = _("For a team map type you must specify a team based role")
+
+        # org/team role needs organization field
+        if (is_org_role or is_team_role) and is_empty(org):
+            errors["organization"] = _("You must specify an organization with the selected role")
+
+        # team role needs team field
+        if is_team_role and is_empty(team):
+            errors["team"] = _("You must specify a team with the selected role")
+
+    except ObjectDoesNotExist:
+        errors['role'] = _("RoleDefinition {role} doesn't exist").format(role=role)
+
+    return errors

ansible_base/authentication/utils/claims.py

@@ -17,6 +17,7 @@
 from rest_framework.serializers import DateTimeField
 
 from ansible_base.authentication.models import Authenticator, AuthenticatorMap, AuthenticatorUser
+from ansible_base.authentication.utils.authenticator_map import check_role_type, expand_syntax
 from ansible_base.lib.abstract_models import AbstractOrganization, AbstractTeam, CommonModel
 from ansible_base.lib.utils.auth import get_organization_model, get_team_model
 from ansible_base.lib.utils.string import is_empty
@@ -58,15 +59,12 @@ def create_claims(authenticator: Authenticator, username: str, attrs: dict, grou
 
     # load the maps
     logger.debug(f"Authenticator ID: {authenticator.id}")
-    maps = AuthenticatorMap.objects.order_by("order")
-    logger.debug(maps)
     maps = AuthenticatorMap.objects.filter(authenticator=authenticator.id).order_by("order")
     logger.debug("==============================================================")
-    logger.debug(maps)
+    logger.debug("Processing {maps.count()} map(s) for this authenticator")
 
     for auth_map in maps:
-        logger.debug(auth_map)
-        logger.debug("++++")
+        logger.debug(f"Processing map {auth_map.name} {auth_map.id}")
         has_permission = None
         trigger_result = TriggerResult.SKIP
         allowed_keys = TRIGGER_DEFINITION.keys()
@@ -109,22 +107,43 @@ def create_claims(authenticator: Authenticator, username: str, attrs: dict, grou
 
         rule_responses.append({auth_map.id: has_permission, 'enabled': auth_map.enabled})
 
+        understood_map = False
         if auth_map.map_type == 'allow' and not has_permission:
             # If any rule does not allow we don't want to return this to true
             access_allowed = False
+            understood_map = True
         elif auth_map.map_type == 'is_superuser':
             is_superuser = has_permission
-        elif auth_map.map_type in ['team', 'role'] and not is_empty(auth_map.organization) and not is_empty(auth_map.team) and not is_empty(auth_map.role):
-            if auth_map.organization not in org_team_mapping:
-                org_team_mapping[auth_map.organization] = {}
-            org_team_mapping[auth_map.organization][auth_map.team] = has_permission
-            _add_rbac_role_mapping(has_permission, rbac_role_mapping, auth_map.role, auth_map.organization, auth_map.team)
-        elif auth_map.map_type in ['organization', 'role'] and not is_empty(auth_map.organization) and not is_empty(auth_map.role):
-            organization_membership[auth_map.organization] = has_permission
-            _add_rbac_role_mapping(has_permission, rbac_role_mapping, auth_map.role, auth_map.organization)
-        elif auth_map.map_type == 'role' and not is_empty(auth_map.role) and is_empty(auth_map.organization) and is_empty(auth_map.team):
-            _add_rbac_role_mapping(has_permission, rbac_role_mapping, auth_map.role)
-        else:
+            understood_map = True
+        elif auth_map.map_type in ['team', 'organization', 'role']:
+            # These types of maps can have expansions
+            for expanded_values in expand_syntax(attrs, auth_map):
+                expanded_organization = expanded_values.get('organization', None)
+                expanded_team = expanded_values.get('team', None)
+                expanded_role = expanded_values.get('role', None)
+
+                if (role_errors := check_role_type(map_type=auth_map.map_type, role=expanded_role, team=expanded_team, org=expanded_organization)) != {}:
+                    logger.info(f"Map type {auth_map.map_type} of rule {auth_map.name} had an invalid role type and will be skipped {role_errors}")
+                elif (
+                    auth_map.map_type in ['team', 'role']
+                    and not is_empty(expanded_organization)
+                    and not is_empty(expanded_team)
+                    and not is_empty(expanded_role)
+                ):
+                    if expanded_organization not in org_team_mapping:
+                        org_team_mapping[expanded_organization] = {}
+                    org_team_mapping[expanded_organization][expanded_team] = has_permission
+                    _add_rbac_role_mapping(has_permission, rbac_role_mapping, expanded_role, expanded_organization, expanded_team)
+                    understood_map = True
+                elif auth_map.map_type in ['organization', 'role'] and not is_empty(expanded_organization) and not is_empty(expanded_role):
+                    organization_membership[expanded_organization] = has_permission
+                    _add_rbac_role_mapping(has_permission, rbac_role_mapping, expanded_role, expanded_organization)
+                    understood_map = True
+                elif auth_map.map_type == 'role' and not is_empty(expanded_role) and is_empty(expanded_organization) and is_empty(expanded_team):
+                    _add_rbac_role_mapping(has_permission, rbac_role_mapping, expanded_role)
+                    understood_map = True
+
+        if not understood_map:
             logger.error(f"Map type {auth_map.map_type} of rule {auth_map.name} does not know how to be processed")
 
     return {