tests/e2e: ensure akadmin is not used

Signed-off-by: Jens Langhammer <jens.langhammer@beryju.org>
This commit is contained in:
Jens Langhammer 2021-11-23 21:30:02 +01:00
parent b2d2e7cbc8
commit 060f19ce06
12 changed files with 122 additions and 126 deletions

View file

@ -17,7 +17,7 @@ from authentik.flows.models import Flow, FlowStageBinding
from authentik.stages.authenticator_static.models import AuthenticatorStaticStage from authentik.stages.authenticator_static.models import AuthenticatorStaticStage
from authentik.stages.authenticator_totp.models import AuthenticatorTOTPStage from authentik.stages.authenticator_totp.models import AuthenticatorTOTPStage
from authentik.stages.authenticator_validate.models import AuthenticatorValidateStage from authentik.stages.authenticator_validate.models import AuthenticatorValidateStage
from tests.e2e.utils import USER, SeleniumTestCase, apply_migration, retry from tests.e2e.utils import SeleniumTestCase, apply_migration, retry
@skipUnless(platform.startswith("linux"), "requires local docker") @skipUnless(platform.startswith("linux"), "requires local docker")
@ -32,8 +32,7 @@ class TestFlowsAuthenticator(SeleniumTestCase):
"""test flow with otp stages""" """test flow with otp stages"""
sleep(1) sleep(1)
# Setup TOTP Device # Setup TOTP Device
user = USER() device = TOTPDevice.objects.create(user=self.user, confirmed=True, digits=6)
device = TOTPDevice.objects.create(user=user, confirmed=True, digits=6)
flow: Flow = Flow.objects.get(slug="default-authentication-flow") flow: Flow = Flow.objects.get(slug="default-authentication-flow")
FlowStageBinding.objects.create( FlowStageBinding.objects.create(
@ -53,7 +52,7 @@ class TestFlowsAuthenticator(SeleniumTestCase):
code_stage.find_element(By.CSS_SELECTOR, "input[name=code]").send_keys(totp.token()) code_stage.find_element(By.CSS_SELECTOR, "input[name=code]").send_keys(totp.token())
code_stage.find_element(By.CSS_SELECTOR, "input[name=code]").send_keys(Keys.ENTER) code_stage.find_element(By.CSS_SELECTOR, "input[name=code]").send_keys(Keys.ENTER)
self.wait_for_url(self.if_user_url("/library")) self.wait_for_url(self.if_user_url("/library"))
self.assert_user(USER()) self.assert_user(self.user)
@retry() @retry()
@apply_migration("authentik_core", "0002_auto_20200523_1133_squashed_0011_provider_name_temp") @apply_migration("authentik_core", "0002_auto_20200523_1133_squashed_0011_provider_name_temp")
@ -68,7 +67,7 @@ class TestFlowsAuthenticator(SeleniumTestCase):
self.login() self.login()
self.wait_for_url(self.if_user_url("/library")) self.wait_for_url(self.if_user_url("/library"))
self.assert_user(USER()) self.assert_user(self.user)
self.driver.get( self.driver.get(
self.url( self.url(
@ -98,7 +97,7 @@ class TestFlowsAuthenticator(SeleniumTestCase):
totp_stage.find_element(By.CSS_SELECTOR, "input[name=code]").send_keys(Keys.ENTER) totp_stage.find_element(By.CSS_SELECTOR, "input[name=code]").send_keys(Keys.ENTER)
sleep(3) sleep(3)
self.assertTrue(TOTPDevice.objects.filter(user=USER(), confirmed=True).exists()) self.assertTrue(TOTPDevice.objects.filter(user=self.user, confirmed=True).exists())
@retry() @retry()
@apply_migration("authentik_core", "0002_auto_20200523_1133_squashed_0011_provider_name_temp") @apply_migration("authentik_core", "0002_auto_20200523_1133_squashed_0011_provider_name_temp")
@ -113,7 +112,7 @@ class TestFlowsAuthenticator(SeleniumTestCase):
self.login() self.login()
self.wait_for_url(self.if_user_url("/library")) self.wait_for_url(self.if_user_url("/library"))
self.assert_user(USER()) self.assert_user(self.user)
self.driver.get( self.driver.get(
self.url( self.url(
@ -134,6 +133,6 @@ class TestFlowsAuthenticator(SeleniumTestCase):
self.wait_for_url(destination_url) self.wait_for_url(destination_url)
sleep(1) sleep(1)
self.assertTrue(StaticDevice.objects.filter(user=USER(), confirmed=True).exists()) self.assertTrue(StaticDevice.objects.filter(user=self.user, confirmed=True).exists())
device = StaticDevice.objects.filter(user=USER(), confirmed=True).first() device = StaticDevice.objects.filter(user=self.user, confirmed=True).first()
self.assertTrue(StaticToken.objects.filter(token=token, device=device).exists()) self.assertTrue(StaticToken.objects.filter(token=token, device=device).exists())

View file

@ -16,7 +16,7 @@ from authentik.stages.identification.models import IdentificationStage
from authentik.stages.prompt.models import FieldTypes, Prompt, PromptStage from authentik.stages.prompt.models import FieldTypes, Prompt, PromptStage
from authentik.stages.user_login.models import UserLoginStage from authentik.stages.user_login.models import UserLoginStage
from authentik.stages.user_write.models import UserWriteStage from authentik.stages.user_write.models import UserWriteStage
from tests.e2e.utils import USER, SeleniumTestCase, apply_migration, retry from tests.e2e.utils import SeleniumTestCase, apply_migration, retry
@skipUnless(platform.startswith("linux"), "requires local docker") @skipUnless(platform.startswith("linux"), "requires local docker")
@ -218,10 +218,10 @@ class TestFlowsEnroll(SeleniumTestCase):
wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, "input[name=username]"))) wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, "input[name=username]")))
prompt_stage.find_element(By.CSS_SELECTOR, "input[name=username]").send_keys("foo") prompt_stage.find_element(By.CSS_SELECTOR, "input[name=username]").send_keys("foo")
prompt_stage.find_element(By.CSS_SELECTOR, "input[name=password]").send_keys( prompt_stage.find_element(By.CSS_SELECTOR, "input[name=password]").send_keys(
USER().username self.user.username
) )
prompt_stage.find_element(By.CSS_SELECTOR, "input[name=password_repeat]").send_keys( prompt_stage.find_element(By.CSS_SELECTOR, "input[name=password_repeat]").send_keys(
USER().username self.user.username
) )
prompt_stage.find_element(By.CSS_SELECTOR, ".pf-c-button").click() prompt_stage.find_element(By.CSS_SELECTOR, ".pf-c-button").click()

View file

@ -2,7 +2,7 @@
from sys import platform from sys import platform
from unittest.case import skipUnless from unittest.case import skipUnless
from tests.e2e.utils import USER, SeleniumTestCase, apply_migration, retry from tests.e2e.utils import SeleniumTestCase, apply_migration, retry
@skipUnless(platform.startswith("linux"), "requires local docker") @skipUnless(platform.startswith("linux"), "requires local docker")
@ -23,4 +23,4 @@ class TestFlowsLogin(SeleniumTestCase):
) )
self.login() self.login()
self.wait_for_url(self.if_user_url("/library")) self.wait_for_url(self.if_user_url("/library"))
self.assert_user(USER()) self.assert_user(self.user)

View file

@ -9,7 +9,7 @@ from authentik.core.models import User
from authentik.flows.models import Flow, FlowDesignation from authentik.flows.models import Flow, FlowDesignation
from authentik.lib.generators import generate_key from authentik.lib.generators import generate_key
from authentik.stages.password.models import PasswordStage from authentik.stages.password.models import PasswordStage
from tests.e2e.utils import USER, SeleniumTestCase, apply_migration, retry from tests.e2e.utils import SeleniumTestCase, apply_migration, retry
@skipUnless(platform.startswith("linux"), "requires local docker") @skipUnless(platform.startswith("linux"), "requires local docker")
@ -63,6 +63,6 @@ class TestFlowsStageSetup(SeleniumTestCase):
) )
self.wait_for_url(self.if_user_url("/library")) self.wait_for_url(self.if_user_url("/library"))
# Because USER() is cached, we need to get the user manually here # Because self.user is cached, we need to get the user manually here
user = User.objects.get(username=USER().username) user = User.objects.get(username=self.user.username)
self.assertTrue(user.check_password(new_password)) self.assertTrue(user.check_password(new_password))

