This repository has been archived on 2024-05-31. You can view files and clone it, but cannot push or open issues or pull requests.
authentik/tests/e2e/test_provider_saml.py
Jens L e25d03d8f4
Managed objects (#519)
* managed: add base manager and Ops

* core: use ManagedModel for Token and PropertyMapping

* providers/saml: implement managed objects for SAML Provider

* sources/ldap: migrate to managed

* providers/oauth2: migrate to managed

* providers/proxy: migrate to managed

* *: load .managed in apps

* managed: add reconcile task, run on startup

* providers/oauth2: fix import path for managed

* providers/saml: don't set FriendlyName when mapping is none

* *: use ObjectManager in tests to ensure objects exist

* ci: use vmImage ubuntu-latest

* providers/saml: add new mapping for username and user id

* tests: remove docker proxy

* tests/e2e: use updated attribute names

* docs: update SAML docs

* tests/e2e: fix remaining saml cases

* outposts: make tokens as managed

* *: make PropertyMapping SerializerModel

* web: add page for property-mappings

* web: add codemirror to common_styles because codemirror

* docs: fix member-of in nextcloud

* docs: nextcloud add admin

* web: fix refresh reloading data two times

* web: add loading lock to table to prevent double loads

* web: add ability to use null in QueryArgs (value will be skipped)

* web: add hide option to property mappings

* web: fix linting
2021-02-03 21:18:31 +01:00

311 lines
12 KiB
Python

"""test SAML Provider flow"""
from json import loads
from sys import platform
from time import sleep
from unittest.case import skipUnless
from docker import DockerClient, from_env
from docker.models.containers import Container
from docker.types import Healthcheck
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support import expected_conditions as ec
from structlog.stdlib import get_logger
from authentik.core.models import Application
from authentik.crypto.models import CertificateKeyPair
from authentik.flows.models import Flow
from authentik.policies.expression.models import ExpressionPolicy
from authentik.policies.models import PolicyBinding
from authentik.providers.saml.models import (
SAMLBindings,
SAMLPropertyMapping,
SAMLProvider,
)
from tests.e2e.utils import USER, SeleniumTestCase, retry
LOGGER = get_logger()
@skipUnless(platform.startswith("linux"), "requires local docker")
class TestProviderSAML(SeleniumTestCase):
"""test SAML Provider flow"""
container: Container
def setup_client(self, provider: SAMLProvider) -> Container:
"""Setup client saml-sp container which we test SAML against"""
client: DockerClient = from_env()
container = client.containers.run(
image="beryju/saml-test-sp",
detach=True,
network_mode="host",
auto_remove=True,
healthcheck=Healthcheck(
test=["CMD", "wget", "--spider", "http://localhost:9009/health"],
interval=5 * 100 * 1000000,
start_period=1 * 100 * 1000000,
),
environment={
"SP_ENTITY_ID": provider.issuer,
"SP_SSO_BINDING": "urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST",
"SP_METADATA_URL": (
self.url(
"authentik_providers_saml:metadata",
application_slug=provider.application.slug,
)
),
},
)
while True:
container.reload()
status = container.attrs.get("State", {}).get("Health", {}).get("Status")
if status == "healthy":
return container
LOGGER.info("Container failed healthcheck")
sleep(1)
@retry()
def test_sp_initiated_implicit(self):
"""test SAML Provider flow SP-initiated flow (implicit consent)"""
# Bootstrap all needed objects
authorization_flow = Flow.objects.get(
slug="default-provider-authorization-implicit-consent"
)
provider: SAMLProvider = SAMLProvider.objects.create(
name="saml-test",
acs_url="http://localhost:9009/saml/acs",
audience="authentik-e2e",
issuer="authentik-e2e",
sp_binding=SAMLBindings.POST,
authorization_flow=authorization_flow,
signing_kp=CertificateKeyPair.objects.first(),
)
provider.property_mappings.set(SAMLPropertyMapping.objects.all())
provider.save()
Application.objects.create(
name="SAML",
slug="authentik-saml",
provider=provider,
)
self.container = self.setup_client(provider)
self.driver.get("http://localhost:9009")
self.driver.find_element(By.ID, "id_uid_field").click()
self.driver.find_element(By.ID, "id_uid_field").send_keys(USER().username)
self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
self.driver.find_element(By.ID, "id_password").send_keys(USER().username)
self.driver.find_element(By.ID, "id_password").send_keys(Keys.ENTER)
self.wait_for_url("http://localhost:9009/")
body = loads(self.driver.find_element(By.CSS_SELECTOR, "pre").text)
self.assertEqual(
body["attr"]["http://schemas.xmlsoap.org/ws/2005/05/identity/claims/name"],
[USER().name],
)
self.assertEqual(
body["attr"][
"http://schemas.microsoft.com/ws/2008/06/identity/claims/windowsaccountname"
],
[USER().username],
)
self.assertEqual(
body["attr"]["http://schemas.goauthentik.io/2021/02/saml/username"],
[USER().username],
)
self.assertEqual(
body["attr"]["http://schemas.goauthentik.io/2021/02/saml/uid"],
[str(USER().pk)],
)
self.assertEqual(
body["attr"][
"http://schemas.xmlsoap.org/ws/2005/05/identity/claims/emailaddress"
],
[USER().email],
)
self.assertEqual(
body["attr"]["http://schemas.xmlsoap.org/ws/2005/05/identity/claims/upn"],
[USER().email],
)
@retry()
def test_sp_initiated_explicit(self):
"""test SAML Provider flow SP-initiated flow (explicit consent)"""
# Bootstrap all needed objects
authorization_flow = Flow.objects.get(
slug="default-provider-authorization-explicit-consent"
)
provider: SAMLProvider = SAMLProvider.objects.create(
name="saml-test",
acs_url="http://localhost:9009/saml/acs",
audience="authentik-e2e",
issuer="authentik-e2e",
sp_binding=SAMLBindings.POST,
authorization_flow=authorization_flow,
signing_kp=CertificateKeyPair.objects.first(),
)
provider.property_mappings.set(SAMLPropertyMapping.objects.all())
provider.save()
app = Application.objects.create(
name="SAML",
slug="authentik-saml",
provider=provider,
)
self.container = self.setup_client(provider)
self.driver.get("http://localhost:9009")
self.driver.find_element(By.ID, "id_uid_field").click()
self.driver.find_element(By.ID, "id_uid_field").send_keys(USER().username)
self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
self.driver.find_element(By.ID, "id_password").send_keys(USER().username)
self.driver.find_element(By.ID, "id_password").send_keys(Keys.ENTER)
self.assertEqual(
app.name,
self.driver.find_element(By.ID, "application-name").text,
)
sleep(1)
self.driver.find_element(By.CSS_SELECTOR, "[type=submit]").click()
self.wait_for_url("http://localhost:9009/")
body = loads(self.driver.find_element(By.CSS_SELECTOR, "pre").text)
self.assertEqual(
body["attr"]["http://schemas.xmlsoap.org/ws/2005/05/identity/claims/name"],
[USER().name],
)
self.assertEqual(
body["attr"][
"http://schemas.microsoft.com/ws/2008/06/identity/claims/windowsaccountname"
],
[USER().username],
)
self.assertEqual(
body["attr"]["http://schemas.goauthentik.io/2021/02/saml/username"],
[USER().username],
)
self.assertEqual(
body["attr"]["http://schemas.goauthentik.io/2021/02/saml/uid"],
[str(USER().pk)],
)
self.assertEqual(
body["attr"][
"http://schemas.xmlsoap.org/ws/2005/05/identity/claims/emailaddress"
],
[USER().email],
)
self.assertEqual(
body["attr"]["http://schemas.xmlsoap.org/ws/2005/05/identity/claims/upn"],
[USER().email],
)
@retry()
def test_idp_initiated_implicit(self):
"""test SAML Provider flow IdP-initiated flow (implicit consent)"""
# Bootstrap all needed objects
authorization_flow = Flow.objects.get(
slug="default-provider-authorization-implicit-consent"
)
provider: SAMLProvider = SAMLProvider.objects.create(
name="saml-test",
acs_url="http://localhost:9009/saml/acs",
audience="authentik-e2e",
issuer="authentik-e2e",
sp_binding=SAMLBindings.POST,
authorization_flow=authorization_flow,
signing_kp=CertificateKeyPair.objects.first(),
)
provider.property_mappings.set(SAMLPropertyMapping.objects.all())
provider.save()
Application.objects.create(
name="SAML",
slug="authentik-saml",
provider=provider,
)
self.container = self.setup_client(provider)
self.driver.get(
self.url(
"authentik_providers_saml:sso-init",
application_slug=provider.application.slug,
)
)
self.driver.find_element(By.ID, "id_uid_field").click()
self.driver.find_element(By.ID, "id_uid_field").send_keys(USER().username)
self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
self.driver.find_element(By.ID, "id_password").send_keys(USER().username)
self.driver.find_element(By.ID, "id_password").send_keys(Keys.ENTER)
sleep(1)
self.wait_for_url("http://localhost:9009/")
body = loads(self.driver.find_element(By.CSS_SELECTOR, "pre").text)
self.assertEqual(
body["attr"]["http://schemas.xmlsoap.org/ws/2005/05/identity/claims/name"],
[USER().name],
)
self.assertEqual(
body["attr"][
"http://schemas.microsoft.com/ws/2008/06/identity/claims/windowsaccountname"
],
[USER().username],
)
self.assertEqual(
body["attr"]["http://schemas.goauthentik.io/2021/02/saml/username"],
[USER().username],
)
self.assertEqual(
body["attr"]["http://schemas.goauthentik.io/2021/02/saml/uid"],
[str(USER().pk)],
)
self.assertEqual(
body["attr"][
"http://schemas.xmlsoap.org/ws/2005/05/identity/claims/emailaddress"
],
[USER().email],
)
self.assertEqual(
body["attr"]["http://schemas.xmlsoap.org/ws/2005/05/identity/claims/upn"],
[USER().email],
)
@retry()
def test_sp_initiated_denied(self):
"""test SAML Provider flow SP-initiated flow (Policy denies access)"""
# Bootstrap all needed objects
authorization_flow = Flow.objects.get(
slug="default-provider-authorization-implicit-consent"
)
negative_policy = ExpressionPolicy.objects.create(
name="negative-static", expression="return False"
)
provider: SAMLProvider = SAMLProvider.objects.create(
name="saml-test",
acs_url="http://localhost:9009/saml/acs",
audience="authentik-e2e",
issuer="authentik-e2e",
sp_binding=SAMLBindings.POST,
authorization_flow=authorization_flow,
signing_kp=CertificateKeyPair.objects.first(),
)
provider.property_mappings.set(SAMLPropertyMapping.objects.all())
provider.save()
app = Application.objects.create(
name="SAML",
slug="authentik-saml",
provider=provider,
)
PolicyBinding.objects.create(target=app, policy=negative_policy, order=0)
self.container = self.setup_client(provider)
self.driver.get("http://localhost:9009/")
self.driver.find_element(By.ID, "id_uid_field").click()
self.driver.find_element(By.ID, "id_uid_field").send_keys(USER().username)
self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
self.driver.find_element(By.ID, "id_password").send_keys(USER().username)
self.driver.find_element(By.ID, "id_password").send_keys(Keys.ENTER)
self.wait.until(
ec.presence_of_element_located((By.CSS_SELECTOR, "header > h1"))
)
self.assertEqual(
self.driver.find_element(By.CSS_SELECTOR, "header > h1").text,
"Permission denied",
)