<filename>tests/e2e/test_source_saml.py
"""test SAML Source"""
from sys import platform
from time import sleep
from typing import Any, Optional
from unittest.case import skipUnless

from docker.types import Healthcheck
from guardian.utils import get_anonymous_user
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support import expected_conditions as ec
from selenium.webdriver.support.wait import WebDriverWait

from authentik.core.models import User
from authentik.crypto.models import CertificateKeyPair
from authentik.flows.models import Flow
from authentik.sources.saml.models import SAMLBindingTypes, SAMLSource
from authentik.stages.identification.models import IdentificationStage
from tests.e2e.utils import SeleniumTestCase, apply_migration, object_manager, retry

IDP_CERT = """-----BEGIN CERTIFICATE-----
<KEY>
-----END CERTIFICATE-----"""

IDP_KEY = """-----BEGIN PRIVATE KEY-----
<KEY>"""


@skipUnless(platform.startswith("linux"), "requires local docker")
class TestSourceSAML(SeleniumTestCase):
    """test SAML Source flow"""

    def get_container_specs(self) -> Optional[dict[str, Any]]:
        return {
            "image": "kristophjunge/test-saml-idp:1.15",
            "detach": True,
            "network_mode": "host",
            "auto_remove": True,
            "healthcheck": Healthcheck(
                test=["CMD", "curl", "http://localhost:8080"],
                interval=5 * 100 * 1000000,
                start_period=1 * 100 * 1000000,
            ),
            "environment": {
                "SIMPLESAMLPHP_SP_ENTITY_ID": "entity-id",
                "SIMPLESAMLPHP_SP_ASSERTION_CONSUMER_SERVICE": (
                    f"{self.live_server_url}/source/saml/saml-idp-test/acs/"
                ),
            },
        }

    @retry()
    @apply_migration("authentik_flows", "0008_default_flows")
    @apply_migration("authentik_flows", "0011_flow_title")
    @apply_migration("authentik_flows", "0009_source_flows")
    @apply_migration("authentik_crypto", "0002_create_self_signed_kp")
    @apply_migration("authentik_sources_saml", "0010_samlsource_pre_authentication_flow")
    @object_manager
    def test_idp_redirect(self):
        """test SAML Source With redirect binding"""
        # Bootstrap all needed objects
        authentication_flow = Flow.objects.get(slug="default-source-authentication")
        enrollment_flow = Flow.objects.get(slug="default-source-enrollment")
        pre_authentication_flow = Flow.objects.get(slug="default-source-pre-authentication")
        keypair = CertificateKeyPair.objects.create(
            name="test-idp-cert",
            certificate_data=IDP_CERT,
            key_data=IDP_KEY,
        )

        source = SAMLSource.objects.create(
            name="saml-idp-test",
            slug="saml-idp-test",
            authentication_flow=authentication_flow,
            enrollment_flow=enrollment_flow,
            pre_authentication_flow=pre_authentication_flow,
            issuer="entity-id",
            sso_url="http://localhost:8080/simplesaml/saml2/idp/SSOService.php",
            binding_type=SAMLBindingTypes.REDIRECT,
            signing_kp=keypair,
        )
        ident_stage = IdentificationStage.objects.first()
        ident_stage.sources.set([source])
        ident_stage.save()

        self.driver.get(self.live_server_url)

        flow_executor = self.get_shadow_root("ak-flow-executor")
        identification_stage = self.get_shadow_root("ak-stage-identification", flow_executor)
        wait = WebDriverWait(identification_stage, self.wait_timeout)

        wait.until(
            ec.presence_of_element_located(
                (By.CSS_SELECTOR, ".pf-c-login__main-footer-links-item > button")
            )
        )
        identification_stage.find_element(
            By.CSS_SELECTOR, ".pf-c-login__main-footer-links-item > button"
        ).click()

        # Now we should be at the IDP, wait for the username field
        self.wait.until(ec.presence_of_element_located((By.ID, "username")))
        self.driver.find_element(By.ID, "username").send_keys("user1")
        self.driver.find_element(By.ID, "password").send_keys("<PASSWORD>")
        self.driver.find_element(By.ID, "password").send_keys(Keys.ENTER)

        # Wait until we're logged in
        self.wait_for_url(self.if_user_url("/library"))
        self.driver.get(self.if_user_url("/settings"))

        self.assert_user(
            User.objects.exclude(username="akadmin")
            .exclude(username__startswith="ak-outpost")
            .exclude(pk=get_anonymous_user().pk)
            .exclude(pk=self.user.pk)
            .first()
        )

    @retry()
    @apply_migration("authentik_flows", "0008_default_flows")
    @apply_migration("authentik_flows", "0011_flow_title")
    @apply_migration("authentik_flows", "0009_source_flows")
    @apply_migration("authentik_crypto", "0002_create_self_signed_kp")
    @apply_migration("authentik_sources_saml", "0010_samlsource_pre_authentication_flow")
    @object_manager
    def test_idp_post(self):
        """test SAML Source With post binding"""
        # Bootstrap all needed objects
        authentication_flow = Flow.objects.get(slug="default-source-authentication")
        enrollment_flow = Flow.objects.get(slug="default-source-enrollment")
        pre_authentication_flow = Flow.objects.get(slug="default-source-pre-authentication")
        keypair = CertificateKeyPair.objects.create(
            name="test-idp-cert",
            certificate_data=IDP_CERT,
            key_data=IDP_KEY,
        )

        source = SAMLSource.objects.create(
            name="saml-idp-test",
            slug="saml-idp-test",
            authentication_flow=authentication_flow,
            enrollment_flow=enrollment_flow,
            pre_authentication_flow=pre_authentication_flow,
            issuer="entity-id",
            sso_url="http://localhost:8080/simplesaml/saml2/idp/SSOService.php",
            binding_type=SAMLBindingTypes.POST,
            signing_kp=keypair,
        )
        ident_stage = IdentificationStage.objects.first()
        ident_stage.sources.set([source])
        ident_stage.save()

        self.driver.get(self.live_server_url)

        flow_executor = self.get_shadow_root("ak-flow-executor")
        identification_stage = self.get_shadow_root("ak-stage-identification", flow_executor)
        wait = WebDriverWait(identification_stage, self.wait_timeout)

        wait.until(
            ec.presence_of_element_located(
                (By.CSS_SELECTOR, ".pf-c-login__main-footer-links-item > button")
            )
        )
        identification_stage.find_element(
            By.CSS_SELECTOR, ".pf-c-login__main-footer-links-item > button"
        ).click()
        sleep(1)

        flow_executor = self.get_shadow_root("ak-flow-executor")
        consent_stage = self.get_shadow_root("ak-stage-consent", flow_executor)

        self.assertIn(
            source.name,
            consent_stage.find_element(By.CSS_SELECTOR, "#header-text").text,
        )
        consent_stage.find_element(
            By.CSS_SELECTOR,
            ("[type=submit]"),
        ).click()

        # Now we should be at the IDP, wait for the username field
        self.wait.until(ec.presence_of_element_located((By.ID, "username")))
        self.driver.find_element(By.ID, "username").send_keys("user1")
        self.driver.find_element(By.ID, "password").send_keys("<PASSWORD>")
        self.driver.find_element(By.ID, "password").send_keys(Keys.ENTER)

        # Wait until we're logged in
        self.wait_for_url(self.if_user_url("/library"))
        self.driver.get(self.if_user_url("/settings"))

        self.assert_user(
            User.objects.exclude(username="akadmin")
            .exclude(username__startswith="ak-outpost")
            .exclude(pk=get_anonymous_user().pk)
            .exclude(pk=self.user.pk)
            .first()
        )

    @retry()
    @apply_migration("authentik_flows", "0008_default_flows")
    @apply_migration("authentik_flows", "0011_flow_title")
    @apply_migration("authentik_flows", "0009_source_flows")
    @apply_migration("authentik_crypto", "0002_create_self_signed_kp")
    @apply_migration("authentik_sources_saml", "0010_samlsource_pre_authentication_flow")
    @object_manager
    def test_idp_post_auto(self):
        """test SAML Source With post binding (auto redirect)"""
        # Bootstrap all needed objects
        authentication_flow = Flow.objects.get(slug="default-source-authentication")
        enrollment_flow = Flow.objects.get(slug="default-source-enrollment")
        pre_authentication_flow = Flow.objects.get(slug="default-source-pre-authentication")
        keypair = CertificateKeyPair.objects.create(
            name="test-idp-cert",
            certificate_data=IDP_CERT,
            key_data=IDP_KEY,
        )

        source = SAMLSource.objects.create(
            name="saml-idp-test",
            slug="saml-idp-test",
            authentication_flow=authentication_flow,
            enrollment_flow=enrollment_flow,
            pre_authentication_flow=pre_authentication_flow,
            issuer="entity-id",
            sso_url="http://localhost:8080/simplesaml/saml2/idp/SSOService.php",
            binding_type=SAMLBindingTypes.POST_AUTO,
            signing_kp=keypair,
        )
        ident_stage = IdentificationStage.objects.first()
        ident_stage.sources.set([source])
        ident_stage.save()

        self.driver.get(self.live_server_url)

        flow_executor = self.get_shadow_root("ak-flow-executor")
        identification_stage = self.get_shadow_root("ak-stage-identification", flow_executor)
        wait = WebDriverWait(identification_stage, self.wait_timeout)

        wait.until(
            ec.presence_of_element_located(
                (By.CSS_SELECTOR, ".pf-c-login__main-footer-links-item > button")
            )
        )
        identification_stage.find_element(
            By.CSS_SELECTOR, ".pf-c-login__main-footer-links-item > button"
        ).click()

        # Now we should be at the IDP, wait for the username field
        self.wait.until(ec.presence_of_element_located((By.ID, "username")))
        self.driver.find_element(By.ID, "username").send_keys("user1")
        self.driver.find_element(By.ID, "password").send_keys("user1pass")
        self.driver.find_element(By.ID, "password").send_keys(Keys.ENTER)

        # Wait until we're logged in
        self.wait_for_url(self.if_user_url("/library"))
        self.driver.get(self.if_user_url("/settings"))

        self.assert_user(
            User.objects.exclude(username="akadmin")
            .exclude(username__startswith="ak-outpost")
            .exclude(pk=get_anonymous_user().pk)
            .exclude(pk=self.user.pk)
            .first()
        )