View file

@ -11,7 +11,6 @@ from ldap3 import ALL, ALL_ATTRIBUTES, ALL_OPERATIONAL_ATTRIBUTES, SUBTREE, Conn
from ldap3.core.exceptions import LDAPInvalidCredentialsResult from ldap3.core.exceptions import LDAPInvalidCredentialsResult
from authentik.core.models import Application, User 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.events.models import Event, EventAction
from authentik.flows.models import Flow from authentik.flows.models import Flow
from authentik.outposts.managed import MANAGED_OUTPOST from authentik.outposts.managed import MANAGED_OUTPOST
@ -49,14 +48,13 @@ class TestProviderLDAP(SeleniumTestCase):
def _prepare(self) -> User: def _prepare(self) -> User:
"""prepare user, provider, app and container""" """prepare user, provider, app and container"""
# set additionalHeaders to test later # set additionalHeaders to test later
user = create_test_admin_user(set_password=True) self.user.attributes["extraAttribute"] = "bar"
user.attributes["extraAttribute"] = "bar" self.user.save()
user.save()
ldap: LDAPProvider = LDAPProvider.objects.create( ldap: LDAPProvider = LDAPProvider.objects.create(
name="ldap_provider", name="ldap_provider",
authorization_flow=Flow.objects.get(slug="default-authentication-flow"), authorization_flow=Flow.objects.get(slug="default-authentication-flow"),
search_group=user.ak_groups.first(), search_group=self.user.ak_groups.first(),
search_mode=SearchModes.CACHED, search_mode=SearchModes.CACHED,
) )
# we need to create an application to actually access the ldap # we need to create an application to actually access the ldap
@ -80,7 +78,7 @@ class TestProviderLDAP(SeleniumTestCase):
healthcheck_retries += 1 healthcheck_retries += 1
sleep(0.5) sleep(0.5)
sleep(5) sleep(5)
return user, outpost return outpost
@retry() @retry()
@apply_migration("authentik_core", "0002_auto_20200523_1133_squashed_0011_provider_name_temp") @apply_migration("authentik_core", "0002_auto_20200523_1133_squashed_0011_provider_name_temp")
@ -88,22 +86,22 @@ class TestProviderLDAP(SeleniumTestCase):
@object_manager @object_manager
def test_ldap_bind_success(self): def test_ldap_bind_success(self):
"""Test simple bind""" """Test simple bind"""
user, _ = self._prepare() self._prepare()
server = Server("ldap://localhost:3389", get_info=ALL) server = Server("ldap://localhost:3389", get_info=ALL)
_connection = Connection( _connection = Connection(
server, server,
raise_exceptions=True, raise_exceptions=True,
user=f"cn={user.username},ou=users,DC=ldap,DC=goauthentik,DC=io", user=f"cn={self.user.username},ou=users,DC=ldap,DC=goauthentik,DC=io",
password=user.username, password=self.user.username,
) )
_connection.bind() _connection.bind()
self.assertTrue( self.assertTrue(
Event.objects.filter( Event.objects.filter(
action=EventAction.LOGIN, action=EventAction.LOGIN,
user={ user={
"pk": user.pk, "pk": self.user.pk,
"email": user.email, "email": self.user.email,
"username": user.username, "username": self.user.username,
}, },
) )
) )
@ -114,22 +112,22 @@ class TestProviderLDAP(SeleniumTestCase):
@object_manager @object_manager
def test_ldap_bind_success_ssl(self): def test_ldap_bind_success_ssl(self):
"""Test simple bind with ssl""" """Test simple bind with ssl"""
user, _ = self._prepare() self._prepare()
server = Server("ldaps://localhost:6636", get_info=ALL) server = Server("ldaps://localhost:6636", get_info=ALL)
_connection = Connection( _connection = Connection(
server, server,
raise_exceptions=True, raise_exceptions=True,
user=f"cn={user.username},ou=users,DC=ldap,DC=goauthentik,DC=io", user=f"cn={self.user.username},ou=users,DC=ldap,DC=goauthentik,DC=io",
password=user.username, password=self.user.username,
) )
_connection.bind() _connection.bind()
self.assertTrue( self.assertTrue(
Event.objects.filter( Event.objects.filter(
action=EventAction.LOGIN, action=EventAction.LOGIN,
user={ user={
"pk": user.pk, "pk": self.user.pk,
"email": user.email, "email": self.user.email,
"username": user.username, "username": self.user.username,
}, },
) )
) )
@ -140,13 +138,13 @@ class TestProviderLDAP(SeleniumTestCase):
@object_manager @object_manager
def test_ldap_bind_fail(self): def test_ldap_bind_fail(self):
"""Test simple bind (failed)""" """Test simple bind (failed)"""
user, _ = self._prepare() self._prepare()
server = Server("ldap://localhost:3389", get_info=ALL) server = Server("ldap://localhost:3389", get_info=ALL)
_connection = Connection( _connection = Connection(
server, server,
raise_exceptions=True, raise_exceptions=True,
user=f"cn={user.username},ou=users,DC=ldap,DC=goauthentik,DC=io", user=f"cn={self.user.username},ou=users,DC=ldap,DC=goauthentik,DC=io",
password=user.username + "fqwerwqer", password=self.user.username + "fqwerwqer",
) )
with self.assertRaises(LDAPInvalidCredentialsResult): with self.assertRaises(LDAPInvalidCredentialsResult):
_connection.bind() _connection.bind()
@ -164,27 +162,29 @@ class TestProviderLDAP(SeleniumTestCase):
@object_manager @object_manager
def test_ldap_bind_search(self): def test_ldap_bind_search(self):
"""Test simple bind + search""" """Test simple bind + search"""
user, outpost = self._prepare() outpost = self._prepare()
server = Server("ldap://localhost:3389", get_info=ALL) server = Server("ldap://localhost:3389", get_info=ALL)
_connection = Connection( _connection = Connection(
server, server,
raise_exceptions=True, raise_exceptions=True,
user=f"cn={user.username},ou=users,dc=ldap,dc=goauthentik,dc=io", user=f"cn={self.user.username},ou=users,dc=ldap,dc=goauthentik,dc=io",
password=user.username, password=self.user.username,
) )
_connection.bind() _connection.bind()
self.assertTrue( self.assertTrue(
Event.objects.filter( Event.objects.filter(
action=EventAction.LOGIN, action=EventAction.LOGIN,
user={ user={
"pk": user.pk, "pk": self.user.pk,
"email": user.email, "email": self.user.email,
"username": user.username, "username": self.user.username,
}, },
) )
) )
embedded_account = Outpost.objects.filter(managed=MANAGED_OUTPOST).first().user embedded_account = Outpost.objects.filter(managed=MANAGED_OUTPOST).first().user
# Remove akadmin to ensure list is correct
User.objects.filter(username="akadmin").delete()
_connection.search( _connection.search(
"ou=users,dc=ldap,dc=goauthentik,dc=io", "ou=users,dc=ldap,dc=goauthentik,dc=io",
@ -256,25 +256,25 @@ class TestProviderLDAP(SeleniumTestCase):
"type": "searchResEntry", "type": "searchResEntry",
}, },
{ {
"dn": f"cn={user.username},ou=users,dc=ldap,dc=goauthentik,dc=io", "dn": f"cn={self.user.username},ou=users,dc=ldap,dc=goauthentik,dc=io",
"attributes": { "attributes": {
"cn": [user.username], "cn": [self.user.username],
"sAMAccountName": [user.username], "sAMAccountName": [self.user.username],
"uid": [user.uid], "uid": [self.user.uid],
"name": [user.name], "name": [self.user.name],
"displayName": [user.name], "displayName": [self.user.name],
"mail": [user.email], "mail": [self.user.email],
"objectClass": [ "objectClass": [
"user", "user",
"organizationalPerson", "organizationalPerson",
"inetOrgPerson", "inetOrgPerson",
"goauthentik.io/ldap/user", "goauthentik.io/ldap/user",
], ],
"uidNumber": [str(2000 + user.pk)], "uidNumber": [str(2000 + self.user.pk)],
"gidNumber": [str(2000 + user.pk)], "gidNumber": [str(2000 + self.user.pk)],
"memberOf": [ "memberOf": [
f"cn={group.name},ou=groups,dc=ldap,dc=goauthentik,dc=io" f"cn={group.name},ou=groups,dc=ldap,dc=goauthentik,dc=io"
for group in user.ak_groups.all() for group in self.user.ak_groups.all()
], ],
"accountStatus": ["true"], "accountStatus": ["true"],
"superuser": ["true"], "superuser": ["true"],

View file

@ -14,7 +14,7 @@ from authentik.lib.generators import generate_id, generate_key
from authentik.policies.expression.models import ExpressionPolicy from authentik.policies.expression.models import ExpressionPolicy
from authentik.policies.models import PolicyBinding from authentik.policies.models import PolicyBinding
from authentik.providers.oauth2.models import ClientTypes, OAuth2Provider from authentik.providers.oauth2.models import ClientTypes, OAuth2Provider
from tests.e2e.utils import USER, SeleniumTestCase, apply_migration, retry from tests.e2e.utils import SeleniumTestCase, apply_migration, retry
@skipUnless(platform.startswith("linux"), "requires local docker") @skipUnless(platform.startswith("linux"), "requires local docker")
@ -89,19 +89,19 @@ class TestProviderOAuth2Github(SeleniumTestCase):
self.driver.get("http://localhost:3000/profile") self.driver.get("http://localhost:3000/profile")
self.assertEqual( self.assertEqual(
self.driver.find_element(By.CLASS_NAME, "page-header__title").text, self.driver.find_element(By.CLASS_NAME, "page-header__title").text,
USER().username, self.user.username,
) )
self.assertEqual( self.assertEqual(
self.driver.find_element(By.CSS_SELECTOR, "input[name=name]").get_attribute("value"), self.driver.find_element(By.CSS_SELECTOR, "input[name=name]").get_attribute("value"),
USER().username, self.user.username,
) )
self.assertEqual( self.assertEqual(
self.driver.find_element(By.CSS_SELECTOR, "input[name=email]").get_attribute("value"), self.driver.find_element(By.CSS_SELECTOR, "input[name=email]").get_attribute("value"),
USER().email, self.user.email,
) )
self.assertEqual( self.assertEqual(
self.driver.find_element(By.CSS_SELECTOR, "input[name=login]").get_attribute("value"), self.driver.find_element(By.CSS_SELECTOR, "input[name=login]").get_attribute("value"),
USER().username, self.user.username,
) )
@retry() @retry()
@ -157,19 +157,19 @@ class TestProviderOAuth2Github(SeleniumTestCase):
self.driver.get("http://localhost:3000/profile") self.driver.get("http://localhost:3000/profile")
self.assertEqual( self.assertEqual(
self.driver.find_element(By.CLASS_NAME, "page-header__title").text, self.driver.find_element(By.CLASS_NAME, "page-header__title").text,
USER().username, self.user.username,
) )
self.assertEqual( self.assertEqual(
self.driver.find_element(By.CSS_SELECTOR, "input[name=name]").get_attribute("value"), self.driver.find_element(By.CSS_SELECTOR, "input[name=name]").get_attribute("value"),
USER().username, self.user.username,
) )
self.assertEqual( self.assertEqual(
self.driver.find_element(By.CSS_SELECTOR, "input[name=email]").get_attribute("value"), self.driver.find_element(By.CSS_SELECTOR, "input[name=email]").get_attribute("value"),
USER().email, self.user.email,
) )
self.assertEqual( self.assertEqual(
self.driver.find_element(By.CSS_SELECTOR, "input[name=login]").get_attribute("value"), self.driver.find_element(By.CSS_SELECTOR, "input[name=login]").get_attribute("value"),
USER().username, self.user.username,
) )
@retry() @retry()

View file

@ -20,7 +20,7 @@ from authentik.providers.oauth2.constants import (
SCOPE_OPENID_PROFILE, SCOPE_OPENID_PROFILE,
) )
from authentik.providers.oauth2.models import ClientTypes, OAuth2Provider, ScopeMapping from authentik.providers.oauth2.models import ClientTypes, OAuth2Provider, ScopeMapping
from tests.e2e.utils import USER, SeleniumTestCase, apply_migration, object_manager, retry from tests.e2e.utils import SeleniumTestCase, apply_migration, object_manager, retry
@skipUnless(platform.startswith("linux"), "requires local docker") @skipUnless(platform.startswith("linux"), "requires local docker")
@ -147,19 +147,19 @@ class TestProviderOAuth2OAuth(SeleniumTestCase):
self.driver.get("http://localhost:3000/profile") self.driver.get("http://localhost:3000/profile")
self.assertEqual( self.assertEqual(
self.driver.find_element(By.CLASS_NAME, "page-header__title").text, self.driver.find_element(By.CLASS_NAME, "page-header__title").text,
USER().name, self.user.name,
) )
self.assertEqual( self.assertEqual(
self.driver.find_element(By.CSS_SELECTOR, "input[name=name]").get_attribute("value"), self.driver.find_element(By.CSS_SELECTOR, "input[name=name]").get_attribute("value"),
USER().name, self.user.name,
) )
self.assertEqual( self.assertEqual(
self.driver.find_element(By.CSS_SELECTOR, "input[name=email]").get_attribute("value"), self.driver.find_element(By.CSS_SELECTOR, "input[name=email]").get_attribute("value"),
USER().email, self.user.email,
) )
self.assertEqual( self.assertEqual(
self.driver.find_element(By.CSS_SELECTOR, "input[name=login]").get_attribute("value"), self.driver.find_element(By.CSS_SELECTOR, "input[name=login]").get_attribute("value"),
USER().email, self.user.email,
) )
@retry() @retry()
@ -204,19 +204,19 @@ class TestProviderOAuth2OAuth(SeleniumTestCase):
self.driver.get("http://localhost:3000/profile") self.driver.get("http://localhost:3000/profile")
self.assertEqual( self.assertEqual(
self.driver.find_element(By.CLASS_NAME, "page-header__title").text, self.driver.find_element(By.CLASS_NAME, "page-header__title").text,
USER().name, self.user.name,
) )
self.assertEqual( self.assertEqual(
self.driver.find_element(By.CSS_SELECTOR, "input[name=name]").get_attribute("value"), self.driver.find_element(By.CSS_SELECTOR, "input[name=name]").get_attribute("value"),
USER().name, self.user.name,
) )
self.assertEqual( self.assertEqual(
self.driver.find_element(By.CSS_SELECTOR, "input[name=email]").get_attribute("value"), self.driver.find_element(By.CSS_SELECTOR, "input[name=email]").get_attribute("value"),
USER().email, self.user.email,
) )
self.assertEqual( self.assertEqual(
self.driver.find_element(By.CSS_SELECTOR, "input[name=login]").get_attribute("value"), self.driver.find_element(By.CSS_SELECTOR, "input[name=login]").get_attribute("value"),
USER().email, self.user.email,
) )
self.driver.get("http://localhost:3000/logout") self.driver.get("http://localhost:3000/logout")
self.wait_for_url( self.wait_for_url(
@ -286,19 +286,19 @@ class TestProviderOAuth2OAuth(SeleniumTestCase):
self.assertEqual( self.assertEqual(
self.driver.find_element(By.CLASS_NAME, "page-header__title").text, self.driver.find_element(By.CLASS_NAME, "page-header__title").text,
USER().name, self.user.name,
) )
self.assertEqual( self.assertEqual(
self.driver.find_element(By.CSS_SELECTOR, "input[name=name]").get_attribute("value"), self.driver.find_element(By.CSS_SELECTOR, "input[name=name]").get_attribute("value"),
USER().name, self.user.name,
) )
self.assertEqual( self.assertEqual(
self.driver.find_element(By.CSS_SELECTOR, "input[name=email]").get_attribute("value"), self.driver.find_element(By.CSS_SELECTOR, "input[name=email]").get_attribute("value"),
USER().email, self.user.email,
) )
self.assertEqual( self.assertEqual(
self.driver.find_element(By.CSS_SELECTOR, "input[name=login]").get_attribute("value"), self.driver.find_element(By.CSS_SELECTOR, "input[name=login]").get_attribute("value"),
USER().email, self.user.email,
) )
@retry() @retry()

View file

@ -22,7 +22,7 @@ from authentik.providers.oauth2.constants import (
SCOPE_OPENID_PROFILE, SCOPE_OPENID_PROFILE,
) )
from authentik.providers.oauth2.models import ClientTypes, OAuth2Provider, ScopeMapping from authentik.providers.oauth2.models import ClientTypes, OAuth2Provider, ScopeMapping
from tests.e2e.utils import USER, SeleniumTestCase, apply_migration, object_manager, retry from tests.e2e.utils import SeleniumTestCase, apply_migration, object_manager, retry
@skipUnless(platform.startswith("linux"), "requires local docker") @skipUnless(platform.startswith("linux"), "requires local docker")
@ -146,14 +146,14 @@ class TestProviderOAuth2OIDC(SeleniumTestCase):
self.wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, "pre"))) self.wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, "pre")))
body = loads(self.driver.find_element(By.CSS_SELECTOR, "pre").text) body = loads(self.driver.find_element(By.CSS_SELECTOR, "pre").text)
self.assertEqual(body["IDTokenClaims"]["nickname"], USER().username) self.assertEqual(body["IDTokenClaims"]["nickname"], self.user.username)
self.assertEqual(body["UserInfo"]["nickname"], USER().username) self.assertEqual(body["UserInfo"]["nickname"], self.user.username)
self.assertEqual(body["IDTokenClaims"]["name"], USER().name) self.assertEqual(body["IDTokenClaims"]["name"], self.user.name)
self.assertEqual(body["UserInfo"]["name"], USER().name) self.assertEqual(body["UserInfo"]["name"], self.user.name)
self.assertEqual(body["IDTokenClaims"]["email"], USER().email) self.assertEqual(body["IDTokenClaims"]["email"], self.user.email)
self.assertEqual(body["UserInfo"]["email"], USER().email) self.assertEqual(body["UserInfo"]["email"], self.user.email)
@retry() @retry()
@apply_migration("authentik_core", "0002_auto_20200523_1133_squashed_0011_provider_name_temp") @apply_migration("authentik_core", "0002_auto_20200523_1133_squashed_0011_provider_name_temp")
@ -211,14 +211,14 @@ class TestProviderOAuth2OIDC(SeleniumTestCase):
self.wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, "pre"))) self.wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, "pre")))
body = loads(self.driver.find_element(By.CSS_SELECTOR, "pre").text) body = loads(self.driver.find_element(By.CSS_SELECTOR, "pre").text)
self.assertEqual(body["IDTokenClaims"]["nickname"], USER().username) self.assertEqual(body["IDTokenClaims"]["nickname"], self.user.username)
self.assertEqual(body["UserInfo"]["nickname"], USER().username) self.assertEqual(body["UserInfo"]["nickname"], self.user.username)
self.assertEqual(body["IDTokenClaims"]["name"], USER().name) self.assertEqual(body["IDTokenClaims"]["name"], self.user.name)
self.assertEqual(body["UserInfo"]["name"], USER().name) self.assertEqual(body["UserInfo"]["name"], self.user.name)
self.assertEqual(body["IDTokenClaims"]["email"], USER().email) self.assertEqual(body["IDTokenClaims"]["email"], self.user.email)
self.assertEqual(body["UserInfo"]["email"], USER().email) self.assertEqual(body["UserInfo"]["email"], self.user.email)
@retry() @retry()
@apply_migration("authentik_core", "0002_auto_20200523_1133_squashed_0011_provider_name_temp") @apply_migration("authentik_core", "0002_auto_20200523_1133_squashed_0011_provider_name_temp")

