SAML fixes

ansible_base/authentication/authenticator_plugins/saml.py

@@ -296,6 +296,24 @@ def extra_data(self, user, backend, response, *args, **kwargs):
         if "Group" in attrs:
             response["Group"] = attrs["Group"]
         data = super().extra_data(user, backend, response, *args, **kwargs)
+
+        # Ideally we would always have a DB instance
+        # But if something was mocked in a test or somehow a db_instance just wasn't past in we don't want to error here
+        if self.database_instance is None:
+            return data
+
+        # The fields we want in extra get embedded into the ENABLED_IDPS field so we need to get them out from there
+        idp_data = self.database_instance.configuration.get('ENABLED_IDPS', {}).get(idp_string, {})
+
+        # We are going to auto-include all of the fields like USER_FIRST_NAME, USER_EMAIL, etc.
+        for field, attr_name in SAMLConfiguration.settings_to_enabled_idps_fields.items():
+            if field in ('IDP_URL', 'IDP_X509_CERT', 'IDP_ENTITY_ID'):
+                continue
+
+            field_name = idp_data.get(attr_name, None)
+            if field_name in attrs:
+                data[field_name] = attrs[field_name]
+
         return data
 
     def get_user_groups(self, extra_groups=[]):

ansible_base/authentication/backend.py

@@ -37,7 +37,11 @@ def authenticate(self, request, *args, **kwargs):
         last_modified = None if last_modified_item is None else last_modified_item.get('modified')
 
         for authenticator_id, authenticator_object in get_authentication_backends(last_modified).items():
-            user = authenticator_object.authenticate(request, *args, **kwargs)
+            try:
+                user = authenticator_object.authenticate(request, *args, **kwargs)
+            except Exception:
+                logger.exception(f"Exception raised while trying to authenticate with {authenticator_object.database_instance.name}")
+                continue
 
             # Social Auth pipeline can return status string when update_user_claims fails (authentication maps deny access)
             if user == SOCIAL_AUTH_PIPELINE_FAILED_STATUS:

ansible_base/authentication/utils/authentication.py

@@ -96,9 +96,13 @@ def check_system_username(uid: str) -> None:
 
 def determine_username_from_uid_social(**kwargs) -> dict:
     uid_field = getattr(kwargs.get('backend', None), 'ID_KEY', 'username')
+    if uid_field is None:
+        uid_field = 'username'
     selected_username = kwargs.get('details', {}).get(uid_field, None)
     if not selected_username:
-        raise AuthException(_('Unable to get associated username from: %(details)s') % {'details': kwargs.get("details", None)})
+        raise AuthException(
+            _('Unable to get associated username from attribute {uid_field}: %(details)s') % {'uid_field': uid_field, 'details': kwargs.get("details", None)}
+        )
 
     authenticator = kwargs.get('backend')
     if not authenticator:

test_app/tests/authentication/authenticator_plugins/test_saml.py

@@ -1,3 +1,4 @@
+from types import SimpleNamespace
 from unittest import mock
 
 import pytest
@@ -213,6 +214,91 @@ def __init__(self):
     assert "mygroup" in rDict["Group"]
 
 
