"""LDAP and Outpost e2e tests""" from dataclasses import asdict from sys import platform from time import sleep from unittest.case import skipUnless from docker.client import DockerClient, from_env from docker.models.containers import Container from guardian.shortcuts import get_anonymous_user from ldap3 import ALL, ALL_ATTRIBUTES, ALL_OPERATIONAL_ATTRIBUTES, SUBTREE, Connection, Server from ldap3.core.exceptions import LDAPInvalidCredentialsResult from authentik.core.models import Application, User from authentik.core.tests.utils import create_test_admin_user from authentik.events.models import Event, EventAction from authentik.flows.models import Flow from authentik.outposts.managed import MANAGED_OUTPOST from authentik.outposts.models import Outpost, OutpostConfig, OutpostType from authentik.providers.ldap.models import LDAPProvider, SearchModes from tests.e2e.utils import SeleniumTestCase, apply_migration, object_manager, retry @skipUnless(platform.startswith("linux"), "requires local docker") class TestProviderLDAP(SeleniumTestCase): """LDAP and Outpost e2e tests""" ldap_container: Container def tearDown(self) -> None: super().tearDown() self.output_container_logs(self.ldap_container) self.ldap_container.kill() def start_ldap(self, outpost: Outpost) -> Container: """Start ldap container based on outpost created""" client: DockerClient = from_env() container = client.containers.run( image=self.get_container_image("goauthentik.io/dev-ldap"), detach=True, network_mode="host", auto_remove=True, environment={ "AUTHENTIK_HOST": self.live_server_url, "AUTHENTIK_TOKEN": outpost.token.key, }, ) return container def _prepare(self) -> User: """prepare user, provider, app and container""" # set additionalHeaders to test later user = create_test_admin_user() user.attributes["extraAttribute"] = "bar" user.save() ldap: LDAPProvider = LDAPProvider.objects.create( name="ldap_provider", authorization_flow=Flow.objects.get(slug="default-authentication-flow"), search_group=user.ak_groups.first(), search_mode=SearchModes.CACHED, ) # we need to create an application to actually access the ldap Application.objects.create(name="ldap", slug="ldap", provider=ldap) outpost: Outpost = Outpost.objects.create( name="ldap_outpost", type=OutpostType.LDAP, _config=asdict(OutpostConfig(log_level="debug")), ) outpost.providers.add(ldap) outpost.save() self.ldap_container = self.start_ldap(outpost) # Wait until outpost healthcheck succeeds healthcheck_retries = 0 while healthcheck_retries < 50: if len(outpost.state) > 0: state = outpost.state[0] if state.last_seen: break healthcheck_retries += 1 sleep(0.5) return user, outpost @retry() @apply_migration("authentik_core", "0002_auto_20200523_1133_squashed_0011_provider_name_temp") @apply_migration("authentik_flows", "0008_default_flows") @object_manager def test_ldap_bind_success(self): """Test simple bind""" user, _ = self._prepare() server = Server("ldap://localhost:3389", get_info=ALL) _connection = Connection( server, raise_exceptions=True, user=f"cn={user.username},ou=users,DC=ldap,DC=goauthentik,DC=io", password=user.username, ) _connection.bind() self.assertTrue( Event.objects.filter( action=EventAction.LOGIN, user={ "pk": user.pk, "email": user.email, "username": user.username, }, ) ) @retry() @apply_migration("authentik_core", "0002_auto_20200523_1133_squashed_0011_provider_name_temp") @apply_migration("authentik_flows", "0008_default_flows") @object_manager def test_ldap_bind_success_ssl(self): """Test simple bind with ssl""" user, _ = self._prepare() server = Server("ldaps://localhost:6636", get_info=ALL) _connection = Connection( server, raise_exceptions=True, user=f"cn={user.username},ou=users,DC=ldap,DC=goauthentik,DC=io", password=user.username, ) _connection.bind() self.assertTrue( Event.objects.filter( action=EventAction.LOGIN, user={ "pk": user.pk, "email": user.email, "username": user.username, }, ) ) @retry() @apply_migration("authentik_core", "0002_auto_20200523_1133_squashed_0011_provider_name_temp") @apply_migration("authentik_flows", "0008_default_flows") @object_manager def test_ldap_bind_fail(self): """Test simple bind (failed)""" user, _ = self._prepare() server = Server("ldap://localhost:3389", get_info=ALL) _connection = Connection( server, raise_exceptions=True, user=f"cn={user.username},ou=users,DC=ldap,DC=goauthentik,DC=io", password=user.username + "fqwerwqer", ) with self.assertRaises(LDAPInvalidCredentialsResult): _connection.bind() anon = get_anonymous_user() self.assertTrue( Event.objects.filter( action=EventAction.LOGIN_FAILED, user={"pk": anon.pk, "email": anon.email, "username": anon.username}, ) ) @retry() @apply_migration("authentik_core", "0002_auto_20200523_1133_squashed_0011_provider_name_temp") @apply_migration("authentik_flows", "0008_default_flows") @object_manager def test_ldap_bind_search(self): """Test simple bind + search""" user, outpost = self._prepare() server = Server("ldap://localhost:3389", get_info=ALL) _connection = Connection( server, raise_exceptions=True, user=f"cn={user.username},ou=users,dc=ldap,dc=goauthentik,dc=io", password=user.username, ) _connection.bind() self.assertTrue( Event.objects.filter( action=EventAction.LOGIN, user={ "pk": user.pk, "email": user.email, "username": user.username, }, ) ) embedded_account = Outpost.objects.filter(managed=MANAGED_OUTPOST).first().user _connection.search( "ou=users,dc=ldap,dc=goauthentik,dc=io", "(objectClass=user)", search_scope=SUBTREE, attributes=[ALL_ATTRIBUTES, ALL_OPERATIONAL_ATTRIBUTES], ) response = _connection.response # Remove raw_attributes to make checking easier for obj in response: del obj["raw_attributes"] del obj["raw_dn"] o_user = outpost.suer self.assertCountEqual( response, [ { "dn": f"cn={o_user.username},ou=users,dc=ldap,dc=goauthentik,dc=io", "attributes": { "cn": [o_user.username], "sAMAccountName": [o_user.username], "uid": [o_user.uid], "name": [""], "displayName": [""], "mail": [""], "objectClass": [ "user", "organizationalPerson", "inetOrgPerson", "goauthentik.io/ldap/user", ], "uidNumber": [str(2000 + o_user.pk)], "gidNumber": [str(2000 + o_user.pk)], "memberOf": [], "accountStatus": ["true"], "superuser": ["false"], "goauthentik.io/ldap/active": ["true"], "goauthentik.io/ldap/superuser": ["false"], "goauthentik.io/user/override-ips": ["true"], "goauthentik.io/user/service-account": ["true"], }, "type": "searchResEntry", }, { "dn": f"cn={embedded_account.username},ou=users,dc=ldap,dc=goauthentik,dc=io", "attributes": { "cn": [embedded_account.username], "sAMAccountName": [embedded_account.username], "uid": [embedded_account.uid], "name": [""], "displayName": [""], "mail": [""], "objectClass": [ "user", "organizationalPerson", "inetOrgPerson", "goauthentik.io/ldap/user", ], "uidNumber": [str(2000 + embedded_account.pk)], "gidNumber": [str(2000 + embedded_account.pk)], "memberOf": [], "accountStatus": ["true"], "superuser": ["false"], "goauthentik.io/ldap/active": ["true"], "goauthentik.io/ldap/superuser": ["false"], "goauthentik.io/user/override-ips": ["true"], "goauthentik.io/user/service-account": ["true"], }, "type": "searchResEntry", }, { "dn": f"cn={user.username},ou=users,dc=ldap,dc=goauthentik,dc=io", "attributes": { "cn": [user.username], "sAMAccountName": [user.username], "uid": [user.uid], "name": [user.name], "displayName": [user.name], "mail": [user.email], "objectClass": [ "user", "organizationalPerson", "inetOrgPerson", "goauthentik.io/ldap/user", ], "uidNumber": [str(2000 + user.pk)], "gidNumber": [str(2000 + user.pk)], "memberOf": [ f"cn={group.name},ou=groups,dc=ldap,dc=goauthentik,dc=io" for group in user.ak_groups.all() ], "accountStatus": ["true"], "superuser": ["true"], "goauthentik.io/ldap/active": ["true"], "goauthentik.io/ldap/superuser": ["true"], "extraAttribute": ["bar"], }, "type": "searchResEntry", }, ], )