View file

@ -22,7 +22,7 @@ from authentik.providers.oauth2.constants import (
SCOPE_OPENID_PROFILE, SCOPE_OPENID_PROFILE,
) )
from authentik.providers.oauth2.models import ClientTypes, OAuth2Provider, ScopeMapping from authentik.providers.oauth2.models import ClientTypes, OAuth2Provider, ScopeMapping
from tests.e2e.utils import USER, SeleniumTestCase, apply_migration, object_manager, retry from tests.e2e.utils import SeleniumTestCase, apply_migration, object_manager, retry
@skipUnless(platform.startswith("linux"), "requires local docker") @skipUnless(platform.startswith("linux"), "requires local docker")
@ -148,9 +148,9 @@ class TestProviderOAuth2OIDCImplicit(SeleniumTestCase):
sleep(1) sleep(1)
body = loads(self.driver.find_element(By.CSS_SELECTOR, "pre").text) body = loads(self.driver.find_element(By.CSS_SELECTOR, "pre").text)
print(body) print(body)
self.assertEqual(body["profile"]["nickname"], USER().username) self.assertEqual(body["profile"]["nickname"], self.user.username)
self.assertEqual(body["profile"]["name"], USER().name) self.assertEqual(body["profile"]["name"], self.user.name)
self.assertEqual(body["profile"]["email"], USER().email) self.assertEqual(body["profile"]["email"], self.user.email)
@retry() @retry()
@apply_migration("authentik_core", "0002_auto_20200523_1133_squashed_0011_provider_name_temp") @apply_migration("authentik_core", "0002_auto_20200523_1133_squashed_0011_provider_name_temp")
@ -210,9 +210,9 @@ class TestProviderOAuth2OIDCImplicit(SeleniumTestCase):
sleep(1) sleep(1)
body = loads(self.driver.find_element(By.CSS_SELECTOR, "pre").text) body = loads(self.driver.find_element(By.CSS_SELECTOR, "pre").text)
self.assertEqual(body["profile"]["nickname"], USER().username) self.assertEqual(body["profile"]["nickname"], self.user.username)
self.assertEqual(body["profile"]["name"], USER().name) self.assertEqual(body["profile"]["name"], self.user.name)
self.assertEqual(body["profile"]["email"], USER().email) self.assertEqual(body["profile"]["email"], self.user.email)
@retry() @retry()
@apply_migration("authentik_core", "0002_auto_20200523_1133_squashed_0011_provider_name_temp") @apply_migration("authentik_core", "0002_auto_20200523_1133_squashed_0011_provider_name_temp")