+@pytest.mark.django_db
+@pytest.mark.parametrize(
+    "idp_fields,expected_results",
+    [
+        (
+            {
+                'attr_email': 'email',
+                'attr_groups': 'Groups',
+                'attr_username': 'username',
+                'attr_last_name': 'last_name',
+                'attr_first_name': 'first_name',
+                'attr_user_permanent_id': 'name_id',
+            },
+            {
+                'email': ['no.one@nowhere.com'],
+                'last_name': ['Admin'],
+                'username': ['gateway_admin'],
+                'first_name': ['Gateway'],
+                'name_id': 'gateway_admin',
+            },
+        ),
+        (
+            {
+                'attr_email': 'not_email',
+                'attr_groups': 'Groups',
+                'attr_username': 'username',
+                'attr_last_name': 'last_name',
+                'attr_first_name': 'first_name',
+                'attr_user_permanent_id': 'name_id',
+            },
+            {
+                'last_name': ['Admin'],
+                'username': ['gateway_admin'],
+                'first_name': ['Gateway'],
+                'name_id': 'gateway_admin',
+            },
+        ),
+        (
+            {
+                'attr_username': 'username',
+                'attr_last_name': 'last_name',
+                'attr_first_name': 'first_name',
+                'attr_user_permanent_id': 'name_id',
+            },
+            {
+                'last_name': ['Admin'],
+                'username': ['gateway_admin'],
+                'first_name': ['Gateway'],
+                'name_id': 'gateway_admin',
+            },
+        ),
+    ],
+)
+def test_extra_data_default_attrs(idp_fields, expected_results):
+    from ansible_base.authentication.authenticator_plugins.saml import idp_string
+    from ansible_base.authentication.models import AuthenticatorUser
+
+    ap = AuthenticatorPlugin()
+    database_instance = SimpleNamespace()
+    enabled_idps = {
+        'ENABLED_IDPS': {
+            idp_string: idp_fields,
+        }
+    }
+    database_instance.configuration = enabled_idps
+    ap.database_instance = database_instance
+
+    response = {
+        'idp_name': 'IdP',
+        'attributes': {
+            'email': ['no.one@nowhere.com'],
+            'last_name': ['Admin'],
+            'is_superuser': ['true'],
+            'username': ['gateway_admin'],
+            'first_name': ['Gateway'],
+            'Role': ['default-roles-gateway realm', 'manage-account', 'uma_authorization', 'view-profile', 'offline_access', 'manage-account-links'],
+            'name_id': 'gateway_admin',
+        },
+    }
+    au = AuthenticatorUser()
+    with mock.patch('social_core.backends.saml.SAMLAuth.extra_data', return_value={}):
+        results = ap.extra_data(None, 'IdP:gateway_admin', response, **{'social': au})
+        assert results == expected_results
+
+
 def test_saml_create_via_api_without_callback_url(admin_api_client, saml_configuration):
     del saml_configuration['CALLBACK_URL']
 

test_app/tests/authentication/test_backend.py

@@ -1,4 +1,5 @@
 from random import shuffle
+from types import SimpleNamespace
 from unittest import mock
 
 import pytest
@@ -153,3 +154,21 @@ def test_authenticate(request, local_authenticator, github_enterprise_authentica
             expected = request.getfixturevalue(expected)
 
         assert auth_return == expected
+
+
+@pytest.mark.django_db
+def test_authentication_exception(expected_log):
+    class MockAuthenticator:
+        database_instance = SimpleNamespace(name='testing')
+
+        def authenticate(*args, **kwars):
+            raise Exception('eeekkkk')
+
+    # Patch the backends to have our Mock authenticator in it
+    with mock.patch(
+        "ansible_base.authentication.backend.get_authentication_backends",
+        return_value={1: MockAuthenticator()},
+    ):
+        # Expect the log we emit
+        with expected_log('ansible_base.authentication.backend.logger', "exception", "Exception raised while trying to authenticate with"):
+            backend.AnsibleBaseAuth().authenticate(None)

test_app/tests/authentication/utils/test_authentication.py

@@ -1,3 +1,5 @@
+from types import SimpleNamespace
+
 import pytest
 from django.conf import settings
 from social_core.exceptions import AuthException
@@ -197,3 +199,17 @@ def test_determine_username_from_uid_social_happy_path(self, ldap_authenticator)
             uid="Bob",
         )
         assert response == {'username': 'Bob'}
+
+    @pytest.mark.parametrize(
+        "kwargs,expected_uid_field",
+        [
+            ({}, 'username'),
+            ({'backend': None}, 'username'),
+            ({'backend': SimpleNamespace(ID_KEY='testing')}, 'testing'),
+            ({'backend': SimpleNamespace(ID_KEY=None)}, 'testing'),
+        ],
+    )
+    def test_determine_username_from_uid_social_authenticator_ID_KEY(self, kwargs, expected_uid_field):
+        with pytest.raises(AuthException) as ae:
+            authentication.determine_username_from_uid_social(**kwargs)
+            assert f'Unable to get associated username from attribute {expected_uid_field}' in ae