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_oauth2_oidc.py

270 lines
10 KiB
Python
Raw Normal View History

2020-08-19 08:32:44 +00:00
"""test OAuth2 OpenID Provider flow"""
from json import loads
2020-09-11 21:21:11 +00:00
from sys import platform
from time import sleep
2020-09-11 21:21:11 +00:00
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
2020-06-21 16:36:43 +00:00
from selenium.webdriver.support import expected_conditions as ec
from structlog.stdlib import get_logger
2020-12-05 21:08:42 +00:00
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.oauth2.constants import (
2020-08-19 08:32:44 +00:00
SCOPE_OPENID,
SCOPE_OPENID_EMAIL,
SCOPE_OPENID_PROFILE,
)
2020-12-05 21:08:42 +00:00
from authentik.providers.oauth2.generators import (
2020-08-19 08:32:44 +00:00
generate_client_id,
generate_client_secret,
)
from authentik.providers.oauth2.models import ClientTypes, OAuth2Provider, ScopeMapping
from tests.e2e.utils import USER, SeleniumTestCase, retry
LOGGER = get_logger()
2020-09-11 21:21:11 +00:00
@skipUnless(platform.startswith("linux"), "requires local docker")
2020-08-19 08:32:44 +00:00
class TestProviderOAuth2OIDC(SeleniumTestCase):
"""test OAuth with OpenID Provider flow"""
def setUp(self):
self.client_id = generate_client_id()
self.client_secret = generate_client_secret()
self.application_slug = "test"
super().setUp()
def setup_client(self) -> Container:
"""Setup client saml-sp container which we test SAML against"""
sleep(1)
client: DockerClient = from_env()
container = client.containers.run(
image="beryju/oidc-test-client",
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={
"OIDC_CLIENT_ID": self.client_id,
"OIDC_CLIENT_SECRET": self.client_secret,
"OIDC_PROVIDER": f"{self.live_server_url}/application/o/{self.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_redirect_uri_error(self):
"""test OpenID Provider flow (invalid redirect URI, check error message)"""
sleep(1)
# Bootstrap all needed objects
authorization_flow = Flow.objects.get(
slug="default-provider-authorization-implicit-consent"
)
2020-08-19 08:32:44 +00:00
provider = OAuth2Provider.objects.create(
name=self.application_slug,
2020-08-19 08:32:44 +00:00
client_type=ClientTypes.CONFIDENTIAL,
client_id=self.client_id,
client_secret=self.client_secret,
2020-08-19 08:32:44 +00:00
rsa_key=CertificateKeyPair.objects.first(),
redirect_uris="http://localhost:9009/",
2020-08-19 08:32:44 +00:00
authorization_flow=authorization_flow,
)
2020-08-19 08:32:44 +00:00
provider.property_mappings.set(
ScopeMapping.objects.filter(
scope_name__in=[SCOPE_OPENID, SCOPE_OPENID_EMAIL, SCOPE_OPENID_PROFILE]
)
)
2020-08-19 08:32:44 +00:00
provider.save()
Application.objects.create(
2020-09-30 17:34:22 +00:00
name=self.application_slug,
slug=self.application_slug,
provider=provider,
)
self.container = self.setup_client()
self.driver.get("http://localhost:9009")
sleep(2)
self.assertEqual(
self.driver.find_element(By.CLASS_NAME, "pf-c-title").text,
"Redirect URI Error",
)
@retry()
def test_authorization_consent_implied(self):
"""test OpenID Provider flow (default authorization flow with implied consent)"""
sleep(1)
# Bootstrap all needed objects
authorization_flow = Flow.objects.get(
slug="default-provider-authorization-implicit-consent"
)
2020-08-19 08:32:44 +00:00
provider = OAuth2Provider.objects.create(
name=self.application_slug,
2020-08-19 08:32:44 +00:00
client_type=ClientTypes.CONFIDENTIAL,
client_id=self.client_id,
client_secret=self.client_secret,
2020-08-19 08:32:44 +00:00
rsa_key=CertificateKeyPair.objects.first(),
redirect_uris="http://localhost:9009/auth/callback",
2020-08-19 08:32:44 +00:00
authorization_flow=authorization_flow,
)
2020-08-19 08:32:44 +00:00
provider.property_mappings.set(
ScopeMapping.objects.filter(
scope_name__in=[SCOPE_OPENID, SCOPE_OPENID_EMAIL, SCOPE_OPENID_PROFILE]
)
)
2020-08-19 08:32:44 +00:00
provider.save()
Application.objects.create(
2020-09-30 17:34:22 +00:00
name=self.application_slug,
slug=self.application_slug,
provider=provider,
)
self.container = self.setup_client()
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, "pre")))
body = loads(self.driver.find_element(By.CSS_SELECTOR, "pre").text)
self.assertEqual(body["IDTokenClaims"]["nickname"], USER().username)
self.assertEqual(body["UserInfo"]["nickname"], USER().username)
self.assertEqual(body["IDTokenClaims"]["name"], USER().name)
self.assertEqual(body["UserInfo"]["name"], USER().name)
self.assertEqual(body["IDTokenClaims"]["email"], USER().email)
self.assertEqual(body["UserInfo"]["email"], USER().email)
@retry()
def test_authorization_consent_explicit(self):
"""test OpenID Provider flow (default authorization flow with explicit consent)"""
sleep(1)
# Bootstrap all needed objects
authorization_flow = Flow.objects.get(
slug="default-provider-authorization-explicit-consent"
)
2020-08-19 08:32:44 +00:00
provider = OAuth2Provider.objects.create(
name=self.application_slug,
2020-08-19 08:32:44 +00:00
authorization_flow=authorization_flow,
client_type=ClientTypes.CONFIDENTIAL,
client_id=self.client_id,
client_secret=self.client_secret,
2020-08-19 08:32:44 +00:00
rsa_key=CertificateKeyPair.objects.first(),
redirect_uris="http://localhost:9009/auth/callback",
)
2020-08-19 08:32:44 +00:00
provider.property_mappings.set(
ScopeMapping.objects.filter(
scope_name__in=[SCOPE_OPENID, SCOPE_OPENID_EMAIL, SCOPE_OPENID_PROFILE]
)
)
2020-08-19 08:32:44 +00:00
provider.save()
app = Application.objects.create(
2020-09-30 17:34:22 +00:00
name=self.application_slug,
slug=self.application_slug,
provider=provider,
)
self.container = self.setup_client()
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(
2020-09-30 17:34:22 +00:00
app.name,
self.driver.find_element(By.ID, "application-name").text,
)
2020-06-21 17:03:13 +00:00
self.wait.until(
ec.presence_of_element_located((By.CSS_SELECTOR, "[type=submit]"))
)
sleep(1)
self.driver.find_element(By.CSS_SELECTOR, "[type=submit]").click()
self.wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, "pre")))
body = loads(self.driver.find_element(By.CSS_SELECTOR, "pre").text)
self.assertEqual(body["IDTokenClaims"]["nickname"], USER().username)
self.assertEqual(body["UserInfo"]["nickname"], USER().username)
self.assertEqual(body["IDTokenClaims"]["name"], USER().name)
self.assertEqual(body["UserInfo"]["name"], USER().name)
self.assertEqual(body["IDTokenClaims"]["email"], USER().email)
self.assertEqual(body["UserInfo"]["email"], USER().email)
@retry()
def test_authorization_denied(self):
"""test OpenID Provider flow (default authorization with access deny)"""
sleep(1)
# Bootstrap all needed objects
authorization_flow = Flow.objects.get(
slug="default-provider-authorization-explicit-consent"
)
2020-08-19 08:32:44 +00:00
provider = OAuth2Provider.objects.create(
name=self.application_slug,
2020-08-19 08:32:44 +00:00
authorization_flow=authorization_flow,
client_type=ClientTypes.CONFIDENTIAL,
client_id=self.client_id,
client_secret=self.client_secret,
2020-08-19 08:32:44 +00:00
rsa_key=CertificateKeyPair.objects.first(),
redirect_uris="http://localhost:9009/auth/callback",
)
2020-08-19 08:32:44 +00:00
provider.property_mappings.set(
ScopeMapping.objects.filter(
scope_name__in=[SCOPE_OPENID, SCOPE_OPENID_EMAIL, SCOPE_OPENID_PROFILE]
)
)
2020-08-19 08:32:44 +00:00
provider.save()
app = Application.objects.create(
2020-09-30 17:34:22 +00:00
name=self.application_slug,
slug=self.application_slug,
provider=provider,
)
negative_policy = ExpressionPolicy.objects.create(
name="negative-static", expression="return False"
)
PolicyBinding.objects.create(target=app, policy=negative_policy, order=0)
self.container = self.setup_client()
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",
)