<filename>tests/e2e/test_source_saml.py
"""test SAML Source"""
from sys import platform
from time import sleep
from typing import Any, Optional
from unittest.case import skipUnless

from docker.types import Healthcheck
from guardian.utils import get_anonymous_user
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support import expected_conditions as ec
from selenium.webdriver.support.wait import WebDriverWait

from authentik.core.models import User
from authentik.crypto.models import CertificateKeyPair
from authentik.flows.models import Flow
from authentik.sources.saml.models import SAMLBindingTypes, SAMLSource
from authentik.stages.identification.models import IdentificationStage
from tests.e2e.utils import SeleniumTestCase, apply_migration, object_manager, retry

IDP_CERT = """-----BEGIN CERTIFICATE-----
<KEY>
-----END CERTIFICATE-----"""

IDP_KEY = """-----BEGIN PRIVATE KEY-----
<KEY>"""


@skipUnless(platform.startswith("linux"), "requires local docker")
class TestSourceSAML(SeleniumTestCase):
    """test SAML Source flow"""

    def get_container_specs(self) -> Optional[dict[str, Any]]:
        return {
            "image": "kristophjunge/test-saml-idp:1.15",
            "detach": True,
            "network_mode": "host",
            "auto_remove": True,
            "healthcheck": Healthcheck(
                test=["CMD", "curl", "http://localhost:8080"],
                interval=5 * 100 * 1000000,
                start_period=1 * 100 * 1000000,
            ),
            "environment": {
                "SIMPLESAMLPHP_SP_ENTITY_ID": "entity-id",
                "SIMPLESAMLPHP_SP_ASSERTION_CONSUMER_SERVICE": (
                    f"{self.live_server_url}/source/saml/saml-idp-test/acs/"
                ),
            },
        }

    @retry()
    @apply_migration("authentik_flows", "0008_default_flows")
    @apply_migration("authentik_flows", "0011_flow_title")
    @apply_migration("authentik_flows", "0009_source_flows")
    @apply_migration("authentik_crypto", "0002_create_self_signed_kp")
    @apply_migration("authentik_sources_saml", "0010_samlsource_pre_authentication_flow")
    @object_manager
    def test_idp_redirect(self):
        """test SAML Source With redirect binding"""
        # Bootstrap all needed objects
        authentication_flow = Flow.objects.get(slug="default-source-authentication")
        enrollment_flow = Flow.objects.get(slug="default-source-enrollment")
        pre_authentication_flow = Flow.objects.get(slug="default-source-pre-authentication")
        keypair = CertificateKeyPair.objects.create(
            name="test-idp-cert",
            certificate_data=IDP_CERT,
            key_data=IDP_KEY,
        )

        source = SAMLSource.objects.create(
            name="saml-idp-test",
            slug="saml-idp-test",
            authentication_flow=authentication_flow,
            enrollment_flow=enrollment_flow,
            pre_authentication_flow=pre_authentication_flow,
            issuer="entity-id",
            sso_url="http://localhost:8080/simplesaml/saml2/idp/SSOService.php",
            binding_type=SAMLBindingTypes.REDIRECT,
            signing_kp=keypair,
        )
        ident_stage = IdentificationStage.objects.first()
        ident_stage.sources.set([source])
        ident_stage.save()

        self.driver.get(self.live_server_url)

        flow_executor = self.get_shadow_root("ak-flow-executor")
        identification_stage = self.get_shadow_root("ak-stage-identification", flow_executor)
        wait = WebDriverWait(identification_stage, self.wait_timeout)

        wait.until(
            ec.presence_of_element_located(
                (By.CSS_SELECTOR, ".pf-c-login__main-footer-links-item > button")
            )
        )
        identification_stage.find_element(
            By.CSS_SELECTOR, ".pf-c-login__main-footer-links-item > button"
        ).click()

        # Now we should be at the IDP, wait for the username field
        self.wait.until(ec.presence_of_element_located((By.ID, "username")))
        self.driver.find_element(By.ID, "username").send_keys("user1")
        self.driver.find_element(By.ID, "password").send_keys("<PASSWORD>")
        self.driver.find_element(By.ID, "password").send_keys(Keys.ENTER)

        # Wait until we're logged in
        self.wait_for_url(self.if_user_url("/library"))
        self.driver.get(self.if_user_url("/settings"))

        self.assert_user(
            User.objects.exclude(username="akadmin")
            .exclude(username__startswith="ak-outpost")
            .exclude(pk=get_anonymous_user().pk)
            .exclude(pk=self.user.pk)
            .first()
        )

    @retry()
    @apply_migration("authentik_flows", "0008_default_flows")
    @apply_migration("authentik_flows", "0011_flow_title")
    @apply_migration("authentik_flows", "0009_source_flows")
    @apply_migration("authentik_crypto", "0002_create_self_signed_kp")
    @apply_migration("authentik_sources_saml", "0010_samlsource_pre_authentication_flow")
    @object_manager
    def test_idp_post(self):
        """test SAML Source With post binding"""
        # Bootstrap all needed objects
        authentication_flow = Flow.objects.get(slug="default-source-authentication")
        enrollment_flow = Flow.objects.get(slug="default-source-enrollment")
        pre_authentication_flow = Flow.objects.get(slug="default-source-pre-authentication")
        keypair = CertificateKeyPair.objects.create(
            name="test-idp-cert",
            certificate_data=IDP_CERT,
            key_data=IDP_KEY,
        )

        source = SAMLSource.objects.create(
            name="saml-idp-test",
            slug="saml-idp-test",
            authentication_flow=authentication_flow,
            enrollment_flow=enrollment_flow,
            pre_authentication_flow=pre_authentication_flow,
            issuer="entity-id",
            sso_url="http://localhost:8080/simplesaml/saml2/idp/SSOService.php",
            binding_type=SAMLBindingTypes.POST,
            signing_kp=keypair,
        )
        ident_stage = IdentificationStage.objects.first()
        ident_stage.sources.set([source])
        ident_stage.save()

        self.driver.get(self.live_server_url)

        flow_executor = self.get_shadow_root("ak-flow-executor")
        identification_stage = self.get_shadow_root("ak-stage-identification", flow_executor)
        wait = WebDriverWait(identification_stage, self.wait_timeout)

        wait.until(
            ec.presence_of_element_located(
                (By.CSS_SELECTOR, ".pf-c-login__main-footer-links-item > button")
            )
        )
        identification_stage.find_element(
            By.CSS_SELECTOR, ".pf-c-login__main-footer-links-item > button"
        ).click()
        sleep(1)

        flow_executor = self.get_shadow_root("ak-flow-executor")
        consent_stage = self.get_shadow_root("ak-stage-consent", flow_executor)

        self.assertIn(
            source.name,
            consent_stage.find_element(By.CSS_SELECTOR, "#header-text").text,
        )
        consent_stage.find_element(
            By.CSS_SELECTOR,
            ("[type=submit]"),
        ).click()

        # Now we should be at the IDP, wait for the username field
        self.wait.until(ec.presence_of_element_located((By.ID, "username")))
        self.driver.find_element(By.ID, "username").send_keys("user1")
        self.driver.find_element(By.ID, "password").send_keys("<PASSWORD>")
        self.driver.find_element(By.ID, "password").send_keys(Keys.ENTER)

        # Wait until we're logged in
        self.wait_for_url(self.if_user_url("/library"))
        self.driver.get(self.if_user_url("/settings"))

        self.assert_user(
            User.objects.exclude(username="akadmin")
            .exclude(username__startswith="ak-outpost")
            .exclude(pk=get_anonymous_user().pk)
            .exclude(pk=self.user.pk)
            .first()
        )

    @retry()
    @apply_migration("authentik_flows", "0008_default_flows")
    @apply_migration("authentik_flows", "0011_flow_title")
    @apply_migration("authentik_flows", "0009_source_flows")
    @apply_migration("authentik_crypto", "0002_create_self_signed_kp")
    @apply_migration("authentik_sources_saml", "0010_samlsource_pre_authentication_flow")
    @object_manager
    def test_idp_post_auto(self):
        """test SAML Source With post binding (auto redirect)"""
        # Bootstrap all needed objects
        authentication_flow = Flow.objects.get(slug="default-source-authentication")
        enrollment_flow = Flow.objects.get(slug="default-source-enrollment")
        pre_authentication_flow = Flow.objects.get(slug="default-source-pre-authentication")
        keypair = CertificateKeyPair.objects.create(
            name="test-idp-cert",
            certificate_data=IDP_CERT,
            key_data=IDP_KEY,
        )

        source = SAMLSource.objects.create(
            name="saml-idp-test",
            slug="saml-idp-test",
            authentication_flow=authentication_flow,
            enrollment_flow=enrollment_flow,
            pre_authentication_flow=pre_authentication_flow,
            issuer="entity-id",
            sso_url="http://localhost:8080/simplesaml/saml2/idp/SSOService.php",
            binding_type=SAMLBindingTypes.POST_AUTO,
            signing_kp=keypair,
        )
        ident_stage = IdentificationStage.objects.first()
        ident_stage.sources.set([source])
        ident_stage.save()

        self.driver.get(self.live_server_url)

        flow_executor = self.get_shadow_root("ak-flow-executor")
        identification_stage = self.get_shadow_root("ak-stage-identification", flow_executor)
        wait = WebDriverWait(identification_stage, self.wait_timeout)

        wait.until(
            ec.presence_of_element_located(
                (By.CSS_SELECTOR, ".pf-c-login__main-footer-links-item > button")
            )
        )
        identification_stage.find_element(
            By.CSS_SELECTOR, ".pf-c-login__main-footer-links-item > button"
        ).click()

        # Now we should be at the IDP, wait for the username field
        self.wait.until(ec.presence_of_element_located((By.ID, "username")))
        self.driver.find_element(By.ID, "username").send_keys("user1")
        self.driver.find_element(By.ID, "password").send_keys("user1pass")
        self.driver.find_element(By.ID, "password").send_keys(Keys.ENTER)

        # Wait until we're logged in
        self.wait_for_url(self.if_user_url("/library"))
        self.driver.get(self.if_user_url("/settings"))

        self.assert_user(
            User.objects.exclude(username="akadmin")
            .exclude(username__startswith="ak-outpost")
            .exclude(pk=get_anonymous_user().pk)
            .exclude(pk=self.user.pk)
            .first()
        )

test SAML Source -----BEGIN CERTIFICATE----- <KEY> -----END CERTIFICATE----- -----BEGIN PRIVATE KEY----- <KEY> test SAML Source flow test SAML Source With redirect binding # Bootstrap all needed objects # Now we should be at the IDP, wait for the username field # Wait until we're logged in test SAML Source With post binding # Bootstrap all needed objects # Now we should be at the IDP, wait for the username field # Wait until we're logged in test SAML Source With post binding (auto redirect) # Bootstrap all needed objects # Now we should be at the IDP, wait for the username field # Wait until we're logged in