External-Mirrors/awx
2
0
Fork 0
mirror of https://github.com/ansible/awx.git synced 2026-04-25 19:55:24 -02:30
Code Issues Packages Projects Releases Wiki Activity

Remove SAML authentication (#15568)

Browse Source 
* remove saml

* remove license file and management command

* update requirements, add migrations

* remove unused imports
This commit is contained in:
jessicamack
2024-10-02 12:47:08 -04:00
parent bf09b95b61
commit 1ca034b0a7
35 changed files with 76 additions and 2439 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
awx/api/conf.py
View File

@@ -37,7 +37,7 @@ register(
label=_('Disable the built-in authentication system'),
help_text=_(
"Controls whether users are prevented from using the built-in authentication system. "
"You probably want to do this if you are using an LDAP or SAML integration."
"You probably want to do this if you are using an LDAP integration."
),
category=_('Authentication'),
category_slug='authentication',
@@ -77,8 +77,8 @@ register(
default=False,
label=_('Allow External Users to Create OAuth2 Tokens'),
help_text=_(
'For security reasons, users from external auth providers (LDAP, SAML, '
'SSO, and others) are not allowed to create OAuth2 tokens. '
'For security reasons, users from external auth providers (LDAP, SSO, '
' and others) are not allowed to create OAuth2 tokens. '
'To change this behavior, enable this setting. Existing tokens will '
'not be deleted when this setting is toggled off.'
),

18
awx/api/views/__init__.py
View File

@@ -689,25 +689,15 @@ class AuthView(APIView):
data = OrderedDict()
err_backend, err_message = request.session.get('social_auth_error', (None, None))
auth_backends = list(load_backends(settings.AUTHENTICATION_BACKENDS, force_load=True).items())
# Return auth backends in consistent order: oidc, saml.
# Return auth backends in consistent order: oidc.
auth_backends.sort(key=lambda x: x[0])
for name, backend in auth_backends:
login_url = reverse('social:begin', args=(name,))
complete_url = request.build_absolute_uri(reverse('social:complete', args=(name,)))
backend_data = {'login_url': login_url, 'complete_url': complete_url}
if name == 'saml':
backend_data['metadata_url'] = reverse('sso:saml_metadata')
for idp in sorted(settings.SOCIAL_AUTH_SAML_ENABLED_IDPS.keys()):
saml_backend_data = dict(backend_data.items())
saml_backend_data['login_url'] = '%s?idp=%s' % (login_url, idp)
full_backend_name = '%s:%s' % (name, idp)
if (err_backend == full_backend_name or err_backend == name) and err_message:
saml_backend_data['error'] = err_message
data[full_backend_name] = saml_backend_data
else:
if err_backend == name and err_message:
backend_data['error'] = err_message
data[name] = backend_data
if err_backend == name and err_message:
backend_data['error'] = err_message
data[name] = backend_data
return Response(data)

40
awx/conf/migrations/0011_remove_saml_auth_conf.py Normal file
View File

@@ -0,0 +1,40 @@
# Generated by Django 4.2.10 on 2024-08-27 14:20
from django.db import migrations
SAML_AUTH_CONF_KEYS = [
'SAML_AUTO_CREATE_OBJECTS',
'SOCIAL_AUTH_SAML_CALLBACK_URL',
'SOCIAL_AUTH_SAML_METADATA_URL',
'SOCIAL_AUTH_SAML_SP_ENTITY_ID',
'SOCIAL_AUTH_SAML_SP_PUBLIC_CERT',
'SOCIAL_AUTH_SAML_SP_PRIVATE_KEY',
'SOCIAL_AUTH_SAML_ORG_INFO',
'SOCIAL_AUTH_SAML_TECHNICAL_CONTACT',
'SOCIAL_AUTH_SAML_SUPPORT_CONTACT',
'SOCIAL_AUTH_SAML_ENABLED_IDPS',
'SOCIAL_AUTH_SAML_SECURITY_CONFIG',
'SOCIAL_AUTH_SAML_SP_EXTRA',
'SOCIAL_AUTH_SAML_EXTRA_DATA',
'SOCIAL_AUTH_SAML_ORGANIZATION_MAP',
'SOCIAL_AUTH_SAML_TEAM_MAP',
'SOCIAL_AUTH_SAML_ORGANIZATION_ATTR',
'SOCIAL_AUTH_SAML_TEAM_ATTR',
'SOCIAL_AUTH_SAML_USER_FLAGS_BY_ATTR',
]
def remove_saml_auth_conf(apps, scheme_editor):
setting = apps.get_model('conf', 'Setting')
setting.objects.filter(key__in=SAML_AUTH_CONF_KEYS).delete()
class Migration(migrations.Migration):
dependencies = [
('conf', '0010_change_to_JSONField'),
]
operations = [
migrations.RunPython(remove_saml_auth_conf),
]

19
awx/conf/tests/functional/test_api.py
View File

@@ -8,7 +8,6 @@ from awx.main.utils.encryption import decrypt_field
from awx.conf import fields
from awx.conf.registry import settings_registry
from awx.conf.models import Setting
from awx.sso import fields as sso_fields
@pytest.fixture
@@ -103,24 +102,6 @@ def test_setting_singleton_update(api_request, dummy_setting):
assert response.data['FOO_BAR'] == 4
@pytest.mark.django_db
def test_setting_singleton_update_hybriddictfield_with_forbidden(api_request, dummy_setting):
# Some HybridDictField subclasses have a child of _Forbidden,
# indicating that only the defined fields can be filled in. Make
# sure that the _Forbidden validator doesn't get used for the
# fields. See also https://github.com/ansible/awx/issues/4099.
with dummy_setting('FOO_BAR', field_class=sso_fields.SAMLOrgAttrField, category='FooBar', category_slug='foobar'), mock.patch(
'awx.conf.views.clear_setting_cache'
):
api_request(
'patch',
reverse('api:setting_singleton_detail', kwargs={'category_slug': 'foobar'}),
data={'FOO_BAR': {'saml_admin_attr': 'Admins', 'saml_attr': 'Orgs'}},
)
response = api_request('get', reverse('api:setting_singleton_detail', kwargs={'category_slug': 'foobar'}))
assert response.data['FOO_BAR'] == {'saml_admin_attr': 'Admins', 'saml_attr': 'Orgs'}
@pytest.mark.django_db
def test_setting_singleton_update_dont_change_readonly_fields(api_request, dummy_setting):
with dummy_setting('FOO_BAR', field_class=fields.IntegerField, read_only=True, default=4, category='FooBar', category_slug='foobar'), mock.patch(

2
awx/main/conf.py
View File

@@ -48,7 +48,7 @@ register(
label=_('Organization Admins Can Manage Users and Teams'),
help_text=_(
'Controls whether any Organization Admin has the privileges to create and manage users and teams. '
'You may want to disable this ability if you are using an LDAP or SAML integration.'
'You may want to disable this ability if you are using an LDAP integration.'
),
category=_('System'),
category_slug='system',

128
awx/main/management/commands/dump_auth_config.py
View File

@@ -1,128 +0,0 @@
import json
import os
import sys
from typing import Any
from django.core.management.base import BaseCommand
from django.conf import settings
from awx.conf import settings_registry
class Command(BaseCommand):
help = 'Dump the current auth configuration in django_ansible_base.authenticator format, currently supports SAML'
DAB_SAML_AUTHENTICATOR_KEYS = {
"SP_ENTITY_ID": True,
"SP_PUBLIC_CERT": True,
"SP_PRIVATE_KEY": True,
"ORG_INFO": True,
"TECHNICAL_CONTACT": True,
"SUPPORT_CONTACT": True,
"SP_EXTRA": False,
"SECURITY_CONFIG": False,
"EXTRA_DATA": False,
"ENABLED_IDPS": True,
"CALLBACK_URL": False,
}
def is_enabled(self, settings, keys):
missing_fields = []
for key, required in keys.items():
if required and not settings.get(key):
missing_fields.append(key)
if missing_fields:
return False, missing_fields
return True, None
def get_awx_saml_settings(self) -> dict[str, Any]:
awx_saml_settings = {}
for awx_saml_setting in settings_registry.get_registered_settings(category_slug='saml'):
awx_saml_settings[awx_saml_setting.removeprefix("SOCIAL_AUTH_SAML_")] = getattr(settings, awx_saml_setting, None)
return awx_saml_settings
def format_config_data(self, enabled, awx_settings, type, keys, name):
config = {
"type": f"ansible_base.authentication.authenticator_plugins.{type}",
"name": name,
"enabled": enabled,
"create_objects": True,
"users_unique": False,
"remove_users": True,
"configuration": {},
}
for k in keys:
v = awx_settings.get(k)
config["configuration"].update({k: v})
if type == "saml":
idp_to_key_mapping = {
"url": "IDP_URL",
"x509cert": "IDP_X509_CERT",
"entity_id": "IDP_ENTITY_ID",
"attr_email": "IDP_ATTR_EMAIL",
"attr_groups": "IDP_GROUPS",
"attr_username": "IDP_ATTR_USERNAME",
"attr_last_name": "IDP_ATTR_LAST_NAME",
"attr_first_name": "IDP_ATTR_FIRST_NAME",
"attr_user_permanent_id": "IDP_ATTR_USER_PERMANENT_ID",
}
for idp_name in awx_settings.get("ENABLED_IDPS", {}):
for key in idp_to_key_mapping:
value = awx_settings["ENABLED_IDPS"][idp_name].get(key)
if value is not None:
config["name"] = idp_name
config["configuration"].update({idp_to_key_mapping[key]: value})
return config
def add_arguments(self, parser):
parser.add_argument(
"output_file",
nargs="?",
type=str,
default=None,
help="Output JSON file path",
)
def handle(self, *args, **options):
try:
data = []
# dump SAML settings
awx_saml_settings = self.get_awx_saml_settings()
awx_saml_enabled, saml_missing_fields = self.is_enabled(awx_saml_settings, self.DAB_SAML_AUTHENTICATOR_KEYS)
if awx_saml_enabled:
awx_saml_name = awx_saml_settings["ENABLED_IDPS"]
data.append(
self.format_config_data(
awx_saml_enabled,
awx_saml_settings,
"saml",
self.DAB_SAML_AUTHENTICATOR_KEYS,
awx_saml_name,
)
)
else:
data.append({"SAML_missing_fields": saml_missing_fields})
# write to file if requested
if options["output_file"]:
# Define the path for the output JSON file
output_file = options["output_file"]
# Ensure the directory exists
os.makedirs(os.path.dirname(output_file), exist_ok=True)
# Write data to the JSON file
with open(output_file, "w") as f:
json.dump(data, f, indent=4)
self.stdout.write(self.style.SUCCESS(f"Auth config data dumped to {output_file}"))
else:
self.stdout.write(json.dumps(data, indent=4))
except Exception as e:
self.stdout.write(self.style.ERROR(f"An error occurred: {str(e)}"))
sys.exit(1)

2
awx/main/models/__init__.py
View File

@@ -248,8 +248,6 @@ def user_is_in_enterprise_category(user, category):
ret = (category,) in user.enterprise_auth.values_list('provider') and not user.has_usable_password()
# NOTE: this if block ensures existing enterprise users are still able to
# log in. Remove it in a future release
if category == 'saml':
ret = ret or user.social_auth.all()
return ret

28
awx/main/tests/functional/api/test_settings.py
View File

@@ -193,31 +193,3 @@ def test_logging_aggregator_connection_test_valid(put, post, admin):
# "Test" the logger
url = reverse('api:setting_logging_test')
post(url, {}, user=admin, expect=202)
@pytest.mark.django_db
@pytest.mark.parametrize('headers', [True, False])
def test_saml_x509cert_validation(patch, get, admin, headers):
cert = "MIIEogIBAAKCAQEA1T4za6qBbHxFpN5f9eFvA74MFjrsjcp1uvzOaE23AYKMDEJghJ6dqQ7GwHLNIeIeumqDFmODauIzrgSDJTT5+NG30Rr+rRi0zDkrkBAj/AtA+SaVhbzqB6ZSd7LaMly9XAc+82OKlNpuWS9hPmFaSShzDTXRu5RRyvm4NDCAOGDu5hyVR2pV/ffKDNfNkChnqzvRRW9laQcVmliZhlTGn7nPZ+JbjpwEy0nwW+4zoAiEvwnT52N4xTqIcYOnXtGiaf13dh7FkUfYmS0tzF3+h8QRKwtIm4y+sq84R/kr79/0t5aRUpJynNrECajzmArpL4IjXKTPIyUpTKirJgGnCwIDAQABAoIBAC6bbbm2hpsjfkVOpUKkhxMWUqX5MwK6oYjBAIwjkEAwPFPhnh7eXC87H42oidVCCt1LsmMOVQbjcdAzBEb5kTkk/Twi3k8O+1U3maHfJT5NZ2INYNjeNXh+jb/Dw5UGWAzpOIUR2JQ4Oa4cgPCVbppW0O6uOKz6+fWXJv+hKiUoBCC0TiY52iseHJdUOaKNxYRD2IyIzCAxFSd5tZRaARIYDsugXp3E/TdbsVWA7bmjIBOXq+SquTrlB8x7j3B7+Pi09nAJ2U/uV4PHE+/2Fl009ywfmqancvnhwnz+GQ5jjP+gTfghJfbO+Z6M346rS0Vw+osrPgfyudNHlCswHOECgYEA/Cfq25gDP07wo6+wYWbx6LIzj/SSZy/Ux9P8zghQfoZiPoaq7BQBPAzwLNt7JWST8U11LZA8/wo6ch+HSTMk+m5ieVuru2cHxTDqeNlh94eCrNwPJ5ayA5U6LxAuSCTAzp+rv6KQUx1JcKSEHuh+nRYTKvUDE6iA6YtPLO96lLUCgYEA2H5rOPX2M4w1Q9zjol77lplbPRdczXNd0PIzhy8Z2ID65qvmr1nxBG4f2H96ykW8CKLXNvSXreNZ1BhOXc/3Hv+3mm46iitB33gDX4mlV4Jyo/w5IWhUKRyoW6qXquFFsScxRzTrx/9M+aZeRRLdsBk27HavFEg6jrbQ0SleZL8CgYAaM6Op8d/UgkVrHOR9Go9kmK/W85kK8+NuaE7Ksf57R0eKK8AzC9kc/lMuthfTyOG+n0ff1i8gaVWtai1Ko+/hvfqplacAsDIUgYK70AroB8LCZ5ODj5sr2CPVpB7LDFakod7c6O2KVW6+L7oy5AHUHOkc+5y4PDg5DGrLxo68SQKBgAlGoWF3aG0c/MtDk51JZI43U+lyLs++ua5SMlMAeaMFI7rucpvgxqrh7Qthqukvw7a7A22fXUBeFWM5B2KNnpD9c+hyAKAa6l+gzMQzKZpuRGsyS2BbEAAS8kO7M3Rm4o2MmFfstI2FKs8nibJ79HOvIONQ0n+T+K5Utu2/UAQRAoGAFB4fiIyQ0nYzCf18Z4Wvi/qeIOW+UoBonIN3y1h4wruBywINHxFMHx4aVImJ6R09hoJ9D3Mxli3xF/8JIjfTG5fBSGrGnuofl14d/XtRDXbT2uhVXrIkeLL/ojODwwEx0VhxIRUEjPTvEl6AFSRRcBp3KKzQ/cu7ENDY6GTlOUI=" # noqa
if headers:
cert = '-----BEGIN CERTIFICATE-----\n' + cert + '\n-----END CERTIFICATE-----'
url = reverse('api:setting_singleton_detail', kwargs={'category_slug': 'saml'})
resp = patch(
url,
user=admin,
data={
'SOCIAL_AUTH_SAML_ENABLED_IDPS': {
"okta": {
"attr_last_name": "LastName",
"attr_username": "login",
"entity_id": "http://www.okta.com/abc123",
"attr_user_permanent_id": "login",
"url": "https://example.okta.com/app/abc123/xyz123/sso/saml",
"attr_email": "Email",
"x509cert": cert,
"attr_first_name": "FirstName",
}
}
},
)
assert resp.status_code == 200

89
awx/main/tests/unit/commands/test_dump_auth_config.py
View File

@@ -1,89 +0,0 @@
from io import StringIO
import json
from django.core.management import call_command
from django.test import TestCase, override_settings
settings_dict = {
"SOCIAL_AUTH_SAML_SP_ENTITY_ID": "SP_ENTITY_ID",
"SOCIAL_AUTH_SAML_SP_PUBLIC_CERT": "SP_PUBLIC_CERT",
"SOCIAL_AUTH_SAML_SP_PRIVATE_KEY": "SP_PRIVATE_KEY",
"SOCIAL_AUTH_SAML_ORG_INFO": "ORG_INFO",
"SOCIAL_AUTH_SAML_TECHNICAL_CONTACT": "TECHNICAL_CONTACT",
"SOCIAL_AUTH_SAML_SUPPORT_CONTACT": "SUPPORT_CONTACT",
"SOCIAL_AUTH_SAML_SP_EXTRA": "SP_EXTRA",
"SOCIAL_AUTH_SAML_SECURITY_CONFIG": "SECURITY_CONFIG",
"SOCIAL_AUTH_SAML_EXTRA_DATA": "EXTRA_DATA",
"SOCIAL_AUTH_SAML_ENABLED_IDPS": {
"Keycloak": {
"attr_last_name": "last_name",
"attr_groups": "groups",
"attr_email": "email",
"attr_user_permanent_id": "name_id",
"attr_username": "username",
"entity_id": "https://example.com/auth/realms/awx",
"url": "https://example.com/auth/realms/awx/protocol/saml",
"x509cert": "-----BEGIN CERTIFICATE-----\nMIIDDjCCAfYCCQCPBeVvpo8+VzANBgkqhkiG9w0BAQsFADBJMQswCQYDVQQGEwJV\nUzELMAkGA1UECAwCTkMxDzANBgNVBAcMBkR1cmhhbTEMMAoGA1UECgwDYXd4MQ4w\nDAYDVQQDDAVsb2NhbDAeFw0yNDAxMTgxNDA4MzFaFw0yNTAxMTcxNDA4MzFaMEkx\nCzAJBgNVBAYTAlVTMQswCQYDVQQIDAJOQzEPMA0GA1UEBwwGRHVyaGFtMQwwCgYD\nVQQKDANhd3gxDjAMBgNVBAMMBWxvY2FsMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8A\nMIIBCgKCAQEAzouj93oyFXsHEABdPESh3CYpp5QJJBM4TLYIIolk6PFOFIVwBuFY\nfExi5w7Hh4A42lPM6RkrT+u3h7LV39H9MRUfqygOSmaxICTOI0sU9ROHc44fWWzN\n756OP4B5zSiqG82q8X7nYVkcID+2F/3ekPLMOlWn53OrcdfKKDIcqavoTkQJefc2\nggXU3WgVCxGki/qCm+e5cZ1Cpl/ykSLOT8dWMEzDd12kin66zJ3KYz9F2Q5kQTh4\nKRAChnBBoEqzOfENHEAaHALiXOlVSy61VcLbtvskRMMwBtsydlnd9n/HGnktgrid\n3Ca0z5wBTHWjAOBvCKxKJuDa+jmyHEnpcQIDAQABMA0GCSqGSIb3DQEBCwUAA4IB\nAQBXvmyPWgXhC26cHYJBgQqj57dZ+n7p00kM1J+27oDMjGmbmX+XIKXLWazw/rG3\ngDjw9MXI2tVCrQMX0ohjphaULXhb/VBUPDOiW+k7C6AB3nZySFRflcR3cM4f83zF\nMoBd0549h5Red4p72FeOKNJRTN8YO4ooH9YNh5g0FQkgqn7fV9w2CNlomeKIW9zP\nm8tjFw0cJUk2wEYBVl8O7ko5rgNlzhkLoZkMvJhKa99AQJA6MAdyoLl1lv56Kq4X\njk+mMEiz9SaInp+ILQ1uQxZEwuC7DoGRW76rV4Fnie6+DLft4WKZfX1497mx8NV3\noR0abutJaKnCj07dwRu4/EsK\n-----END CERTIFICATE-----",
"attr_first_name": "first_name",
}
},
"SOCIAL_AUTH_SAML_CALLBACK_URL": "CALLBACK_URL",
}
@override_settings(**settings_dict)
class TestDumpAuthConfigCommand(TestCase):
def setUp(self):
super().setUp()
self.expected_config = [
{
"type": "ansible_base.authentication.authenticator_plugins.saml",
"name": "Keycloak",
"enabled": True,
"create_objects": True,
"users_unique": False,
"remove_users": True,
"configuration": {
"SP_ENTITY_ID": "SP_ENTITY_ID",
"SP_PUBLIC_CERT": "SP_PUBLIC_CERT",
"SP_PRIVATE_KEY": "SP_PRIVATE_KEY",
"ORG_INFO": "ORG_INFO",
"TECHNICAL_CONTACT": "TECHNICAL_CONTACT",
"SUPPORT_CONTACT": "SUPPORT_CONTACT",
"SP_EXTRA": "SP_EXTRA",
"SECURITY_CONFIG": "SECURITY_CONFIG",
"EXTRA_DATA": "EXTRA_DATA",
"ENABLED_IDPS": {
"Keycloak": {
"attr_last_name": "last_name",
"attr_groups": "groups",
"attr_email": "email",
"attr_user_permanent_id": "name_id",
"attr_username": "username",
"entity_id": "https://example.com/auth/realms/awx",
"url": "https://example.com/auth/realms/awx/protocol/saml",
"x509cert": "-----BEGIN CERTIFICATE-----\nMIIDDjCCAfYCCQCPBeVvpo8+VzANBgkqhkiG9w0BAQsFADBJMQswCQYDVQQGEwJV\nUzELMAkGA1UECAwCTkMxDzANBgNVBAcMBkR1cmhhbTEMMAoGA1UECgwDYXd4MQ4w\nDAYDVQQDDAVsb2NhbDAeFw0yNDAxMTgxNDA4MzFaFw0yNTAxMTcxNDA4MzFaMEkx\nCzAJBgNVBAYTAlVTMQswCQYDVQQIDAJOQzEPMA0GA1UEBwwGRHVyaGFtMQwwCgYD\nVQQKDANhd3gxDjAMBgNVBAMMBWxvY2FsMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8A\nMIIBCgKCAQEAzouj93oyFXsHEABdPESh3CYpp5QJJBM4TLYIIolk6PFOFIVwBuFY\nfExi5w7Hh4A42lPM6RkrT+u3h7LV39H9MRUfqygOSmaxICTOI0sU9ROHc44fWWzN\n756OP4B5zSiqG82q8X7nYVkcID+2F/3ekPLMOlWn53OrcdfKKDIcqavoTkQJefc2\nggXU3WgVCxGki/qCm+e5cZ1Cpl/ykSLOT8dWMEzDd12kin66zJ3KYz9F2Q5kQTh4\nKRAChnBBoEqzOfENHEAaHALiXOlVSy61VcLbtvskRMMwBtsydlnd9n/HGnktgrid\n3Ca0z5wBTHWjAOBvCKxKJuDa+jmyHEnpcQIDAQABMA0GCSqGSIb3DQEBCwUAA4IB\nAQBXvmyPWgXhC26cHYJBgQqj57dZ+n7p00kM1J+27oDMjGmbmX+XIKXLWazw/rG3\ngDjw9MXI2tVCrQMX0ohjphaULXhb/VBUPDOiW+k7C6AB3nZySFRflcR3cM4f83zF\nMoBd0549h5Red4p72FeOKNJRTN8YO4ooH9YNh5g0FQkgqn7fV9w2CNlomeKIW9zP\nm8tjFw0cJUk2wEYBVl8O7ko5rgNlzhkLoZkMvJhKa99AQJA6MAdyoLl1lv56Kq4X\njk+mMEiz9SaInp+ILQ1uQxZEwuC7DoGRW76rV4Fnie6+DLft4WKZfX1497mx8NV3\noR0abutJaKnCj07dwRu4/EsK\n-----END CERTIFICATE-----",
"attr_first_name": "first_name",
}
},
"CALLBACK_URL": "CALLBACK_URL",
"IDP_URL": "https://example.com/auth/realms/awx/protocol/saml",
"IDP_X509_CERT": "-----BEGIN CERTIFICATE-----\nMIIDDjCCAfYCCQCPBeVvpo8+VzANBgkqhkiG9w0BAQsFADBJMQswCQYDVQQGEwJV\nUzELMAkGA1UECAwCTkMxDzANBgNVBAcMBkR1cmhhbTEMMAoGA1UECgwDYXd4MQ4w\nDAYDVQQDDAVsb2NhbDAeFw0yNDAxMTgxNDA4MzFaFw0yNTAxMTcxNDA4MzFaMEkx\nCzAJBgNVBAYTAlVTMQswCQYDVQQIDAJOQzEPMA0GA1UEBwwGRHVyaGFtMQwwCgYD\nVQQKDANhd3gxDjAMBgNVBAMMBWxvY2FsMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8A\nMIIBCgKCAQEAzouj93oyFXsHEABdPESh3CYpp5QJJBM4TLYIIolk6PFOFIVwBuFY\nfExi5w7Hh4A42lPM6RkrT+u3h7LV39H9MRUfqygOSmaxICTOI0sU9ROHc44fWWzN\n756OP4B5zSiqG82q8X7nYVkcID+2F/3ekPLMOlWn53OrcdfKKDIcqavoTkQJefc2\nggXU3WgVCxGki/qCm+e5cZ1Cpl/ykSLOT8dWMEzDd12kin66zJ3KYz9F2Q5kQTh4\nKRAChnBBoEqzOfENHEAaHALiXOlVSy61VcLbtvskRMMwBtsydlnd9n/HGnktgrid\n3Ca0z5wBTHWjAOBvCKxKJuDa+jmyHEnpcQIDAQABMA0GCSqGSIb3DQEBCwUAA4IB\nAQBXvmyPWgXhC26cHYJBgQqj57dZ+n7p00kM1J+27oDMjGmbmX+XIKXLWazw/rG3\ngDjw9MXI2tVCrQMX0ohjphaULXhb/VBUPDOiW+k7C6AB3nZySFRflcR3cM4f83zF\nMoBd0549h5Red4p72FeOKNJRTN8YO4ooH9YNh5g0FQkgqn7fV9w2CNlomeKIW9zP\nm8tjFw0cJUk2wEYBVl8O7ko5rgNlzhkLoZkMvJhKa99AQJA6MAdyoLl1lv56Kq4X\njk+mMEiz9SaInp+ILQ1uQxZEwuC7DoGRW76rV4Fnie6+DLft4WKZfX1497mx8NV3\noR0abutJaKnCj07dwRu4/EsK\n-----END CERTIFICATE-----",
"IDP_ENTITY_ID": "https://example.com/auth/realms/awx",
"IDP_ATTR_EMAIL": "email",
"IDP_GROUPS": "groups",
"IDP_ATTR_USERNAME": "username",
"IDP_ATTR_LAST_NAME": "last_name",
"IDP_ATTR_FIRST_NAME": "first_name",
"IDP_ATTR_USER_PERMANENT_ID": "name_id",
},
},
]
def test_json_returned_from_cmd(self):
output = StringIO()
call_command("dump_auth_config", stdout=output)
cmmd_output = json.loads(output.getvalue())
# check configured SAML return
assert cmmd_output[0] == self.expected_config[0]

18
awx/settings/defaults.py
View File

@@ -392,8 +392,6 @@ REST_FRAMEWORK = {
}
AUTHENTICATION_BACKENDS = (
'social_core.backends.open_id_connect.OpenIdConnectAuth',
'awx.sso.backends.SAMLAuth',
'awx.main.backends.AWXModelBackend',
)
@@ -488,13 +486,12 @@ _SOCIAL_AUTH_PIPELINE_BASE = (
'social_core.pipeline.user.user_details',
'awx.sso.social_base_pipeline.prevent_inactive_login',
)
SOCIAL_AUTH_PIPELINE = _SOCIAL_AUTH_PIPELINE_BASE + (
'awx.sso.social_pipeline.update_user_orgs',
'awx.sso.social_pipeline.update_user_teams',
'ansible_base.resource_registry.utils.service_backed_sso_pipeline.redirect_to_resource_server',
)
SOCIAL_AUTH_SAML_PIPELINE = _SOCIAL_AUTH_PIPELINE_BASE + ('awx.sso.saml_pipeline.populate_user', 'awx.sso.saml_pipeline.update_user_flags')
SAML_AUTO_CREATE_OBJECTS = True
SOCIAL_AUTH_LOGIN_URL = '/'
SOCIAL_AUTH_LOGIN_REDIRECT_URL = '/sso/complete/'
@@ -509,19 +506,6 @@ SOCIAL_AUTH_CLEAN_USERNAMES = True
SOCIAL_AUTH_SANITIZE_REDIRECTS = True
SOCIAL_AUTH_REDIRECT_IS_HTTPS = False
# Note: These settings may be overridden by database settings.
SOCIAL_AUTH_SAML_SP_ENTITY_ID = ''
SOCIAL_AUTH_SAML_SP_PUBLIC_CERT = ''
SOCIAL_AUTH_SAML_SP_PRIVATE_KEY = ''
SOCIAL_AUTH_SAML_ORG_INFO = {}
SOCIAL_AUTH_SAML_TECHNICAL_CONTACT = {}
SOCIAL_AUTH_SAML_SUPPORT_CONTACT = {}
SOCIAL_AUTH_SAML_ENABLED_IDPS = {}
SOCIAL_AUTH_SAML_ORGANIZATION_ATTR = {}
SOCIAL_AUTH_SAML_TEAM_ATTR = {}
SOCIAL_AUTH_SAML_USER_FLAGS_BY_ATTR = {}
# Any ANSIBLE_* settings will be passed to the task runner subprocess
# environment

93
awx/sso/backends.py
View File

@@ -8,11 +8,6 @@ import logging
from django.contrib.auth.models import User
from django.conf import settings as django_settings
# social
from social_core.backends.saml import OID_USERID
from social_core.backends.saml import SAMLAuth as BaseSAMLAuth
from social_core.backends.saml import SAMLIdentityProvider as BaseSAMLIdentityProvider
# Ansible Tower
from awx.sso.models import UserEnterpriseAuth
@@ -38,91 +33,3 @@ def _get_or_set_enterprise_user(username, password, provider):
if created or user.is_in_enterprise_category(provider):
return user
logger.warning("Enterprise user %s already defined in Tower." % username)
class TowerSAMLIdentityProvider(BaseSAMLIdentityProvider):
"""
Custom Identity Provider to make attributes to what we expect.
"""
def get_user_permanent_id(self, attributes):
uid = attributes[self.conf.get('attr_user_permanent_id', OID_USERID)]
if isinstance(uid, str):
return uid
return uid[0]
def get_attr(self, attributes, conf_key, default_attribute):
"""
Get the attribute 'default_attribute' out of the attributes,
unless self.conf[conf_key] overrides the default by specifying
another attribute to use.
"""
key = self.conf.get(conf_key, default_attribute)
value = attributes[key] if key in attributes else None
# In certain implementations (like https://pagure.io/ipsilon) this value is a string, not a list
if isinstance(value, (list, tuple)):
value = value[0]
if conf_key in ('attr_first_name', 'attr_last_name', 'attr_username', 'attr_email') and value is None:
logger.warning(
"Could not map user detail '%s' from SAML attribute '%s'; update SOCIAL_AUTH_SAML_ENABLED_IDPS['%s']['%s'] with the correct SAML attribute.",
conf_key[5:],
key,
self.name,
conf_key,
)
return str(value) if value is not None else value
class SAMLAuth(BaseSAMLAuth):
"""
Custom SAMLAuth backend to verify license status
"""
def get_idp(self, idp_name):
idp_config = self.setting('ENABLED_IDPS')[idp_name]
return TowerSAMLIdentityProvider(idp_name, **idp_config)
def authenticate(self, request, *args, **kwargs):
if not all(
[
django_settings.SOCIAL_AUTH_SAML_SP_ENTITY_ID,
django_settings.SOCIAL_AUTH_SAML_SP_PUBLIC_CERT,
django_settings.SOCIAL_AUTH_SAML_SP_PRIVATE_KEY,
django_settings.SOCIAL_AUTH_SAML_ORG_INFO,
django_settings.SOCIAL_AUTH_SAML_TECHNICAL_CONTACT,
django_settings.SOCIAL_AUTH_SAML_SUPPORT_CONTACT,
django_settings.SOCIAL_AUTH_SAML_ENABLED_IDPS,
]
):
return None
pipeline_result = super(SAMLAuth, self).authenticate(request, *args, **kwargs)
if isinstance(pipeline_result, HttpResponse):
return pipeline_result
else:
user = pipeline_result
# Comes from https://github.com/omab/python-social-auth/blob/v0.2.21/social/backends/base.py#L91
if getattr(user, 'is_new', False):
enterprise_auth = _decorate_enterprise_user(user, 'saml')
logger.debug("Created enterprise user %s from %s backend." % (user.username, enterprise_auth.get_provider_display()))
elif user and not user.is_in_enterprise_category('saml'):
return None
if user:
logger.debug("Enterprise user %s already created in Tower." % user.username)
return user
def get_user(self, user_id):
if not all(
[
django_settings.SOCIAL_AUTH_SAML_SP_ENTITY_ID,
django_settings.SOCIAL_AUTH_SAML_SP_PUBLIC_CERT,
django_settings.SOCIAL_AUTH_SAML_SP_PRIVATE_KEY,
django_settings.SOCIAL_AUTH_SAML_ORG_INFO,
django_settings.SOCIAL_AUTH_SAML_TECHNICAL_CONTACT,
django_settings.SOCIAL_AUTH_SAML_SUPPORT_CONTACT,
django_settings.SOCIAL_AUTH_SAML_ENABLED_IDPS,
]
):
return None
return super(SAMLAuth, self).get_user(user_id)

8
awx/sso/common.py
View File

@@ -186,12 +186,10 @@ def get_external_account(user):
def is_remote_auth_enabled():
from django.conf import settings
settings_that_turn_on_remote_auth = [
'SOCIAL_AUTH_SAML_ENABLED_IDPS',
]
# Also include any SOCAIL_AUTH_*KEY (except SAML)
settings_that_turn_on_remote_auth = []
# Also include any SOCAIL_AUTH_*KEY
for social_auth_key in dir(settings):
if social_auth_key.startswith('SOCIAL_AUTH_') and social_auth_key.endswith('_KEY') and 'SAML' not in social_auth_key:
if social_auth_key.startswith('SOCIAL_AUTH_') and social_auth_key.endswith('_KEY'):
settings_that_turn_on_remote_auth.append(social_auth_key)
return any(getattr(settings, s, None) for s in settings_that_turn_on_remote_auth)

330
awx/sso/conf.py
View File

@@ -11,17 +11,9 @@ from django.utils.translation import gettext_lazy as _
from awx.conf import register, fields
from awx.sso.fields import (
AuthenticationBackendsField,
SAMLContactField,
SAMLEnabledIdPsField,
SAMLOrgAttrField,
SAMLOrgInfoField,
SAMLSecurityField,
SAMLTeamAttrField,
SAMLUserFlagsAttrField,
SocialOrganizationMapField,
SocialTeamMapField,
)
from awx.main.validators import validate_private_key, validate_certificate
class SocialAuthCallbackURL(object):
@@ -143,328 +135,6 @@ if settings.ALLOW_LOCAL_RESOURCE_MANAGEMENT:
category_slug='authentication',
)
###############################################################################
# SAML AUTHENTICATION SETTINGS
###############################################################################
def get_saml_metadata_url():
return urlparse.urljoin(settings.TOWER_URL_BASE, reverse('sso:saml_metadata'))
def get_saml_entity_id():
return settings.TOWER_URL_BASE
register(
'SAML_AUTO_CREATE_OBJECTS',
field_class=fields.BooleanField,
default=True,
label=_('Automatically Create Organizations and Teams on SAML Login'),
help_text=_('When enabled (the default), mapped Organizations and Teams will be created automatically on successful SAML login.'),
category=_('SAML'),
category_slug='saml',
)
register(
'SOCIAL_AUTH_SAML_CALLBACK_URL',
field_class=fields.CharField,
read_only=True,
default=SocialAuthCallbackURL('saml'),
label=_('SAML Assertion Consumer Service (ACS) URL'),
help_text=_(
'Register the service as a service provider (SP) with each identity '
'provider (IdP) you have configured. Provide your SP Entity ID '
'and this ACS URL for your application.'
),
category=_('SAML'),
category_slug='saml',
depends_on=['TOWER_URL_BASE'],
)
register(
'SOCIAL_AUTH_SAML_METADATA_URL',
field_class=fields.CharField,
read_only=True,
default=get_saml_metadata_url,
label=_('SAML Service Provider Metadata URL'),
help_text=_('If your identity provider (IdP) allows uploading an XML metadata file, you can download one from this URL.'),
category=_('SAML'),
category_slug='saml',
)
register(
'SOCIAL_AUTH_SAML_SP_ENTITY_ID',
field_class=fields.CharField,
allow_blank=True,
default=get_saml_entity_id,
label=_('SAML Service Provider Entity ID'),
help_text=_(
'The application-defined unique identifier used as the '
'audience of the SAML service provider (SP) configuration. '
'This is usually the URL for the service.'
),
category=_('SAML'),
category_slug='saml',
depends_on=['TOWER_URL_BASE'],
)
register(
'SOCIAL_AUTH_SAML_SP_PUBLIC_CERT',
field_class=fields.CharField,
allow_blank=True,
validators=[validate_certificate],
label=_('SAML Service Provider Public Certificate'),
help_text=_('Create a keypair to use as a service provider (SP) and include the certificate content here.'),
category=_('SAML'),
category_slug='saml',
)
register(
'SOCIAL_AUTH_SAML_SP_PRIVATE_KEY',
field_class=fields.CharField,
allow_blank=True,
validators=[validate_private_key],
label=_('SAML Service Provider Private Key'),
help_text=_('Create a keypair to use as a service provider (SP) and include the private key content here.'),
category=_('SAML'),
category_slug='saml',
encrypted=True,
)
register(
'SOCIAL_AUTH_SAML_ORG_INFO',
field_class=SAMLOrgInfoField,
label=_('SAML Service Provider Organization Info'),
help_text=_('Provide the URL, display name, and the name of your app. Refer to the documentation for example syntax.'),
category=_('SAML'),
category_slug='saml',
placeholder=collections.OrderedDict(
[('en-US', collections.OrderedDict([('name', 'example'), ('displayname', 'Example'), ('url', 'http://www.example.com')]))]
),
)
register(
'SOCIAL_AUTH_SAML_TECHNICAL_CONTACT',
field_class=SAMLContactField,
allow_blank=True,
label=_('SAML Service Provider Technical Contact'),
help_text=_('Provide the name and email address of the technical contact for your service provider. Refer to the documentation for example syntax.'),
category=_('SAML'),
category_slug='saml',
placeholder=collections.OrderedDict([('givenName', 'Technical Contact'), ('emailAddress', 'techsup@example.com')]),
)
register(
'SOCIAL_AUTH_SAML_SUPPORT_CONTACT',
field_class=SAMLContactField,
allow_blank=True,
label=_('SAML Service Provider Support Contact'),
help_text=_('Provide the name and email address of the support contact for your service provider. Refer to the documentation for example syntax.'),
category=_('SAML'),
category_slug='saml',
placeholder=collections.OrderedDict([('givenName', 'Support Contact'), ('emailAddress', 'support@example.com')]),
)
register(
'SOCIAL_AUTH_SAML_ENABLED_IDPS',
field_class=SAMLEnabledIdPsField,
default={},
label=_('SAML Enabled Identity Providers'),
help_text=_(
'Configure the Entity ID, SSO URL and certificate for each identity'
' provider (IdP) in use. Multiple SAML IdPs are supported. Some IdPs'
' may provide user data using attribute names that differ from the'
' default OIDs. Attribute names may be overridden for each IdP. Refer'
' to the Ansible documentation for additional details and syntax.'
),
category=_('SAML'),
category_slug='saml',
placeholder=collections.OrderedDict(
[
(
'Okta',
collections.OrderedDict(
[
('entity_id', 'http://www.okta.com/HHniyLkaxk9e76wD0Thh'),
('url', 'https://dev-123456.oktapreview.com/app/ansibletower/HHniyLkaxk9e76wD0Thh/sso/saml'),
('x509cert', 'MIIDpDCCAoygAwIBAgIGAVVZ4rPzMA0GCSqGSIb3...'),
('attr_user_permanent_id', 'username'),
('attr_first_name', 'first_name'),
('attr_last_name', 'last_name'),
('attr_username', 'username'),
('attr_email', 'email'),
]
),
),
(
'OneLogin',
collections.OrderedDict(
[
('entity_id', 'https://app.onelogin.com/saml/metadata/123456'),
('url', 'https://example.onelogin.com/trust/saml2/http-post/sso/123456'),
('x509cert', 'MIIEJjCCAw6gAwIBAgIUfuSD54OPSBhndDHh3gZo...'),
('attr_user_permanent_id', 'name_id'),
('attr_first_name', 'User.FirstName'),
('attr_last_name', 'User.LastName'),
('attr_username', 'User.email'),
('attr_email', 'User.email'),
]
),
),
]
),
)
register(
'SOCIAL_AUTH_SAML_SECURITY_CONFIG',
field_class=SAMLSecurityField,
allow_null=True,
default={'requestedAuthnContext': False},
label=_('SAML Security Config'),
help_text=_(
'A dict of key value pairs that are passed to the underlying python-saml security setting https://github.com/onelogin/python-saml#settings'
),
category=_('SAML'),
category_slug='saml',
placeholder=collections.OrderedDict(
[
("nameIdEncrypted", False),
("authnRequestsSigned", False),
("logoutRequestSigned", False),
("logoutResponseSigned", False),
("signMetadata", False),
("wantMessagesSigned", False),
("wantAssertionsSigned", False),
("wantAssertionsEncrypted", False),
("wantNameId", True),
("wantNameIdEncrypted", False),
("wantAttributeStatement", True),
("requestedAuthnContext", True),
("requestedAuthnContextComparison", "exact"),
("metadataValidUntil", "2015-06-26T20:00:00Z"),
("metadataCacheDuration", "PT518400S"),
("signatureAlgorithm", "http://www.w3.org/2000/09/xmldsig#rsa-sha1"),
("digestAlgorithm", "http://www.w3.org/2000/09/xmldsig#sha1"),
]
),
)
register(
'SOCIAL_AUTH_SAML_SP_EXTRA',
field_class=fields.DictField,
allow_null=True,
default=None,
label=_('SAML Service Provider extra configuration data'),
help_text=_('A dict of key value pairs to be passed to the underlying python-saml Service Provider configuration setting.'),
category=_('SAML'),
category_slug='saml',
placeholder=collections.OrderedDict(),
)
register(
'SOCIAL_AUTH_SAML_EXTRA_DATA',
field_class=fields.ListTuplesField,
allow_null=True,
default=None,
label=_('SAML IDP to extra_data attribute mapping'),
help_text=_('A list of tuples that maps IDP attributes to extra_attributes.' ' Each attribute will be a list of values, even if only 1 value.'),
category=_('SAML'),
category_slug='saml',
placeholder=[('attribute_name', 'extra_data_name_for_attribute'), ('department', 'department'), ('manager_full_name', 'manager_full_name')],
)
register(
'SOCIAL_AUTH_SAML_ORGANIZATION_MAP',
field_class=SocialOrganizationMapField,
allow_null=True,
default=None,
label=_('SAML Organization Map'),
help_text=SOCIAL_AUTH_ORGANIZATION_MAP_HELP_TEXT,
category=_('SAML'),
category_slug='saml',
placeholder=SOCIAL_AUTH_ORGANIZATION_MAP_PLACEHOLDER,
)
register(
'SOCIAL_AUTH_SAML_TEAM_MAP',
field_class=SocialTeamMapField,
allow_null=True,
default=None,
label=_('SAML Team Map'),
help_text=SOCIAL_AUTH_TEAM_MAP_HELP_TEXT,
category=_('SAML'),
category_slug='saml',
placeholder=SOCIAL_AUTH_TEAM_MAP_PLACEHOLDER,
)
register(
'SOCIAL_AUTH_SAML_ORGANIZATION_ATTR',
field_class=SAMLOrgAttrField,
allow_null=True,
default=None,
label=_('SAML Organization Attribute Mapping'),
help_text=_('Used to translate user organization membership.'),
category=_('SAML'),
category_slug='saml',
placeholder=collections.OrderedDict(
[
('saml_attr', 'organization'),
('saml_admin_attr', 'organization_admin'),
('saml_auditor_attr', 'organization_auditor'),
('remove', True),
('remove_admins', True),
('remove_auditors', True),
]
),
)
register(
'SOCIAL_AUTH_SAML_TEAM_ATTR',
field_class=SAMLTeamAttrField,
allow_null=True,
default=None,
label=_('SAML Team Attribute Mapping'),
help_text=_('Used to translate user team membership.'),
category=_('SAML'),
category_slug='saml',
placeholder=collections.OrderedDict(
[
('saml_attr', 'team'),
('remove', True),
(
'team_org_map',
[
collections.OrderedDict([('team', 'Marketing'), ('organization', 'Red Hat')]),
collections.OrderedDict([('team', 'Human Resources'), ('organization', 'Red Hat')]),
collections.OrderedDict([('team', 'Engineering'), ('organization', 'Red Hat')]),
collections.OrderedDict([('team', 'Engineering'), ('organization', 'Ansible')]),
collections.OrderedDict([('team', 'Quality Engineering'), ('organization', 'Ansible')]),
collections.OrderedDict([('team', 'Sales'), ('organization', 'Ansible')]),
],
),
]
),
)
register(
'SOCIAL_AUTH_SAML_USER_FLAGS_BY_ATTR',
field_class=SAMLUserFlagsAttrField,
allow_null=True,
default=None,
label=_('SAML User Flags Attribute Mapping'),
help_text=_('Used to map super users and system auditors from SAML.'),
category=_('SAML'),
category_slug='saml',
placeholder=[
('is_superuser_attr', 'saml_attr'),
('is_superuser_value', ['value']),
('is_superuser_role', ['saml_role']),
('remove_superusers', True),
('is_system_auditor_attr', 'saml_attr'),
('is_system_auditor_value', ['value']),
('is_system_auditor_role', ['saml_role']),
('remove_system_auditors', True),
],
)
register(
'LOCAL_PASSWORD_MIN_LENGTH',
field_class=fields.IntegerField,

116
awx/sso/fields.py
View File

@@ -13,7 +13,6 @@ from rest_framework.fields import empty, Field, SkipField
# AWX
from awx.conf import fields
from awx.main.validators import validate_certificate
def get_subclasses(cls):
@@ -108,18 +107,6 @@ class AuthenticationBackendsField(fields.StringListField):
REQUIRED_BACKEND_SETTINGS = collections.OrderedDict(
[
('social_core.backends.open_id_connect.OpenIdConnectAuth', ['SOCIAL_AUTH_OIDC_KEY', 'SOCIAL_AUTH_OIDC_SECRET', 'SOCIAL_AUTH_OIDC_OIDC_ENDPOINT']),
(
'awx.sso.backends.SAMLAuth',
[
'SOCIAL_AUTH_SAML_SP_ENTITY_ID',
'SOCIAL_AUTH_SAML_SP_PUBLIC_CERT',
'SOCIAL_AUTH_SAML_SP_PRIVATE_KEY',
'SOCIAL_AUTH_SAML_ORG_INFO',
'SOCIAL_AUTH_SAML_TECHNICAL_CONTACT',
'SOCIAL_AUTH_SAML_SUPPORT_CONTACT',
'SOCIAL_AUTH_SAML_ENABLED_IDPS',
],
),
('django.contrib.auth.backends.ModelBackend', []),
('awx.main.backends.AWXModelBackend', []),
]
@@ -240,106 +227,3 @@ class SocialSingleTeamMapField(HybridDictField):
class SocialTeamMapField(fields.DictField):
child = SocialSingleTeamMapField()
class SAMLOrgInfoValueField(HybridDictField):
name = fields.CharField()
displayname = fields.CharField()
url = fields.URLField()
class SAMLOrgInfoField(fields.DictField):
default_error_messages = {'invalid_lang_code': _('Invalid language code(s) for org info: {invalid_lang_codes}.')}
child = SAMLOrgInfoValueField()
def to_internal_value(self, data):
data = super(SAMLOrgInfoField, self).to_internal_value(data)
invalid_keys = set()
for key in data.keys():
if not re.match(r'^[a-z]{2}(?:-[a-z]{2})??$', key, re.I):
invalid_keys.add(key)
if invalid_keys:
invalid_keys = sorted(list(invalid_keys))
keys_display = json.dumps(invalid_keys).lstrip('[').rstrip(']')
self.fail('invalid_lang_code', invalid_lang_codes=keys_display)
return data
class SAMLContactField(HybridDictField):
givenName = fields.CharField()
emailAddress = fields.EmailField()
class SAMLIdPField(HybridDictField):
entity_id = fields.CharField()
url = fields.URLField()
x509cert = fields.CharField(validators=[validate_certificate])
attr_user_permanent_id = fields.CharField(required=False)
attr_first_name = fields.CharField(required=False)
attr_last_name = fields.CharField(required=False)
attr_username = fields.CharField(required=False)
attr_email = fields.CharField(required=False)
class SAMLEnabledIdPsField(fields.DictField):
child = SAMLIdPField()
class SAMLSecurityField(HybridDictField):
nameIdEncrypted = fields.BooleanField(required=False)
authnRequestsSigned = fields.BooleanField(required=False)
logoutRequestSigned = fields.BooleanField(required=False)
logoutResponseSigned = fields.BooleanField(required=False)
signMetadata = fields.BooleanField(required=False)
wantMessagesSigned = fields.BooleanField(required=False)
wantAssertionsSigned = fields.BooleanField(required=False)
wantAssertionsEncrypted = fields.BooleanField(required=False)
wantNameId = fields.BooleanField(required=False)
wantNameIdEncrypted = fields.BooleanField(required=False)
wantAttributeStatement = fields.BooleanField(required=False)
requestedAuthnContext = fields.StringListBooleanField(required=False)
requestedAuthnContextComparison = fields.CharField(required=False)
metadataValidUntil = fields.CharField(allow_null=True, required=False)
metadataCacheDuration = fields.CharField(allow_null=True, required=False)
signatureAlgorithm = fields.CharField(allow_null=True, required=False)
digestAlgorithm = fields.CharField(allow_null=True, required=False)
class SAMLOrgAttrField(HybridDictField):
remove = fields.BooleanField(required=False)
saml_attr = fields.CharField(required=False, allow_null=True)
remove_admins = fields.BooleanField(required=False)
saml_admin_attr = fields.CharField(required=False, allow_null=True)
remove_auditors = fields.BooleanField(required=False)
saml_auditor_attr = fields.CharField(required=False, allow_null=True)
child = _Forbidden()
class SAMLTeamAttrTeamOrgMapField(HybridDictField):
team = fields.CharField(required=True, allow_null=False)
team_alias = fields.CharField(required=False, allow_null=True)
organization = fields.CharField(required=True, allow_null=False)
child = _Forbidden()
class SAMLTeamAttrField(HybridDictField):
team_org_map = fields.ListField(required=False, child=SAMLTeamAttrTeamOrgMapField(), allow_null=True)
remove = fields.BooleanField(required=False)
saml_attr = fields.CharField(required=False, allow_null=True)
child = _Forbidden()
class SAMLUserFlagsAttrField(HybridDictField):
is_superuser_attr = fields.CharField(required=False, allow_null=True)
is_superuser_value = fields.StringListField(required=False, allow_null=True)
is_superuser_role = fields.StringListField(required=False, allow_null=True)
remove_superusers = fields.BooleanField(required=False, allow_null=True)
is_system_auditor_attr = fields.CharField(required=False, allow_null=True)
is_system_auditor_value = fields.StringListField(required=False, allow_null=True)
is_system_auditor_role = fields.StringListField(required=False, allow_null=True)
remove_system_auditors = fields.BooleanField(required=False, allow_null=True)
child = _Forbidden()

55
awx/sso/migrations/0003_convert_saml_string_to_list.py
View File

@@ -1,58 +1,9 @@
from django.db import migrations, connection
import json
_values_to_change = ['is_superuser_value', 'is_superuser_role', 'is_system_auditor_value', 'is_system_auditor_role']
def _get_setting():
with connection.cursor() as cursor:
cursor.execute('SELECT value FROM conf_setting WHERE key= %s', ['SOCIAL_AUTH_SAML_USER_FLAGS_BY_ATTR'])
row = cursor.fetchone()
if row == None:
return {}
existing_setting = row[0]
try:
existing_json = json.loads(existing_setting)
except json.decoder.JSONDecodeError as e:
print("Failed to decode existing json setting:")
print(existing_setting)
raise e
return existing_json
def _set_setting(value):
with connection.cursor() as cursor:
cursor.execute('UPDATE conf_setting SET value = %s WHERE key = %s', [json.dumps(value), 'SOCIAL_AUTH_SAML_USER_FLAGS_BY_ATTR'])
def forwards(app, schema_editor):
# The Operation should use schema_editor to apply any changes it
# wants to make to the database.
existing_json = _get_setting()
for key in _values_to_change:
if existing_json.get(key, None) and isinstance(existing_json.get(key), str):
existing_json[key] = [existing_json.get(key)]
_set_setting(existing_json)
def backwards(app, schema_editor):
existing_json = _get_setting()
for key in _values_to_change:
if existing_json.get(key, None) and not isinstance(existing_json.get(key), str):
try:
existing_json[key] = existing_json.get(key).pop()
except IndexError:
existing_json[key] = ""
_set_setting(existing_json)
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
('sso', '0002_expand_provider_options'),
]
operations = [
migrations.RunPython(forwards, backwards),
]
# NOOP, migration is kept to preserve integrity.
operations = []

18
awx/sso/migrations/0004_alter_userenterpriseauth_provider.py Normal file
View File

@@ -0,0 +1,18 @@
# Generated by Django 4.2.10 on 2024-10-02 12:44
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('sso', '0003_convert_saml_string_to_list'),
]
operations = [
migrations.AlterField(
model_name='userenterpriseauth',
name='provider',
field=models.CharField(choices=[('radius', 'RADIUS'), ('tacacs+', 'TACACS+')], max_length=32),
),
]

2
awx/sso/models.py
View File

@@ -11,7 +11,7 @@ from django.utils.translation import gettext_lazy as _
class UserEnterpriseAuth(models.Model):
"""Enterprise Auth association model"""
PROVIDER_CHOICES = (('radius', _('RADIUS')), ('tacacs+', _('TACACS+')), ('saml', _('SAML')))
PROVIDER_CHOICES = (('radius', _('RADIUS')), ('tacacs+', _('TACACS+')))
class Meta:
unique_together = ('user', 'provider')

291
awx/sso/saml_pipeline.py
View File

@@ -1,291 +0,0 @@
# Copyright (c) 2015 Ansible, Inc.
# All Rights Reserved.
# Python
import re
import logging
# Django
from django.conf import settings
from awx.main.models import Team
from awx.sso.common import create_org_and_teams, reconcile_users_org_team_mappings, get_orgs_by_ids
logger = logging.getLogger('awx.sso.saml_pipeline')
def populate_user(backend, details, user=None, *args, **kwargs):
if not user:
return
# Build the in-memory settings for how this user should be modeled
desired_org_state = {}
desired_team_state = {}
orgs_to_create = []
teams_to_create = {}
_update_user_orgs_by_saml_attr(backend, desired_org_state, orgs_to_create, **kwargs)
_update_user_teams_by_saml_attr(desired_team_state, teams_to_create, **kwargs)
_update_user_orgs(backend, desired_org_state, orgs_to_create, user)
_update_user_teams(backend, desired_team_state, teams_to_create, user)
# If the SAML adapter is allowed to create objects, lets do that first
create_org_and_teams(orgs_to_create, teams_to_create, 'SAML', settings.SAML_AUTO_CREATE_OBJECTS)
# Finally reconcile the user
reconcile_users_org_team_mappings(user, desired_org_state, desired_team_state, 'SAML')
def _update_m2m_from_expression(user, expr, remove=True):
"""
Helper function to update m2m relationship based on user matching one or
more expressions.
"""
should_add = False
if expr is None or not expr:
pass
elif expr is True:
should_add = True
else:
if isinstance(expr, (str, type(re.compile('')))):
expr = [expr]
for ex in expr:
if isinstance(ex, str):
if user.username == ex or user.email == ex:
should_add = True
elif isinstance(ex, type(re.compile(''))):
if ex.match(user.username) or ex.match(user.email):
should_add = True
if should_add:
return True
elif remove:
return False
else:
return None
def _update_user_orgs(backend, desired_org_state, orgs_to_create, user=None):
"""
Update organization memberships for the given user based on mapping rules
defined in settings.
"""
org_map = backend.setting('ORGANIZATION_MAP') or {}
for org_name, org_opts in org_map.items():
organization_alias = org_opts.get('organization_alias')
if organization_alias:
organization_name = organization_alias
else:
organization_name = org_name
if organization_name not in orgs_to_create:
orgs_to_create.append(organization_name)
remove = bool(org_opts.get('remove', True))
if organization_name not in desired_org_state:
desired_org_state[organization_name] = {}
for role_name, user_type in (('admin_role', 'admins'), ('member_role', 'users'), ('auditor_role', 'auditors')):
is_member_expression = org_opts.get(user_type, None)
remove_members = bool(org_opts.get('remove_{}'.format(user_type), remove))
has_role = _update_m2m_from_expression(user, is_member_expression, remove_members)
desired_org_state[organization_name][role_name] = desired_org_state[organization_name].get(role_name, False) or has_role
def _update_user_teams(backend, desired_team_state, teams_to_create, user=None):
"""
Update team memberships for the given user based on mapping rules defined
in settings.
"""
team_map = backend.setting('TEAM_MAP') or {}
for team_name, team_opts in team_map.items():
# Get or create the org to update.
if 'organization' not in team_opts:
continue
teams_to_create[team_name] = team_opts['organization']
users_expr = team_opts.get('users', None)
remove = bool(team_opts.get('remove', True))
add_or_remove = _update_m2m_from_expression(user, users_expr, remove)
if add_or_remove is not None:
org_name = team_opts['organization']
if org_name not in desired_team_state:
desired_team_state[org_name] = {}
desired_team_state[org_name][team_name] = {'member_role': add_or_remove}
def _update_user_orgs_by_saml_attr(backend, desired_org_state, orgs_to_create, **kwargs):
org_map = settings.SOCIAL_AUTH_SAML_ORGANIZATION_ATTR
roles_and_flags = (
('member_role', 'remove', 'saml_attr'),
('admin_role', 'remove_admins', 'saml_admin_attr'),
('auditor_role', 'remove_auditors', 'saml_auditor_attr'),
)
# If the remove_flag was present we need to load all of the orgs and remove the user from the role
all_orgs = None
for role, remove_flag, _ in roles_and_flags:
remove = bool(org_map.get(remove_flag, True))
if remove:
# Only get the all orgs once, and only if needed
if all_orgs is None:
all_orgs = get_orgs_by_ids()
for org_name in all_orgs.keys():
if org_name not in desired_org_state:
desired_org_state[org_name] = {}
desired_org_state[org_name][role] = False
# Now we can add the user as a member/admin/auditor for any orgs they have specified
for role, _, attr_flag in roles_and_flags:
if org_map.get(attr_flag) is None:
continue
saml_attr_values = kwargs.get('response', {}).get('attributes', {}).get(org_map.get(attr_flag), [])
for org_name in saml_attr_values:
try:
organization_alias = backend.setting('ORGANIZATION_MAP').get(org_name).get('organization_alias')
if organization_alias is not None:
organization_name = organization_alias
else:
organization_name = org_name
except Exception:
organization_name = org_name
if organization_name not in orgs_to_create:
orgs_to_create.append(organization_name)
if organization_name not in desired_org_state:
desired_org_state[organization_name] = {}
desired_org_state[organization_name][role] = True
def _update_user_teams_by_saml_attr(desired_team_state, teams_to_create, **kwargs):
#
# Map users into organizations based on SOCIAL_AUTH_SAML_TEAM_ATTR setting
#
team_map = settings.SOCIAL_AUTH_SAML_TEAM_ATTR
if team_map.get('saml_attr') is None:
return
all_teams = None
# The role and flag is hard coded here but intended to be flexible in case we ever wanted to add another team type
for role, remove_flag in [('member_role', 'remove')]:
remove = bool(team_map.get(remove_flag, True))
if remove:
# Only get the all orgs once, and only if needed
if all_teams is None:
all_teams = Team.objects.all().values_list('name', 'organization__name')
for team_name, organization_name in all_teams:
if organization_name not in desired_team_state:
desired_team_state[organization_name] = {}
desired_team_state[organization_name][team_name] = {role: False}
saml_team_names = set(kwargs.get('response', {}).get('attributes', {}).get(team_map['saml_attr'], []))
for team_name_map in team_map.get('team_org_map', []):
team_name = team_name_map.get('team', None)
team_alias = team_name_map.get('team_alias', None)
organization_name = team_name_map.get('organization', None)
if team_name in saml_team_names:
if not organization_name:
# Settings field validation should prevent this.
logger.error("organization name invalid for team {}".format(team_name))
continue
if team_alias:
team_name = team_alias
teams_to_create[team_name] = organization_name
user_is_member_of_team = True
else:
user_is_member_of_team = False
if organization_name not in desired_team_state:
desired_team_state[organization_name] = {}
desired_team_state[organization_name][team_name] = {'member_role': user_is_member_of_team}
def _get_matches(list1, list2):
# Because we are just doing an intersection here we don't really care which list is in which parameter
# A SAML provider could return either a string or a list of items so we need to coerce the SAML value into a list (if needed)
if not isinstance(list1, (list, tuple)):
list1 = [list1]
# In addition, we used to allow strings in the SAML config instead of Lists. The migration should take case of that but just in case, we will convert our list too
if not isinstance(list2, (list, tuple)):
list2 = [list2]
return set(list1).intersection(set(list2))
def _check_flag(user, flag, attributes, user_flags_settings):
'''
Helper function to set the is_superuser is_system_auditor flags for the SAML adapter
Returns the new flag and whether or not it changed the flag
'''
new_flag = False
is_role_key = "is_%s_role" % (flag)
is_attr_key = "is_%s_attr" % (flag)
is_value_key = "is_%s_value" % (flag)
remove_setting = "remove_%ss" % (flag)
# Check to see if we are respecting a role and, if so, does our user have that role?
required_roles = user_flags_settings.get(is_role_key, None)
if required_roles:
matching_roles = _get_matches(required_roles, attributes.get('Role', []))
# We do a 2 layer check here so that we don't spit out the else message if there is no role defined
if matching_roles:
logger.debug("User %s has %s role(s) %s" % (user.username, flag, ', '.join(matching_roles)))
new_flag = True
else:
logger.debug("User %s is missing the %s role(s) %s" % (user.username, flag, ', '.join(required_roles)))
# Next, check to see if we are respecting an attribute; this will take priority over the role if its defined
attr_setting = user_flags_settings.get(is_attr_key, None)
if attr_setting and attributes.get(attr_setting, None):
# Do we have a required value for the attribute
required_value = user_flags_settings.get(is_value_key, None)
if required_value:
# If so, check and see if the value of the attr matches the required value
saml_user_attribute_value = attributes.get(attr_setting, None)
matching_values = _get_matches(required_value, saml_user_attribute_value)
if matching_values:
logger.debug("Giving %s %s from attribute %s with matching values %s" % (user.username, flag, attr_setting, ', '.join(matching_values)))
new_flag = True
# if they don't match make sure that new_flag is false
else:
logger.debug(
"Refusing %s for %s because attr %s (%s) did not match value(s) %s"
% (flag, user.username, attr_setting, ", ".join(saml_user_attribute_value), ', '.join(required_value))
)
new_flag = False
# If there was no required value then we can just allow them in because of the attribute
else:
logger.debug("Giving %s %s from attribute %s" % (user.username, flag, attr_setting))
new_flag = True
# Get the users old flag
old_value = getattr(user, "is_%s" % (flag))
# If we are not removing the flag and they were a system admin and now we don't want them to be just return
remove_flag = user_flags_settings.get(remove_setting, True)
if not remove_flag and (old_value and not new_flag):
logger.debug("Remove flag %s preventing removal of %s for %s" % (remove_flag, flag, user.username))
return old_value, False
# If the user was flagged and we are going to make them not flagged make sure there is a message
if old_value and not new_flag:
logger.debug("Revoking %s from %s" % (flag, user.username))
return new_flag, old_value != new_flag
def update_user_flags(backend, details, user=None, *args, **kwargs):
user_flags_settings = settings.SOCIAL_AUTH_SAML_USER_FLAGS_BY_ATTR
attributes = kwargs.get('response', {}).get('attributes', {})
logger.debug("User attributes for %s: %s" % (user.username, attributes))
user.is_superuser, superuser_changed = _check_flag(user, 'superuser', attributes, user_flags_settings)
user.is_system_auditor, auditor_changed = _check_flag(user, 'system_auditor', attributes, user_flags_settings)
if superuser_changed or auditor_changed:
user.save()

3
awx/sso/tests/functional/test_common.py
View File

@@ -324,11 +324,8 @@ class TestCommonFunctions:
[
# Set none of the social auth settings
('JUNK_SETTING', False),
('SOCIAL_AUTH_SAML_ENABLED_IDPS', True),
# Try a hypothetical future one
('SOCIAL_AUTH_GIBBERISH_KEY', True),
# Do a SAML one
('SOCIAL_AUTH_SAML_SP_PRIVATE_KEY', False),
],
)
def test_is_remote_auth_enabled(self, setting, expected):

711
awx/sso/tests/functional/test_saml_pipeline.py
View File

@@ -1,711 +0,0 @@
import pytest
import re
from django.test.utils import override_settings
from awx.main.models import User, Organization, Team
from awx.sso.saml_pipeline import (
_update_m2m_from_expression,
_update_user_orgs,
_update_user_teams,
_update_user_orgs_by_saml_attr,
_update_user_teams_by_saml_attr,
_check_flag,
)
# from unittest import mock
# from django.utils.timezone import now
# , Credential, CredentialType
@pytest.fixture
def users():
u1 = User.objects.create(username='user1@foo.com', last_name='foo', first_name='bar', email='user1@foo.com')
u2 = User.objects.create(username='user2@foo.com', last_name='foo', first_name='bar', email='user2@foo.com')
u3 = User.objects.create(username='user3@foo.com', last_name='foo', first_name='bar', email='user3@foo.com')
return (u1, u2, u3)
@pytest.mark.django_db
class TestSAMLPopulateUser:
# The main populate_user does not need to be tested since its just a conglomeration of other functions that we test
# This test is here in case someone alters the code in the future in a way that does require testing
def test_populate_user(self):
assert True
@pytest.mark.django_db
class TestSAMLSimpleMaps:
# This tests __update_user_orgs and __update_user_teams
@pytest.fixture
def backend(self):
class Backend:
s = {
'ORGANIZATION_MAP': {
'Default': {
'remove': True,
'admins': 'foobar',
'remove_admins': True,
'users': 'foo',
'remove_users': True,
'organization_alias': '',
}
},
'TEAM_MAP': {'Blue': {'organization': 'Default', 'remove': True, 'users': ''}, 'Red': {'organization': 'Default', 'remove': True, 'users': ''}},
}
def setting(self, key):
return self.s[key]
return Backend()
def test__update_user_orgs(self, backend, users):
u1, u2, u3 = users
# Test user membership logic with regular expressions
backend.setting('ORGANIZATION_MAP')['Default']['admins'] = re.compile('.*')
backend.setting('ORGANIZATION_MAP')['Default']['users'] = re.compile('.*')
desired_org_state = {}
orgs_to_create = []
_update_user_orgs(backend, desired_org_state, orgs_to_create, u1)
_update_user_orgs(backend, desired_org_state, orgs_to_create, u2)
_update_user_orgs(backend, desired_org_state, orgs_to_create, u3)
assert desired_org_state == {'Default': {'member_role': True, 'admin_role': True, 'auditor_role': False}}
assert orgs_to_create == ['Default']
# Test remove feature enabled
backend.setting('ORGANIZATION_MAP')['Default']['admins'] = ''
backend.setting('ORGANIZATION_MAP')['Default']['users'] = ''
backend.setting('ORGANIZATION_MAP')['Default']['remove_admins'] = True
backend.setting('ORGANIZATION_MAP')['Default']['remove_users'] = True
desired_org_state = {}
orgs_to_create = []
_update_user_orgs(backend, desired_org_state, orgs_to_create, u1)
assert desired_org_state == {'Default': {'member_role': False, 'admin_role': False, 'auditor_role': False}}
assert orgs_to_create == ['Default']
# Test remove feature disabled
backend.setting('ORGANIZATION_MAP')['Default']['remove_admins'] = False
backend.setting('ORGANIZATION_MAP')['Default']['remove_users'] = False
desired_org_state = {}
orgs_to_create = []
_update_user_orgs(backend, desired_org_state, orgs_to_create, u2)
assert desired_org_state == {'Default': {'member_role': None, 'admin_role': None, 'auditor_role': False}}
assert orgs_to_create == ['Default']
# Test organization alias feature
backend.setting('ORGANIZATION_MAP')['Default']['organization_alias'] = 'Default_Alias'
orgs_to_create = []
_update_user_orgs(backend, {}, orgs_to_create, u1)
assert orgs_to_create == ['Default_Alias']
def test__update_user_teams(self, backend, users):
u1, u2, u3 = users
# Test user membership logic with regular expressions
backend.setting('TEAM_MAP')['Blue']['users'] = re.compile('.*')
backend.setting('TEAM_MAP')['Red']['users'] = re.compile('.*')
desired_team_state = {}
teams_to_create = {}
_update_user_teams(backend, desired_team_state, teams_to_create, u1)
assert teams_to_create == {'Red': 'Default', 'Blue': 'Default'}
assert desired_team_state == {'Default': {'Blue': {'member_role': True}, 'Red': {'member_role': True}}}
# Test remove feature enabled
backend.setting('TEAM_MAP')['Blue']['remove'] = True
backend.setting('TEAM_MAP')['Red']['remove'] = True
backend.setting('TEAM_MAP')['Blue']['users'] = ''
backend.setting('TEAM_MAP')['Red']['users'] = ''
desired_team_state = {}
teams_to_create = {}
_update_user_teams(backend, desired_team_state, teams_to_create, u1)
assert teams_to_create == {'Red': 'Default', 'Blue': 'Default'}
assert desired_team_state == {'Default': {'Blue': {'member_role': False}, 'Red': {'member_role': False}}}
# Test remove feature disabled
backend.setting('TEAM_MAP')['Blue']['remove'] = False
backend.setting('TEAM_MAP')['Red']['remove'] = False
desired_team_state = {}
teams_to_create = {}
_update_user_teams(backend, desired_team_state, teams_to_create, u2)
assert teams_to_create == {'Red': 'Default', 'Blue': 'Default'}
# If we don't care about team memberships we just don't add them to the hash so this would be an empty hash
assert desired_team_state == {}
@pytest.mark.django_db
class TestSAMLM2M:
@pytest.mark.parametrize(
"expression, remove, expected_return",
[
# No expression with no remove
(None, False, None),
("", False, None),
# No expression with remove
(None, True, False),
# True expression with and without remove
(True, False, True),
(True, True, True),
# Single string matching the user name
("user1", False, True),
# Single string matching the user email
("user1@foo.com", False, True),
# Single string not matching username or email, no remove
("user27", False, None),
# Single string not matching username or email, with remove
("user27", True, False),
# Same tests with arrays instead of strings
(["user1"], False, True),
(["user1@foo.com"], False, True),
(["user27"], False, None),
(["user27"], True, False),
# Arrays with nothing matching
(["user27", "user28"], False, None),
(["user27", "user28"], True, False),
# Arrays with all matches
(["user1", "user1@foo.com"], False, True),
# Arrays with some match, some not
(["user1", "user28", "user27"], False, True),
#
# Note: For RE's, usually settings takes care of the compilation for us, so we have to do it manually for testing.
# we also need to remove any / or flags for the compile to happen
#
# Matching username regex non-array
(re.compile("^user.*"), False, True),
(re.compile("^user.*"), True, True),
# Matching email regex non-array
(re.compile(".*@foo.com$"), False, True),
(re.compile(".*@foo.com$"), True, True),
# Non-array not matching username or email
(re.compile("^$"), False, None),
(re.compile("^$"), True, False),
# All re tests just in array form
([re.compile("^user.*")], False, True),
([re.compile("^user.*")], True, True),
([re.compile(".*@foo.com$")], False, True),
([re.compile(".*@foo.com$")], True, True),
([re.compile("^$")], False, None),
([re.compile("^$")], True, False),
# An re with username matching but not email
([re.compile("^user.*"), re.compile(".*@bar.com$")], False, True),
# An re with email matching but not username
([re.compile("^user27$"), re.compile(".*@foo.com$")], False, True),
# An re array with no matching
([re.compile("^user27$"), re.compile(".*@bar.com$")], False, None),
([re.compile("^user27$"), re.compile(".*@bar.com$")], True, False),
#
# A mix of re and strings
#
# String matches, re does not
(["user1", re.compile(".*@bar.com$")], False, True),
# String does not match, re does
(["user27", re.compile(".*@foo.com$")], False, True),
# Nothing matches
(["user27", re.compile(".*@bar.com$")], False, None),
(["user27", re.compile(".*@bar.com$")], True, False),
],
)
def test__update_m2m_from_expression(self, expression, remove, expected_return):
user = User.objects.create(username='user1', last_name='foo', first_name='bar', email='user1@foo.com')
return_val = _update_m2m_from_expression(user, expression, remove)
assert return_val == expected_return
@pytest.mark.django_db
class TestSAMLAttrMaps:
@pytest.fixture
def backend(self):
class Backend:
s = {
'ORGANIZATION_MAP': {
'Default1': {
'remove': True,
'admins': 'foobar',
'remove_admins': True,
'users': 'foo',
'remove_users': True,
'organization_alias': 'o1_alias',
}
}
}
def setting(self, key):
return self.s[key]
return Backend()
@pytest.mark.parametrize(
"setting, expected_state, expected_orgs_to_create, kwargs_member_of_mods",
[
(
# Default test, make sure that our roles get applied and removed as specified (with an alias)
{
'saml_attr': 'memberOf',
'saml_admin_attr': 'admins',
'saml_auditor_attr': 'auditors',
'remove': True,
'remove_admins': True,
},
{
'Default2': {'member_role': True},
'Default3': {'admin_role': True},
'Default4': {'auditor_role': True},
'o1_alias': {'member_role': True},
'Rando1': {'admin_role': False, 'auditor_role': False, 'member_role': False},
},
[
'o1_alias',
'Default2',
'Default3',
'Default4',
],
None,
),
(
# Similar test, we are just going to override the values "coming from the IdP" to limit the teams
{
'saml_attr': 'memberOf',
'saml_admin_attr': 'admins',
'saml_auditor_attr': 'auditors',
'remove': True,
'remove_admins': True,
},
{
'Default3': {'admin_role': True, 'member_role': True},
'Default4': {'auditor_role': True},
'Rando1': {'admin_role': False, 'auditor_role': False, 'member_role': False},
},
[
'Default3',
'Default4',
],
['Default3'],
),
(
# Test to make sure the remove logic is working
{
'saml_attr': 'memberOf',
'saml_admin_attr': 'admins',
'saml_auditor_attr': 'auditors',
'remove': False,
'remove_admins': False,
'remove_auditors': False,
},
{
'Default2': {'member_role': True},
'Default3': {'admin_role': True},
'Default4': {'auditor_role': True},
'o1_alias': {'member_role': True},
},
[
'o1_alias',
'Default2',
'Default3',
'Default4',
],
['Default1', 'Default2'],
),
],
)
def test__update_user_orgs_by_saml_attr(self, backend, setting, expected_state, expected_orgs_to_create, kwargs_member_of_mods):
kwargs = {
'username': u'cmeyers@redhat.com',
'uid': 'idp:cmeyers@redhat.com',
'request': {u'SAMLResponse': [], u'RelayState': [u'idp']},
'is_new': False,
'response': {
'session_index': '_0728f0e0-b766-0135-75fa-02842b07c044',
'idp_name': u'idp',
'attributes': {
'memberOf': ['Default1', 'Default2'],
'admins': ['Default3'],
'auditors': ['Default4'],
'groups': ['Blue', 'Red'],
'User.email': ['cmeyers@redhat.com'],
'User.LastName': ['Meyers'],
'name_id': 'cmeyers@redhat.com',
'User.FirstName': ['Chris'],
'PersonImmutableID': [],
},
},
'social': None,
'strategy': None,
'new_association': False,
}
if kwargs_member_of_mods:
kwargs['response']['attributes']['memberOf'] = kwargs_member_of_mods
# Create a random organization in the database for testing
Organization.objects.create(name='Rando1')
with override_settings(SOCIAL_AUTH_SAML_ORGANIZATION_ATTR=setting):
desired_org_state = {}
orgs_to_create = []
_update_user_orgs_by_saml_attr(backend, desired_org_state, orgs_to_create, **kwargs)
assert desired_org_state == expected_state
assert orgs_to_create == expected_orgs_to_create
@pytest.mark.parametrize(
"setting, expected_team_state, expected_teams_to_create, kwargs_group_override",
[
(
{
'saml_attr': 'groups',
'remove': False,
'team_org_map': [
{'team': 'Blue', 'organization': 'Default1'},
{'team': 'Blue', 'organization': 'Default2'},
{'team': 'Blue', 'organization': 'Default3'},
{'team': 'Red', 'organization': 'Default1'},
{'team': 'Green', 'organization': 'Default1'},
{'team': 'Green', 'organization': 'Default3'},
{'team': 'Yellow', 'team_alias': 'Yellow_Alias', 'organization': 'Default4', 'organization_alias': 'Default4_Alias'},
],
},
{
'Default1': {
'Blue': {'member_role': True},
'Green': {'member_role': False},
'Red': {'member_role': True},
},
'Default2': {
'Blue': {'member_role': True},
},
'Default3': {
'Blue': {'member_role': True},
'Green': {'member_role': False},
},
'Default4': {
'Yellow': {'member_role': False},
},
},
{
'Blue': 'Default3',
'Red': 'Default1',
},
None,
),
(
{
'saml_attr': 'groups',
'remove': False,
'team_org_map': [
{'team': 'Blue', 'organization': 'Default1'},
{'team': 'Blue', 'organization': 'Default2'},
{'team': 'Blue', 'organization': 'Default3'},
{'team': 'Red', 'organization': 'Default1'},
{'team': 'Green', 'organization': 'Default1'},
{'team': 'Green', 'organization': 'Default3'},
{'team': 'Yellow', 'team_alias': 'Yellow_Alias', 'organization': 'Default4', 'organization_alias': 'Default4_Alias'},
],
},
{
'Default1': {
'Blue': {'member_role': True},
'Green': {'member_role': True},
'Red': {'member_role': True},
},
'Default2': {
'Blue': {'member_role': True},
},
'Default3': {
'Blue': {'member_role': True},
'Green': {'member_role': True},
},
'Default4': {
'Yellow': {'member_role': False},
},
},
{
'Blue': 'Default3',
'Red': 'Default1',
'Green': 'Default3',
},
['Blue', 'Red', 'Green'],
),
(
{
'saml_attr': 'groups',
'remove': True,
'team_org_map': [
{'team': 'Blue', 'organization': 'Default1'},
{'team': 'Blue', 'organization': 'Default2'},
{'team': 'Blue', 'organization': 'Default3'},
{'team': 'Red', 'organization': 'Default1'},
{'team': 'Green', 'organization': 'Default1'},
{'team': 'Green', 'organization': 'Default3'},
{'team': 'Yellow', 'team_alias': 'Yellow_Alias', 'organization': 'Default4', 'organization_alias': 'Default4_Alias'},
],
},
{
'Default1': {
'Blue': {'member_role': False},
'Green': {'member_role': True},
'Red': {'member_role': False},
},
'Default2': {
'Blue': {'member_role': False},
},
'Default3': {
'Blue': {'member_role': False},
'Green': {'member_role': True},
},
'Default4': {
'Yellow': {'member_role': False},
},
'Rando1': {
'Rando1': {'member_role': False},
},
},
{
'Green': 'Default3',
},
['Green'],
),
],
)
def test__update_user_teams_by_saml_attr(self, setting, expected_team_state, expected_teams_to_create, kwargs_group_override):
kwargs = {
'username': u'cmeyers@redhat.com',
'uid': 'idp:cmeyers@redhat.com',
'request': {u'SAMLResponse': [], u'RelayState': [u'idp']},
'is_new': False,
'response': {
'session_index': '_0728f0e0-b766-0135-75fa-02842b07c044',
'idp_name': u'idp',
'attributes': {
'memberOf': ['Default1', 'Default2'],
'admins': ['Default3'],
'auditors': ['Default4'],
'groups': ['Blue', 'Red'],
'User.email': ['cmeyers@redhat.com'],
'User.LastName': ['Meyers'],
'name_id': 'cmeyers@redhat.com',
'User.FirstName': ['Chris'],
'PersonImmutableID': [],
},
},
'social': None,
'strategy': None,
'new_association': False,
}
if kwargs_group_override:
kwargs['response']['attributes']['groups'] = kwargs_group_override
o = Organization.objects.create(name='Rando1')
Team.objects.create(name='Rando1', organization_id=o.id)
with override_settings(SOCIAL_AUTH_SAML_TEAM_ATTR=setting):
desired_team_state = {}
teams_to_create = {}
_update_user_teams_by_saml_attr(desired_team_state, teams_to_create, **kwargs)
assert desired_team_state == expected_team_state
assert teams_to_create == expected_teams_to_create
@pytest.mark.django_db
class TestSAMLUserFlags:
@pytest.mark.parametrize(
"user_flags_settings, expected, is_superuser",
[
# In this case we will pass no user flags so new_flag should be false and changed will def be false
(
{},
(False, False),
False,
),
# NOTE: The first handful of tests test role/value as string instead of lists.
# This was from the initial implementation of these fields but the code should be able to handle this
# There are a couple tests at the end of this which will validate arrays in these values.
#
# In this case we will give the user a group to make them an admin
(
{'is_superuser_role': 'test-role-1'},
(True, True),
False,
),
# In this case we will give the user a flag that will make then an admin
(
{'is_superuser_attr': 'is_superuser'},
(True, True),
False,
),
# In this case we will give the user a flag but the wrong value
(
{'is_superuser_attr': 'is_superuser', 'is_superuser_value': 'junk'},
(False, False),
False,
),
# In this case we will give the user a flag and the right value
(
{'is_superuser_attr': 'is_superuser', 'is_superuser_value': 'true'},
(True, True),
False,
),
# In this case we will give the user a proper role and an is_superuser_attr role that they don't have, this should make them an admin
(
{'is_superuser_role': 'test-role-1', 'is_superuser_attr': 'gibberish', 'is_superuser_value': 'true'},
(True, True),
False,
),
# In this case we will give the user a proper role and an is_superuser_attr role that they have, this should make them an admin
(
{'is_superuser_role': 'test-role-1', 'is_superuser_attr': 'test-role-1'},
(True, True),
False,
),
# In this case we will give the user a proper role and an is_superuser_attr role that they have but a bad value, this should make them an admin
(
{'is_superuser_role': 'test-role-1', 'is_superuser_attr': 'is_superuser', 'is_superuser_value': 'junk'},
(False, False),
False,
),
# In this case we will give the user everything
(
{'is_superuser_role': 'test-role-1', 'is_superuser_attr': 'is_superuser', 'is_superuser_value': 'true'},
(True, True),
False,
),
# In this test case we will validate that a single attribute (instead of a list) still works
(
{'is_superuser_attr': 'name_id', 'is_superuser_value': 'test_id'},
(True, True),
False,
),
# This will be a negative test for a single attribute
(
{'is_superuser_attr': 'name_id', 'is_superuser_value': 'junk'},
(False, False),
False,
),
# The user is already a superuser so we should remove them
(
{'is_superuser_attr': 'name_id', 'is_superuser_value': 'junk', 'remove_superusers': True},
(False, True),
True,
),
# The user is already a superuser but we don't have a remove field
(
{'is_superuser_attr': 'name_id', 'is_superuser_value': 'junk', 'remove_superusers': False},
(True, False),
True,
),
# Positive test for multiple values for is_superuser_value
(
{'is_superuser_attr': 'is_superuser', 'is_superuser_value': ['junk', 'junk2', 'else', 'junk']},
(True, True),
False,
),
# Negative test for multiple values for is_superuser_value
(
{'is_superuser_attr': 'is_superuser', 'is_superuser_value': ['junk', 'junk2', 'junk']},
(False, True),
True,
),
# Positive test for multiple values of is_superuser_role
(
{'is_superuser_role': ['junk', 'junk2', 'something', 'junk']},
(True, True),
False,
),
# Negative test for multiple values of is_superuser_role
(
{'is_superuser_role': ['junk', 'junk2', 'junk']},
(False, True),
True,
),
],
)
def test__check_flag(self, user_flags_settings, expected, is_superuser):
user = User()
user.username = 'John'
user.is_superuser = is_superuser
attributes = {
'email': ['noone@nowhere.com'],
'last_name': ['Westcott'],
'is_superuser': ['something', 'else', 'true'],
'username': ['test_id'],
'first_name': ['John'],
'Role': ['test-role-1', 'something', 'different'],
'name_id': 'test_id',
}
assert expected == _check_flag(user, 'superuser', attributes, user_flags_settings)
@pytest.mark.django_db
def test__update_user_orgs_org_map_and_saml_attr():
"""
This combines the action of two other tests where an org membership is defined both by
the ORGANIZATION_MAP and the SOCIAL_AUTH_SAML_ORGANIZATION_ATTR at the same time
"""
# This data will make the user a member
class BackendClass:
s = {
'ORGANIZATION_MAP': {
'Default1': {
'remove': True,
'remove_admins': True,
'users': 'foobar',
'remove_users': True,
'organization_alias': 'o1_alias',
}
}
}
def setting(self, key):
return self.s[key]
backend = BackendClass()
setting = {
'saml_attr': 'memberOf',
'saml_admin_attr': 'admins',
'saml_auditor_attr': 'auditors',
'remove': True,
'remove_admins': True,
}
# This data from the server will make the user an admin of the organization
kwargs = {
'username': 'foobar',
'uid': 'idp:cmeyers@redhat.com',
'request': {u'SAMLResponse': [], u'RelayState': [u'idp']},
'is_new': False,
'response': {
'session_index': '_0728f0e0-b766-0135-75fa-02842b07c044',
'idp_name': u'idp',
'attributes': {
'admins': ['Default1'],
},
},
'social': None,
'strategy': None,
'new_association': False,
}
this_user = User.objects.create(username='foobar')
with override_settings(SOCIAL_AUTH_SAML_ORGANIZATION_ATTR=setting):
desired_org_state = {}
orgs_to_create = []
# this should add user as an admin of the org
_update_user_orgs_by_saml_attr(backend, desired_org_state, orgs_to_create, **kwargs)
assert desired_org_state['o1_alias']['admin_role'] is True
assert set(orgs_to_create) == set(['o1_alias'])
# this should add user as a member of the org without reverting the admin status
_update_user_orgs(backend, desired_org_state, orgs_to_create, this_user)
assert desired_org_state['o1_alias']['member_role'] is True
assert desired_org_state['o1_alias']['admin_role'] is True
assert set(orgs_to_create) == set(['o1_alias'])

189
awx/sso/tests/unit/test_fields.py
View File

@@ -2,192 +2,3 @@ import pytest
from rest_framework.exceptions import ValidationError
from awx.sso.fields import SAMLOrgAttrField, SAMLTeamAttrField, SAMLUserFlagsAttrField
class TestSAMLOrgAttrField:
@pytest.mark.parametrize(
"data, expected",
[
({}, {}),
({'remove': True, 'saml_attr': 'foobar'}, {'remove': True, 'saml_attr': 'foobar'}),
({'remove': True, 'saml_attr': 1234}, {'remove': True, 'saml_attr': '1234'}),
({'remove': True, 'saml_attr': 3.14}, {'remove': True, 'saml_attr': '3.14'}),
({'saml_attr': 'foobar'}, {'saml_attr': 'foobar'}),
({'remove': True}, {'remove': True}),
({'remove': True, 'saml_admin_attr': 'foobar'}, {'remove': True, 'saml_admin_attr': 'foobar'}),
({'saml_admin_attr': 'foobar'}, {'saml_admin_attr': 'foobar'}),
({'remove_admins': True, 'saml_admin_attr': 'foobar'}, {'remove_admins': True, 'saml_admin_attr': 'foobar'}),
(
{'remove': True, 'saml_attr': 'foo', 'remove_admins': True, 'saml_admin_attr': 'bar'},
{'remove': True, 'saml_attr': 'foo', 'remove_admins': True, 'saml_admin_attr': 'bar'},
),
],
)
def test_internal_value_valid(self, data, expected):
field = SAMLOrgAttrField()
res = field.to_internal_value(data)
assert res == expected
@pytest.mark.parametrize(
"data, expected",
[
({'remove': 'blah', 'saml_attr': 'foobar'}, {'remove': ['Must be a valid boolean.']}),
({'remove': True, 'saml_attr': False}, {'saml_attr': ['Not a valid string.']}),
(
{'remove': True, 'saml_attr': False, 'foo': 'bar', 'gig': 'ity'},
{'saml_attr': ['Not a valid string.'], 'foo': ['Invalid field.'], 'gig': ['Invalid field.']},
),
({'remove_admins': True, 'saml_admin_attr': False}, {'saml_admin_attr': ['Not a valid string.']}),
({'remove_admins': 'blah', 'saml_admin_attr': 'foobar'}, {'remove_admins': ['Must be a valid boolean.']}),
],
)
def test_internal_value_invalid(self, data, expected):
field = SAMLOrgAttrField()
with pytest.raises(ValidationError) as e:
field.to_internal_value(data)
assert e.value.detail == expected
class TestSAMLTeamAttrField:
@pytest.mark.parametrize(
"data",
[
{},
{'remove': True, 'saml_attr': 'foobar', 'team_org_map': []},
{'remove': True, 'saml_attr': 'foobar', 'team_org_map': [{'team': 'Engineering', 'organization': 'Ansible'}]},
{
'remove': True,
'saml_attr': 'foobar',
'team_org_map': [
{'team': 'Engineering', 'organization': 'Ansible'},
{'team': 'Engineering', 'organization': 'Ansible2'},
{'team': 'Engineering2', 'organization': 'Ansible'},
],
},
{
'remove': True,
'saml_attr': 'foobar',
'team_org_map': [
{'team': 'Engineering', 'organization': 'Ansible'},
{'team': 'Engineering', 'organization': 'Ansible2'},
{'team': 'Engineering2', 'organization': 'Ansible'},
],
},
{
'remove': True,
'saml_attr': 'foobar',
'team_org_map': [
{'team': 'Engineering', 'team_alias': 'Engineering Team', 'organization': 'Ansible'},
{'team': 'Engineering', 'organization': 'Ansible2'},
{'team': 'Engineering2', 'organization': 'Ansible'},
],
},
],
)
def test_internal_value_valid(self, data):
field = SAMLTeamAttrField()
res = field.to_internal_value(data)
assert res == data
@pytest.mark.parametrize(
"data, expected",
[
(
{'remove': True, 'saml_attr': 'foobar', 'team_org_map': [{'team': 'foobar', 'not_a_valid_key': 'blah', 'organization': 'Ansible'}]},
{'team_org_map': {0: {'not_a_valid_key': ['Invalid field.']}}},
),
(
{'remove': False, 'saml_attr': 'foobar', 'team_org_map': [{'organization': 'Ansible'}]},
{'team_org_map': {0: {'team': ['This field is required.']}}},
),
(
{'remove': False, 'saml_attr': 'foobar', 'team_org_map': [{}]},
{'team_org_map': {0: {'organization': ['This field is required.'], 'team': ['This field is required.']}}},
),
],
)
def test_internal_value_invalid(self, data, expected):
field = SAMLTeamAttrField()
with pytest.raises(ValidationError) as e:
field.to_internal_value(data)
assert e.value.detail == expected
class TestSAMLUserFlagsAttrField:
@pytest.mark.parametrize(
"data",
[
{},
{'is_superuser_attr': 'something'},
{'is_superuser_value': ['value']},
{'is_superuser_role': ['my_peeps']},
{'remove_superusers': False},
{'is_system_auditor_attr': 'something_else'},
{'is_system_auditor_value': ['value2']},
{'is_system_auditor_role': ['other_peeps']},
{'remove_system_auditors': False},
],
)
def test_internal_value_valid(self, data):
field = SAMLUserFlagsAttrField()
res = field.to_internal_value(data)
assert res == data
@pytest.mark.parametrize(
"data, expected",
[
(
{
'junk': 'something',
'is_superuser_value': 'value',
'is_superuser_role': 'my_peeps',
'is_system_auditor_attr': 'else',
'is_system_auditor_value': 'value2',
'is_system_auditor_role': 'other_peeps',
},
{
'junk': ['Invalid field.'],
'is_superuser_role': ['Expected a list of items but got type "str".'],
'is_superuser_value': ['Expected a list of items but got type "str".'],
'is_system_auditor_role': ['Expected a list of items but got type "str".'],
'is_system_auditor_value': ['Expected a list of items but got type "str".'],
},
),
(
{
'junk': 'something',
},
{
'junk': ['Invalid field.'],
},
),
(
{
'junk': 'something',
'junk2': 'else',
},
{
'junk': ['Invalid field.'],
'junk2': ['Invalid field.'],
},
),
# make sure we can't pass a string to the boolean fields
(
{
'remove_superusers': 'test',
'remove_system_auditors': 'test',
},
{
"remove_superusers": ["Must be a valid boolean."],
"remove_system_auditors": ["Must be a valid boolean."],
},
),
],
)
def test_internal_value_invalid(self, data, expected):
field = SAMLUserFlagsAttrField()
with pytest.raises(ValidationError) as e:
field.to_internal_value(data)
print(e.value.detail)
assert e.value.detail == expected

1
awx/sso/tests/unit/test_pipelines.py
View File

@@ -4,7 +4,6 @@ import pytest
@pytest.mark.parametrize(
"lib",
[
("saml_pipeline"),
("social_pipeline"),
],
)

3
awx/sso/urls.py
View File

@@ -3,7 +3,7 @@
from django.urls import re_path
from awx.sso.views import sso_complete, sso_error, sso_inactive, saml_metadata
from awx.sso.views import sso_complete, sso_error, sso_inactive
app_name = 'sso'
@@ -11,5 +11,4 @@ urlpatterns = [
re_path(r'^complete/$', sso_complete, name='sso_complete'),
re_path(r'^error/$', sso_error, name='sso_error'),
re_path(r'^inactive/$', sso_inactive, name='sso_inactive'),
re_path(r'^metadata/saml/$', saml_metadata, name='saml_metadata'),
]

22
awx/sso/views.py
View File

@@ -7,8 +7,6 @@ import logging
# Django
from django.urls import reverse
from django.http import HttpResponse
from django.views.generic import View
from django.views.generic.base import RedirectView
from django.utils.encoding import smart_str
from django.conf import settings
@@ -46,23 +44,3 @@ class CompleteView(BaseRedirectView):
sso_complete = CompleteView.as_view()
class MetadataView(View):
def get(self, request, *args, **kwargs):
from social_django.utils import load_backend, load_strategy
complete_url = reverse('social:complete', args=('saml',))
try:
saml_backend = load_backend(load_strategy(request), 'saml', redirect_uri=complete_url)
metadata, errors = saml_backend.generate_metadata_xml()
except Exception as e:
logger.exception('unable to generate SAML metadata')
errors = e
if not errors:
return HttpResponse(content=metadata, content_type='text/xml')
else:
return HttpResponse(content=str(errors), content_type='text/plain')
saml_metadata = MetadataView.as_view()