View file

@ -16,7 +16,7 @@ from authentik.flows.models import Flow
from authentik.outposts.models import DockerServiceConnection, Outpost, OutpostConfig, OutpostType from authentik.outposts.models import DockerServiceConnection, Outpost, OutpostConfig, OutpostType
from authentik.outposts.tasks import outpost_local_connection from authentik.outposts.tasks import outpost_local_connection
from authentik.providers.proxy.models import ProxyProvider from authentik.providers.proxy.models import ProxyProvider
from tests.e2e.utils import USER, SeleniumTestCase, apply_migration, object_manager, retry from tests.e2e.utils import SeleniumTestCase, apply_migration, object_manager, retry
@skipUnless(platform.startswith("linux"), "requires local docker") @skipUnless(platform.startswith("linux"), "requires local docker")
@ -62,9 +62,8 @@ class TestProviderProxy(SeleniumTestCase):
def test_proxy_simple(self): def test_proxy_simple(self):
"""Test simple outpost setup with single provider""" """Test simple outpost setup with single provider"""
# set additionalHeaders to test later # set additionalHeaders to test later
user = USER() self.user.attributes["additionalHeaders"] = {"X-Foo": "bar"}
user.attributes["additionalHeaders"] = {"X-Foo": "bar"} self.user.save()
user.save()
proxy: ProxyProvider = ProxyProvider.objects.create( proxy: ProxyProvider = ProxyProvider.objects.create(
name="proxy_provider", name="proxy_provider",

View file

@ -16,7 +16,7 @@ from authentik.flows.models import Flow
from authentik.policies.expression.models import ExpressionPolicy from authentik.policies.expression.models import ExpressionPolicy
from authentik.policies.models import PolicyBinding from authentik.policies.models import PolicyBinding
from authentik.providers.saml.models import SAMLBindings, SAMLPropertyMapping, SAMLProvider from authentik.providers.saml.models import SAMLBindings, SAMLPropertyMapping, SAMLProvider
from tests.e2e.utils import USER, SeleniumTestCase, apply_migration, object_manager, retry from tests.e2e.utils import SeleniumTestCase, apply_migration, object_manager, retry
@skipUnless(platform.startswith("linux"), "requires local docker") @skipUnless(platform.startswith("linux"), "requires local docker")
@ -96,29 +96,29 @@ class TestProviderSAML(SeleniumTestCase):
self.assertEqual( self.assertEqual(
body["attr"]["http://schemas.xmlsoap.org/ws/2005/05/identity/claims/name"], body["attr"]["http://schemas.xmlsoap.org/ws/2005/05/identity/claims/name"],
[USER().name], [self.user.name],
) )
self.assertEqual( self.assertEqual(
body["attr"][ body["attr"][
"http://schemas.microsoft.com/ws/2008/06/identity/claims/windowsaccountname" "http://schemas.microsoft.com/ws/2008/06/identity/claims/windowsaccountname"
], ],
[USER().username], [self.user.username],
) )
self.assertEqual( self.assertEqual(
body["attr"]["http://schemas.goauthentik.io/2021/02/saml/username"], body["attr"]["http://schemas.goauthentik.io/2021/02/saml/username"],
[USER().username], [self.user.username],
) )
self.assertEqual( self.assertEqual(
body["attr"]["http://schemas.goauthentik.io/2021/02/saml/uid"], body["attr"]["http://schemas.goauthentik.io/2021/02/saml/uid"],
[str(USER().pk)], [str(self.user.pk)],
) )
self.assertEqual( self.assertEqual(
body["attr"]["http://schemas.xmlsoap.org/ws/2005/05/identity/claims/emailaddress"], body["attr"]["http://schemas.xmlsoap.org/ws/2005/05/identity/claims/emailaddress"],
[USER().email], [self.user.email],
) )
self.assertEqual( self.assertEqual(
body["attr"]["http://schemas.xmlsoap.org/ws/2005/05/identity/claims/upn"], body["attr"]["http://schemas.xmlsoap.org/ws/2005/05/identity/claims/upn"],
[USER().email], [self.user.email],
) )
@retry() @retry()
@ -174,29 +174,29 @@ class TestProviderSAML(SeleniumTestCase):
self.assertEqual( self.assertEqual(
body["attr"]["http://schemas.xmlsoap.org/ws/2005/05/identity/claims/name"], body["attr"]["http://schemas.xmlsoap.org/ws/2005/05/identity/claims/name"],
[USER().name], [self.user.name],
) )
self.assertEqual( self.assertEqual(
body["attr"][ body["attr"][
"http://schemas.microsoft.com/ws/2008/06/identity/claims/windowsaccountname" "http://schemas.microsoft.com/ws/2008/06/identity/claims/windowsaccountname"
], ],
[USER().username], [self.user.username],
) )
self.assertEqual( self.assertEqual(
body["attr"]["http://schemas.goauthentik.io/2021/02/saml/username"], body["attr"]["http://schemas.goauthentik.io/2021/02/saml/username"],
[USER().username], [self.user.username],
) )
self.assertEqual( self.assertEqual(
body["attr"]["http://schemas.goauthentik.io/2021/02/saml/uid"], body["attr"]["http://schemas.goauthentik.io/2021/02/saml/uid"],
[str(USER().pk)], [str(self.user.pk)],
) )
self.assertEqual( self.assertEqual(
body["attr"]["http://schemas.xmlsoap.org/ws/2005/05/identity/claims/emailaddress"], body["attr"]["http://schemas.xmlsoap.org/ws/2005/05/identity/claims/emailaddress"],
[USER().email], [self.user.email],
) )
self.assertEqual( self.assertEqual(
body["attr"]["http://schemas.xmlsoap.org/ws/2005/05/identity/claims/upn"], body["attr"]["http://schemas.xmlsoap.org/ws/2005/05/identity/claims/upn"],
[USER().email], [self.user.email],
) )
@retry() @retry()
@ -243,29 +243,29 @@ class TestProviderSAML(SeleniumTestCase):
self.assertEqual( self.assertEqual(
body["attr"]["http://schemas.xmlsoap.org/ws/2005/05/identity/claims/name"], body["attr"]["http://schemas.xmlsoap.org/ws/2005/05/identity/claims/name"],
[USER().name], [self.user.name],
) )
self.assertEqual( self.assertEqual(
body["attr"][ body["attr"][
"http://schemas.microsoft.com/ws/2008/06/identity/claims/windowsaccountname" "http://schemas.microsoft.com/ws/2008/06/identity/claims/windowsaccountname"
], ],
[USER().username], [self.user.username],
) )
self.assertEqual( self.assertEqual(
body["attr"]["http://schemas.goauthentik.io/2021/02/saml/username"], body["attr"]["http://schemas.goauthentik.io/2021/02/saml/username"],
[USER().username], [self.user.username],
) )
self.assertEqual( self.assertEqual(
body["attr"]["http://schemas.goauthentik.io/2021/02/saml/uid"], body["attr"]["http://schemas.goauthentik.io/2021/02/saml/uid"],
[str(USER().pk)], [str(self.user.pk)],
) )
self.assertEqual( self.assertEqual(
body["attr"]["http://schemas.xmlsoap.org/ws/2005/05/identity/claims/emailaddress"], body["attr"]["http://schemas.xmlsoap.org/ws/2005/05/identity/claims/emailaddress"],
[USER().email], [self.user.email],
) )
self.assertEqual( self.assertEqual(
body["attr"]["http://schemas.xmlsoap.org/ws/2005/05/identity/claims/upn"], body["attr"]["http://schemas.xmlsoap.org/ws/2005/05/identity/claims/upn"],
[USER().email], [self.user.email],
) )
@retry() @retry()

View file

@ -27,15 +27,11 @@ from structlog.stdlib import get_logger
from authentik.core.api.users import UserSerializer from authentik.core.api.users import UserSerializer
from authentik.core.models import User from authentik.core.models import User
from authentik.core.tests.utils import create_test_admin_user
from authentik.managed.manager import ObjectManager from authentik.managed.manager import ObjectManager
RETRIES = int(environ.get("RETRIES", "3")) RETRIES = int(environ.get("RETRIES", "3"))
# pylint: disable=invalid-name
def USER() -> User: # noqa
"""Cached function that always returns akadmin"""
return User.objects.get(username="akadmin")
def get_docker_tag() -> str: def get_docker_tag() -> str:
"""Get docker-tag based off of CI variables""" """Get docker-tag based off of CI variables"""
@ -53,6 +49,7 @@ class SeleniumTestCase(ChannelsLiveServerTestCase):
container: Optional[Container] = None container: Optional[Container] = None
wait_timeout: int wait_timeout: int
user: User
def setUp(self): def setUp(self):
super().setUp() super().setUp()
@ -63,6 +60,7 @@ class SeleniumTestCase(ChannelsLiveServerTestCase):
self.driver.implicitly_wait(30) self.driver.implicitly_wait(30)
self.wait = WebDriverWait(self.driver, self.wait_timeout) self.wait = WebDriverWait(self.driver, self.wait_timeout)
self.logger = get_logger() self.logger = get_logger()
self.user = create_test_admin_user(set_password=True)
if specs := self.get_container_specs(): if specs := self.get_container_specs():
self.container = self._start_container(specs) self.container = self._start_container(specs)
@ -162,7 +160,7 @@ class SeleniumTestCase(ChannelsLiveServerTestCase):
identification_stage.find_element(By.CSS_SELECTOR, "input[name=uidField]").click() identification_stage.find_element(By.CSS_SELECTOR, "input[name=uidField]").click()
identification_stage.find_element(By.CSS_SELECTOR, "input[name=uidField]").send_keys( identification_stage.find_element(By.CSS_SELECTOR, "input[name=uidField]").send_keys(
USER().username self.user.username
) )
identification_stage.find_element(By.CSS_SELECTOR, "input[name=uidField]").send_keys( identification_stage.find_element(By.CSS_SELECTOR, "input[name=uidField]").send_keys(
Keys.ENTER Keys.ENTER
@ -171,7 +169,7 @@ class SeleniumTestCase(ChannelsLiveServerTestCase):
flow_executor = self.get_shadow_root("ak-flow-executor") flow_executor = self.get_shadow_root("ak-flow-executor")
password_stage = self.get_shadow_root("ak-stage-password", flow_executor) password_stage = self.get_shadow_root("ak-stage-password", flow_executor)
password_stage.find_element(By.CSS_SELECTOR, "input[name=password]").send_keys( password_stage.find_element(By.CSS_SELECTOR, "input[name=password]").send_keys(
USER().username self.user.username
) )
password_stage.find_element(By.CSS_SELECTOR, "input[name=password]").send_keys(Keys.ENTER) password_stage.find_element(By.CSS_SELECTOR, "input[name=password]").send_keys(Keys.ENTER)
sleep(1) sleep(1)