e2e: add oidc tests using oidc-test-client
This commit is contained in:
parent
863111ac57
commit
67b69cb5d3
|
@ -0,0 +1,368 @@
|
||||||
|
"""test OAuth2 OpenID Provider flow"""
|
||||||
|
from sys import platform
|
||||||
|
from time import sleep
|
||||||
|
from typing import Any, Dict, Optional
|
||||||
|
from unittest.case import skipUnless
|
||||||
|
|
||||||
|
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 import get_logger
|
||||||
|
|
||||||
|
from e2e.utils import USER, SeleniumTestCase
|
||||||
|
from passbook.core.models import Application
|
||||||
|
from passbook.crypto.models import CertificateKeyPair
|
||||||
|
from passbook.flows.models import Flow
|
||||||
|
from passbook.policies.expression.models import ExpressionPolicy
|
||||||
|
from passbook.policies.models import PolicyBinding
|
||||||
|
from passbook.providers.oauth2.constants import (
|
||||||
|
SCOPE_OPENID,
|
||||||
|
SCOPE_OPENID_EMAIL,
|
||||||
|
SCOPE_OPENID_PROFILE,
|
||||||
|
)
|
||||||
|
from passbook.providers.oauth2.generators import (
|
||||||
|
generate_client_id,
|
||||||
|
generate_client_secret,
|
||||||
|
)
|
||||||
|
from passbook.providers.oauth2.models import (
|
||||||
|
ClientTypes,
|
||||||
|
OAuth2Provider,
|
||||||
|
ResponseTypes,
|
||||||
|
ScopeMapping,
|
||||||
|
)
|
||||||
|
|
||||||
|
LOGGER = get_logger()
|
||||||
|
APPLICATION_SLUG = "grafana"
|
||||||
|
|
||||||
|
|
||||||
|
@skipUnless(platform.startswith("linux"), "requires local docker")
|
||||||
|
class TestProviderOAuth2OAuth(SeleniumTestCase):
|
||||||
|
"""test OAuth with OAuth Provider flow"""
|
||||||
|
|
||||||
|
def setUp(self):
|
||||||
|
self.client_id = generate_client_id()
|
||||||
|
self.client_secret = generate_client_secret()
|
||||||
|
super().setUp()
|
||||||
|
|
||||||
|
def get_container_specs(self) -> Optional[Dict[str, Any]]:
|
||||||
|
return {
|
||||||
|
"image": "grafana/grafana:7.1.0",
|
||||||
|
"detach": True,
|
||||||
|
"network_mode": "host",
|
||||||
|
"auto_remove": True,
|
||||||
|
"healthcheck": Healthcheck(
|
||||||
|
test=["CMD", "wget", "--spider", "http://localhost:3000"],
|
||||||
|
interval=5 * 100 * 1000000,
|
||||||
|
start_period=1 * 100 * 1000000,
|
||||||
|
),
|
||||||
|
"environment": {
|
||||||
|
"GF_AUTH_GENERIC_OAUTH_ENABLED": "true",
|
||||||
|
"GF_AUTH_GENERIC_OAUTH_CLIENT_ID": self.client_id,
|
||||||
|
"GF_AUTH_GENERIC_OAUTH_CLIENT_SECRET": self.client_secret,
|
||||||
|
"GF_AUTH_GENERIC_OAUTH_SCOPES": "openid email profile",
|
||||||
|
"GF_AUTH_GENERIC_OAUTH_AUTH_URL": (
|
||||||
|
self.url("passbook_providers_oauth2:authorize")
|
||||||
|
),
|
||||||
|
"GF_AUTH_GENERIC_OAUTH_TOKEN_URL": (
|
||||||
|
self.url("passbook_providers_oauth2:token")
|
||||||
|
),
|
||||||
|
"GF_AUTH_GENERIC_OAUTH_API_URL": (
|
||||||
|
self.url("passbook_providers_oauth2:userinfo")
|
||||||
|
),
|
||||||
|
"GF_AUTH_SIGNOUT_REDIRECT_URL": (
|
||||||
|
self.url(
|
||||||
|
"passbook_providers_oauth2:end-session",
|
||||||
|
application_slug=APPLICATION_SLUG,
|
||||||
|
)
|
||||||
|
),
|
||||||
|
"GF_LOG_LEVEL": "debug",
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
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"
|
||||||
|
)
|
||||||
|
provider = OAuth2Provider.objects.create(
|
||||||
|
name="grafana",
|
||||||
|
client_type=ClientTypes.CONFIDENTIAL,
|
||||||
|
client_id=self.client_id,
|
||||||
|
client_secret=self.client_secret,
|
||||||
|
rsa_key=CertificateKeyPair.objects.first(),
|
||||||
|
redirect_uris="http://localhost:3000/",
|
||||||
|
authorization_flow=authorization_flow,
|
||||||
|
response_type=ResponseTypes.CODE,
|
||||||
|
)
|
||||||
|
provider.property_mappings.set(
|
||||||
|
ScopeMapping.objects.filter(
|
||||||
|
scope_name__in=[SCOPE_OPENID, SCOPE_OPENID_EMAIL, SCOPE_OPENID_PROFILE]
|
||||||
|
)
|
||||||
|
)
|
||||||
|
provider.save()
|
||||||
|
Application.objects.create(
|
||||||
|
name="Grafana", slug=APPLICATION_SLUG, provider=provider,
|
||||||
|
)
|
||||||
|
|
||||||
|
self.driver.get("http://localhost:3000")
|
||||||
|
self.driver.find_element(By.CLASS_NAME, "btn-service--oauth").click()
|
||||||
|
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(2)
|
||||||
|
self.assertEqual(
|
||||||
|
self.driver.find_element(By.CLASS_NAME, "pf-c-title").text,
|
||||||
|
"Redirect URI Error",
|
||||||
|
)
|
||||||
|
|
||||||
|
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"
|
||||||
|
)
|
||||||
|
provider = OAuth2Provider.objects.create(
|
||||||
|
name="grafana",
|
||||||
|
client_type=ClientTypes.CONFIDENTIAL,
|
||||||
|
client_id=self.client_id,
|
||||||
|
client_secret=self.client_secret,
|
||||||
|
rsa_key=CertificateKeyPair.objects.first(),
|
||||||
|
redirect_uris="http://localhost:3000/login/generic_oauth",
|
||||||
|
authorization_flow=authorization_flow,
|
||||||
|
response_type=ResponseTypes.CODE,
|
||||||
|
)
|
||||||
|
provider.property_mappings.set(
|
||||||
|
ScopeMapping.objects.filter(
|
||||||
|
scope_name__in=[SCOPE_OPENID, SCOPE_OPENID_EMAIL, SCOPE_OPENID_PROFILE]
|
||||||
|
)
|
||||||
|
)
|
||||||
|
provider.save()
|
||||||
|
Application.objects.create(
|
||||||
|
name="Grafana", slug=APPLICATION_SLUG, provider=provider,
|
||||||
|
)
|
||||||
|
|
||||||
|
self.driver.get("http://localhost:3000")
|
||||||
|
self.driver.find_element(By.CLASS_NAME, "btn-service--oauth").click()
|
||||||
|
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.driver.find_element(By.XPATH, "//a[contains(@href, '/profile')]").click()
|
||||||
|
self.assertEqual(
|
||||||
|
self.driver.find_element(By.CLASS_NAME, "page-header__title").text,
|
||||||
|
USER().name,
|
||||||
|
)
|
||||||
|
self.assertEqual(
|
||||||
|
self.driver.find_element(By.CSS_SELECTOR, "input[name=name]").get_attribute(
|
||||||
|
"value"
|
||||||
|
),
|
||||||
|
USER().name,
|
||||||
|
)
|
||||||
|
self.assertEqual(
|
||||||
|
self.driver.find_element(
|
||||||
|
By.CSS_SELECTOR, "input[name=email]"
|
||||||
|
).get_attribute("value"),
|
||||||
|
USER().email,
|
||||||
|
)
|
||||||
|
self.assertEqual(
|
||||||
|
self.driver.find_element(
|
||||||
|
By.CSS_SELECTOR, "input[name=login]"
|
||||||
|
).get_attribute("value"),
|
||||||
|
USER().email,
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_authorization_logout(self):
|
||||||
|
"""test OpenID Provider flow with logout"""
|
||||||
|
sleep(1)
|
||||||
|
# Bootstrap all needed objects
|
||||||
|
authorization_flow = Flow.objects.get(
|
||||||
|
slug="default-provider-authorization-implicit-consent"
|
||||||
|
)
|
||||||
|
provider = OAuth2Provider.objects.create(
|
||||||
|
name="grafana",
|
||||||
|
client_type=ClientTypes.CONFIDENTIAL,
|
||||||
|
client_id=self.client_id,
|
||||||
|
client_secret=self.client_secret,
|
||||||
|
rsa_key=CertificateKeyPair.objects.first(),
|
||||||
|
redirect_uris="http://localhost:3000/login/generic_oauth",
|
||||||
|
authorization_flow=authorization_flow,
|
||||||
|
response_type=ResponseTypes.CODE,
|
||||||
|
)
|
||||||
|
provider.property_mappings.set(
|
||||||
|
ScopeMapping.objects.filter(
|
||||||
|
scope_name__in=[SCOPE_OPENID, SCOPE_OPENID_EMAIL, SCOPE_OPENID_PROFILE]
|
||||||
|
)
|
||||||
|
)
|
||||||
|
provider.save()
|
||||||
|
Application.objects.create(
|
||||||
|
name="Grafana", slug=APPLICATION_SLUG, provider=provider,
|
||||||
|
)
|
||||||
|
|
||||||
|
self.driver.get("http://localhost:3000")
|
||||||
|
self.driver.find_element(By.CLASS_NAME, "btn-service--oauth").click()
|
||||||
|
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.driver.find_element(By.XPATH, "//a[contains(@href, '/profile')]").click()
|
||||||
|
self.assertEqual(
|
||||||
|
self.driver.find_element(By.CLASS_NAME, "page-header__title").text,
|
||||||
|
USER().name,
|
||||||
|
)
|
||||||
|
self.assertEqual(
|
||||||
|
self.driver.find_element(By.CSS_SELECTOR, "input[name=name]").get_attribute(
|
||||||
|
"value"
|
||||||
|
),
|
||||||
|
USER().name,
|
||||||
|
)
|
||||||
|
self.assertEqual(
|
||||||
|
self.driver.find_element(
|
||||||
|
By.CSS_SELECTOR, "input[name=email]"
|
||||||
|
).get_attribute("value"),
|
||||||
|
USER().email,
|
||||||
|
)
|
||||||
|
self.assertEqual(
|
||||||
|
self.driver.find_element(
|
||||||
|
By.CSS_SELECTOR, "input[name=login]"
|
||||||
|
).get_attribute("value"),
|
||||||
|
USER().email,
|
||||||
|
)
|
||||||
|
self.driver.find_element(By.CSS_SELECTOR, "[href='/logout']").click()
|
||||||
|
self.wait_for_url(
|
||||||
|
self.url(
|
||||||
|
"passbook_providers_oauth2:end-session",
|
||||||
|
application_slug=APPLICATION_SLUG,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
self.driver.find_element(By.ID, "logout").click()
|
||||||
|
|
||||||
|
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"
|
||||||
|
)
|
||||||
|
provider = OAuth2Provider.objects.create(
|
||||||
|
name="grafana",
|
||||||
|
authorization_flow=authorization_flow,
|
||||||
|
response_type=ResponseTypes.CODE,
|
||||||
|
client_type=ClientTypes.CONFIDENTIAL,
|
||||||
|
client_id=self.client_id,
|
||||||
|
client_secret=self.client_secret,
|
||||||
|
rsa_key=CertificateKeyPair.objects.first(),
|
||||||
|
redirect_uris="http://localhost:3000/login/generic_oauth",
|
||||||
|
)
|
||||||
|
provider.property_mappings.set(
|
||||||
|
ScopeMapping.objects.filter(
|
||||||
|
scope_name__in=[SCOPE_OPENID, SCOPE_OPENID_EMAIL, SCOPE_OPENID_PROFILE]
|
||||||
|
)
|
||||||
|
)
|
||||||
|
provider.save()
|
||||||
|
app = Application.objects.create(
|
||||||
|
name="Grafana", slug=APPLICATION_SLUG, provider=provider,
|
||||||
|
)
|
||||||
|
|
||||||
|
self.driver.get("http://localhost:3000")
|
||||||
|
self.driver.find_element(By.CLASS_NAME, "btn-service--oauth").click()
|
||||||
|
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.assertIn(
|
||||||
|
app.name,
|
||||||
|
self.driver.find_element(
|
||||||
|
By.XPATH, "/html/body/div[2]/div/main/div/form/div[2]/p[1]"
|
||||||
|
).text,
|
||||||
|
)
|
||||||
|
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.XPATH, "//a[contains(@href, '/profile')]")
|
||||||
|
)
|
||||||
|
)
|
||||||
|
self.driver.find_element(By.XPATH, "//a[contains(@href, '/profile')]").click()
|
||||||
|
self.assertEqual(
|
||||||
|
self.driver.find_element(By.CLASS_NAME, "page-header__title").text,
|
||||||
|
USER().name,
|
||||||
|
)
|
||||||
|
self.assertEqual(
|
||||||
|
self.driver.find_element(By.CSS_SELECTOR, "input[name=name]").get_attribute(
|
||||||
|
"value"
|
||||||
|
),
|
||||||
|
USER().name,
|
||||||
|
)
|
||||||
|
self.assertEqual(
|
||||||
|
self.driver.find_element(
|
||||||
|
By.CSS_SELECTOR, "input[name=email]"
|
||||||
|
).get_attribute("value"),
|
||||||
|
USER().email,
|
||||||
|
)
|
||||||
|
self.assertEqual(
|
||||||
|
self.driver.find_element(
|
||||||
|
By.CSS_SELECTOR, "input[name=login]"
|
||||||
|
).get_attribute("value"),
|
||||||
|
USER().email,
|
||||||
|
)
|
||||||
|
|
||||||
|
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"
|
||||||
|
)
|
||||||
|
provider = OAuth2Provider.objects.create(
|
||||||
|
name="grafana",
|
||||||
|
authorization_flow=authorization_flow,
|
||||||
|
response_type=ResponseTypes.CODE,
|
||||||
|
client_type=ClientTypes.CONFIDENTIAL,
|
||||||
|
client_id=self.client_id,
|
||||||
|
client_secret=self.client_secret,
|
||||||
|
rsa_key=CertificateKeyPair.objects.first(),
|
||||||
|
redirect_uris="http://localhost:3000/login/generic_oauth",
|
||||||
|
)
|
||||||
|
provider.property_mappings.set(
|
||||||
|
ScopeMapping.objects.filter(
|
||||||
|
scope_name__in=[SCOPE_OPENID, SCOPE_OPENID_EMAIL, SCOPE_OPENID_PROFILE]
|
||||||
|
)
|
||||||
|
)
|
||||||
|
provider.save()
|
||||||
|
app = Application.objects.create(
|
||||||
|
name="Grafana", slug=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.driver.get("http://localhost:3000")
|
||||||
|
self.driver.find_element(By.CLASS_NAME, "btn-service--oauth").click()
|
||||||
|
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",
|
||||||
|
)
|
|
@ -1,9 +1,11 @@
|
||||||
"""test OAuth2 OpenID Provider flow"""
|
"""test OAuth2 OpenID Provider flow"""
|
||||||
|
from json import loads
|
||||||
from sys import platform
|
from sys import platform
|
||||||
from time import sleep
|
from time import sleep
|
||||||
from typing import Any, Dict, Optional
|
|
||||||
from unittest.case import skipUnless
|
from unittest.case import skipUnless
|
||||||
|
|
||||||
|
from docker import DockerClient, from_env
|
||||||
|
from docker.models.containers import Container
|
||||||
from docker.types import Healthcheck
|
from docker.types import Healthcheck
|
||||||
from selenium.webdriver.common.by import By
|
from selenium.webdriver.common.by import By
|
||||||
from selenium.webdriver.common.keys import Keys
|
from selenium.webdriver.common.keys import Keys
|
||||||
|
@ -33,7 +35,6 @@ from passbook.providers.oauth2.models import (
|
||||||
)
|
)
|
||||||
|
|
||||||
LOGGER = get_logger()
|
LOGGER = get_logger()
|
||||||
APPLICATION_SLUG = "grafana"
|
|
||||||
|
|
||||||
|
|
||||||
@skipUnless(platform.startswith("linux"), "requires local docker")
|
@skipUnless(platform.startswith("linux"), "requires local docker")
|
||||||
|
@ -43,42 +44,36 @@ class TestProviderOAuth2OIDC(SeleniumTestCase):
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
self.client_id = generate_client_id()
|
self.client_id = generate_client_id()
|
||||||
self.client_secret = generate_client_secret()
|
self.client_secret = generate_client_secret()
|
||||||
|
self.application_slug = "test"
|
||||||
super().setUp()
|
super().setUp()
|
||||||
|
|
||||||
def get_container_specs(self) -> Optional[Dict[str, Any]]:
|
def setup_client(self) -> Container:
|
||||||
return {
|
"""Setup client saml-sp container which we test SAML against"""
|
||||||
"image": "grafana/grafana:7.1.0",
|
sleep(1)
|
||||||
"detach": True,
|
client: DockerClient = from_env()
|
||||||
"network_mode": "host",
|
container = client.containers.run(
|
||||||
"auto_remove": True,
|
image="beryju/oidc-test-client",
|
||||||
"healthcheck": Healthcheck(
|
detach=True,
|
||||||
test=["CMD", "wget", "--spider", "http://localhost:3000"],
|
network_mode="host",
|
||||||
|
auto_remove=True,
|
||||||
|
healthcheck=Healthcheck(
|
||||||
|
test=["CMD", "wget", "--spider", "http://localhost:9009/health"],
|
||||||
interval=5 * 100 * 1000000,
|
interval=5 * 100 * 1000000,
|
||||||
start_period=1 * 100 * 1000000,
|
start_period=1 * 100 * 1000000,
|
||||||
),
|
),
|
||||||
"environment": {
|
environment={
|
||||||
"GF_AUTH_GENERIC_OAUTH_ENABLED": "true",
|
"OIDC_CLIENT_ID": self.client_id,
|
||||||
"GF_AUTH_GENERIC_OAUTH_CLIENT_ID": self.client_id,
|
"OIDC_CLIENT_SECRET": self.client_secret,
|
||||||
"GF_AUTH_GENERIC_OAUTH_CLIENT_SECRET": self.client_secret,
|
"OIDC_PROVIDER": f"{self.live_server_url}/application/o/{self.application_slug}/",
|
||||||
"GF_AUTH_GENERIC_OAUTH_SCOPES": "openid email profile",
|
|
||||||
"GF_AUTH_GENERIC_OAUTH_AUTH_URL": (
|
|
||||||
self.url("passbook_providers_oauth2:authorize")
|
|
||||||
),
|
|
||||||
"GF_AUTH_GENERIC_OAUTH_TOKEN_URL": (
|
|
||||||
self.url("passbook_providers_oauth2:token")
|
|
||||||
),
|
|
||||||
"GF_AUTH_GENERIC_OAUTH_API_URL": (
|
|
||||||
self.url("passbook_providers_oauth2:userinfo")
|
|
||||||
),
|
|
||||||
"GF_AUTH_SIGNOUT_REDIRECT_URL": (
|
|
||||||
self.url(
|
|
||||||
"passbook_providers_oauth2:end-session",
|
|
||||||
application_slug=APPLICATION_SLUG,
|
|
||||||
)
|
|
||||||
),
|
|
||||||
"GF_LOG_LEVEL": "debug",
|
|
||||||
},
|
},
|
||||||
}
|
)
|
||||||
|
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)
|
||||||
|
|
||||||
def test_redirect_uri_error(self):
|
def test_redirect_uri_error(self):
|
||||||
"""test OpenID Provider flow (invalid redirect URI, check error message)"""
|
"""test OpenID Provider flow (invalid redirect URI, check error message)"""
|
||||||
|
@ -88,12 +83,12 @@ class TestProviderOAuth2OIDC(SeleniumTestCase):
|
||||||
slug="default-provider-authorization-implicit-consent"
|
slug="default-provider-authorization-implicit-consent"
|
||||||
)
|
)
|
||||||
provider = OAuth2Provider.objects.create(
|
provider = OAuth2Provider.objects.create(
|
||||||
name="grafana",
|
name=self.application_slug,
|
||||||
client_type=ClientTypes.CONFIDENTIAL,
|
client_type=ClientTypes.CONFIDENTIAL,
|
||||||
client_id=self.client_id,
|
client_id=self.client_id,
|
||||||
client_secret=self.client_secret,
|
client_secret=self.client_secret,
|
||||||
rsa_key=CertificateKeyPair.objects.first(),
|
rsa_key=CertificateKeyPair.objects.first(),
|
||||||
redirect_uris="http://localhost:3000/",
|
redirect_uris="http://localhost:9009/",
|
||||||
authorization_flow=authorization_flow,
|
authorization_flow=authorization_flow,
|
||||||
response_type=ResponseTypes.CODE,
|
response_type=ResponseTypes.CODE,
|
||||||
)
|
)
|
||||||
|
@ -104,11 +99,12 @@ class TestProviderOAuth2OIDC(SeleniumTestCase):
|
||||||
)
|
)
|
||||||
provider.save()
|
provider.save()
|
||||||
Application.objects.create(
|
Application.objects.create(
|
||||||
name="Grafana", slug=APPLICATION_SLUG, provider=provider,
|
name=self.application_slug, slug=self.application_slug, provider=provider,
|
||||||
)
|
)
|
||||||
|
self.container = self.setup_client()
|
||||||
|
|
||||||
|
self.driver.get("http://localhost:9009")
|
||||||
|
|
||||||
self.driver.get("http://localhost:3000")
|
|
||||||
self.driver.find_element(By.CLASS_NAME, "btn-service--oauth").click()
|
|
||||||
self.driver.find_element(By.ID, "id_uid_field").click()
|
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(USER().username)
|
||||||
self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
|
self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
|
||||||
|
@ -128,12 +124,12 @@ class TestProviderOAuth2OIDC(SeleniumTestCase):
|
||||||
slug="default-provider-authorization-implicit-consent"
|
slug="default-provider-authorization-implicit-consent"
|
||||||
)
|
)
|
||||||
provider = OAuth2Provider.objects.create(
|
provider = OAuth2Provider.objects.create(
|
||||||
name="grafana",
|
name=self.application_slug,
|
||||||
client_type=ClientTypes.CONFIDENTIAL,
|
client_type=ClientTypes.CONFIDENTIAL,
|
||||||
client_id=self.client_id,
|
client_id=self.client_id,
|
||||||
client_secret=self.client_secret,
|
client_secret=self.client_secret,
|
||||||
rsa_key=CertificateKeyPair.objects.first(),
|
rsa_key=CertificateKeyPair.objects.first(),
|
||||||
redirect_uris="http://localhost:3000/login/generic_oauth",
|
redirect_uris="http://localhost:9009/auth/callback",
|
||||||
authorization_flow=authorization_flow,
|
authorization_flow=authorization_flow,
|
||||||
response_type=ResponseTypes.CODE,
|
response_type=ResponseTypes.CODE,
|
||||||
)
|
)
|
||||||
|
@ -144,105 +140,29 @@ class TestProviderOAuth2OIDC(SeleniumTestCase):
|
||||||
)
|
)
|
||||||
provider.save()
|
provider.save()
|
||||||
Application.objects.create(
|
Application.objects.create(
|
||||||
name="Grafana", slug=APPLICATION_SLUG, provider=provider,
|
name=self.application_slug, slug=self.application_slug, provider=provider,
|
||||||
)
|
)
|
||||||
|
self.container = self.setup_client()
|
||||||
|
|
||||||
|
self.driver.get("http://localhost:9009")
|
||||||
|
|
||||||
self.driver.get("http://localhost:3000")
|
|
||||||
self.driver.find_element(By.CLASS_NAME, "btn-service--oauth").click()
|
|
||||||
self.driver.find_element(By.ID, "id_uid_field").click()
|
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(USER().username)
|
||||||
self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
|
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(USER().username)
|
||||||
self.driver.find_element(By.ID, "id_password").send_keys(Keys.ENTER)
|
self.driver.find_element(By.ID, "id_password").send_keys(Keys.ENTER)
|
||||||
self.driver.find_element(By.XPATH, "//a[contains(@href, '/profile')]").click()
|
|
||||||
self.assertEqual(
|
|
||||||
self.driver.find_element(By.CLASS_NAME, "page-header__title").text,
|
|
||||||
USER().name,
|
|
||||||
)
|
|
||||||
self.assertEqual(
|
|
||||||
self.driver.find_element(By.CSS_SELECTOR, "input[name=name]").get_attribute(
|
|
||||||
"value"
|
|
||||||
),
|
|
||||||
USER().name,
|
|
||||||
)
|
|
||||||
self.assertEqual(
|
|
||||||
self.driver.find_element(
|
|
||||||
By.CSS_SELECTOR, "input[name=email]"
|
|
||||||
).get_attribute("value"),
|
|
||||||
USER().email,
|
|
||||||
)
|
|
||||||
self.assertEqual(
|
|
||||||
self.driver.find_element(
|
|
||||||
By.CSS_SELECTOR, "input[name=login]"
|
|
||||||
).get_attribute("value"),
|
|
||||||
USER().email,
|
|
||||||
)
|
|
||||||
|
|
||||||
def test_authorization_logout(self):
|
self.wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, "pre")))
|
||||||
"""test OpenID Provider flow with logout"""
|
body = loads(self.driver.find_element(By.CSS_SELECTOR, "pre").text)
|
||||||
sleep(1)
|
|
||||||
# Bootstrap all needed objects
|
|
||||||
authorization_flow = Flow.objects.get(
|
|
||||||
slug="default-provider-authorization-implicit-consent"
|
|
||||||
)
|
|
||||||
provider = OAuth2Provider.objects.create(
|
|
||||||
name="grafana",
|
|
||||||
client_type=ClientTypes.CONFIDENTIAL,
|
|
||||||
client_id=self.client_id,
|
|
||||||
client_secret=self.client_secret,
|
|
||||||
rsa_key=CertificateKeyPair.objects.first(),
|
|
||||||
redirect_uris="http://localhost:3000/login/generic_oauth",
|
|
||||||
authorization_flow=authorization_flow,
|
|
||||||
response_type=ResponseTypes.CODE,
|
|
||||||
)
|
|
||||||
provider.property_mappings.set(
|
|
||||||
ScopeMapping.objects.filter(
|
|
||||||
scope_name__in=[SCOPE_OPENID, SCOPE_OPENID_EMAIL, SCOPE_OPENID_PROFILE]
|
|
||||||
)
|
|
||||||
)
|
|
||||||
provider.save()
|
|
||||||
Application.objects.create(
|
|
||||||
name="Grafana", slug=APPLICATION_SLUG, provider=provider,
|
|
||||||
)
|
|
||||||
|
|
||||||
self.driver.get("http://localhost:3000")
|
self.assertEqual(body["IDTokenClaims"]["nickname"], USER().username)
|
||||||
self.driver.find_element(By.CLASS_NAME, "btn-service--oauth").click()
|
self.assertEqual(body["UserInfo"]["nickname"], USER().username)
|
||||||
self.driver.find_element(By.ID, "id_uid_field").click()
|
|
||||||
self.driver.find_element(By.ID, "id_uid_field").send_keys(USER().username)
|
self.assertEqual(body["IDTokenClaims"]["name"], USER().name)
|
||||||
self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
|
self.assertEqual(body["UserInfo"]["name"], USER().name)
|
||||||
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(body["IDTokenClaims"]["email"], USER().email)
|
||||||
self.driver.find_element(By.XPATH, "//a[contains(@href, '/profile')]").click()
|
self.assertEqual(body["UserInfo"]["email"], USER().email)
|
||||||
self.assertEqual(
|
|
||||||
self.driver.find_element(By.CLASS_NAME, "page-header__title").text,
|
|
||||||
USER().name,
|
|
||||||
)
|
|
||||||
self.assertEqual(
|
|
||||||
self.driver.find_element(By.CSS_SELECTOR, "input[name=name]").get_attribute(
|
|
||||||
"value"
|
|
||||||
),
|
|
||||||
USER().name,
|
|
||||||
)
|
|
||||||
self.assertEqual(
|
|
||||||
self.driver.find_element(
|
|
||||||
By.CSS_SELECTOR, "input[name=email]"
|
|
||||||
).get_attribute("value"),
|
|
||||||
USER().email,
|
|
||||||
)
|
|
||||||
self.assertEqual(
|
|
||||||
self.driver.find_element(
|
|
||||||
By.CSS_SELECTOR, "input[name=login]"
|
|
||||||
).get_attribute("value"),
|
|
||||||
USER().email,
|
|
||||||
)
|
|
||||||
self.driver.find_element(By.CSS_SELECTOR, "[href='/logout']").click()
|
|
||||||
self.wait_for_url(
|
|
||||||
self.url(
|
|
||||||
"passbook_providers_oauth2:end-session",
|
|
||||||
application_slug=APPLICATION_SLUG,
|
|
||||||
)
|
|
||||||
)
|
|
||||||
self.driver.find_element(By.ID, "logout").click()
|
|
||||||
|
|
||||||
def test_authorization_consent_explicit(self):
|
def test_authorization_consent_explicit(self):
|
||||||
"""test OpenID Provider flow (default authorization flow with explicit consent)"""
|
"""test OpenID Provider flow (default authorization flow with explicit consent)"""
|
||||||
|
@ -252,14 +172,14 @@ class TestProviderOAuth2OIDC(SeleniumTestCase):
|
||||||
slug="default-provider-authorization-explicit-consent"
|
slug="default-provider-authorization-explicit-consent"
|
||||||
)
|
)
|
||||||
provider = OAuth2Provider.objects.create(
|
provider = OAuth2Provider.objects.create(
|
||||||
name="grafana",
|
name=self.application_slug,
|
||||||
authorization_flow=authorization_flow,
|
authorization_flow=authorization_flow,
|
||||||
response_type=ResponseTypes.CODE,
|
response_type=ResponseTypes.CODE,
|
||||||
client_type=ClientTypes.CONFIDENTIAL,
|
client_type=ClientTypes.CONFIDENTIAL,
|
||||||
client_id=self.client_id,
|
client_id=self.client_id,
|
||||||
client_secret=self.client_secret,
|
client_secret=self.client_secret,
|
||||||
rsa_key=CertificateKeyPair.objects.first(),
|
rsa_key=CertificateKeyPair.objects.first(),
|
||||||
redirect_uris="http://localhost:3000/login/generic_oauth",
|
redirect_uris="http://localhost:9009/auth/callback",
|
||||||
)
|
)
|
||||||
provider.property_mappings.set(
|
provider.property_mappings.set(
|
||||||
ScopeMapping.objects.filter(
|
ScopeMapping.objects.filter(
|
||||||
|
@ -268,11 +188,12 @@ class TestProviderOAuth2OIDC(SeleniumTestCase):
|
||||||
)
|
)
|
||||||
provider.save()
|
provider.save()
|
||||||
app = Application.objects.create(
|
app = Application.objects.create(
|
||||||
name="Grafana", slug=APPLICATION_SLUG, provider=provider,
|
name=self.application_slug, slug=self.application_slug, provider=provider,
|
||||||
)
|
)
|
||||||
|
self.container = self.setup_client()
|
||||||
|
|
||||||
|
self.driver.get("http://localhost:9009")
|
||||||
|
|
||||||
self.driver.get("http://localhost:3000")
|
|
||||||
self.driver.find_element(By.CLASS_NAME, "btn-service--oauth").click()
|
|
||||||
self.driver.find_element(By.ID, "id_uid_field").click()
|
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(USER().username)
|
||||||
self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
|
self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
|
||||||
|
@ -291,34 +212,17 @@ class TestProviderOAuth2OIDC(SeleniumTestCase):
|
||||||
sleep(1)
|
sleep(1)
|
||||||
self.driver.find_element(By.CSS_SELECTOR, "[type=submit]").click()
|
self.driver.find_element(By.CSS_SELECTOR, "[type=submit]").click()
|
||||||
|
|
||||||
self.wait.until(
|
self.wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, "pre")))
|
||||||
ec.presence_of_element_located(
|
body = loads(self.driver.find_element(By.CSS_SELECTOR, "pre").text)
|
||||||
(By.XPATH, "//a[contains(@href, '/profile')]")
|
|
||||||
)
|
self.assertEqual(body["IDTokenClaims"]["nickname"], USER().username)
|
||||||
)
|
self.assertEqual(body["UserInfo"]["nickname"], USER().username)
|
||||||
self.driver.find_element(By.XPATH, "//a[contains(@href, '/profile')]").click()
|
|
||||||
self.assertEqual(
|
self.assertEqual(body["IDTokenClaims"]["name"], USER().name)
|
||||||
self.driver.find_element(By.CLASS_NAME, "page-header__title").text,
|
self.assertEqual(body["UserInfo"]["name"], USER().name)
|
||||||
USER().name,
|
|
||||||
)
|
self.assertEqual(body["IDTokenClaims"]["email"], USER().email)
|
||||||
self.assertEqual(
|
self.assertEqual(body["UserInfo"]["email"], USER().email)
|
||||||
self.driver.find_element(By.CSS_SELECTOR, "input[name=name]").get_attribute(
|
|
||||||
"value"
|
|
||||||
),
|
|
||||||
USER().name,
|
|
||||||
)
|
|
||||||
self.assertEqual(
|
|
||||||
self.driver.find_element(
|
|
||||||
By.CSS_SELECTOR, "input[name=email]"
|
|
||||||
).get_attribute("value"),
|
|
||||||
USER().email,
|
|
||||||
)
|
|
||||||
self.assertEqual(
|
|
||||||
self.driver.find_element(
|
|
||||||
By.CSS_SELECTOR, "input[name=login]"
|
|
||||||
).get_attribute("value"),
|
|
||||||
USER().email,
|
|
||||||
)
|
|
||||||
|
|
||||||
def test_authorization_denied(self):
|
def test_authorization_denied(self):
|
||||||
"""test OpenID Provider flow (default authorization with access deny)"""
|
"""test OpenID Provider flow (default authorization with access deny)"""
|
||||||
|
@ -328,14 +232,14 @@ class TestProviderOAuth2OIDC(SeleniumTestCase):
|
||||||
slug="default-provider-authorization-explicit-consent"
|
slug="default-provider-authorization-explicit-consent"
|
||||||
)
|
)
|
||||||
provider = OAuth2Provider.objects.create(
|
provider = OAuth2Provider.objects.create(
|
||||||
name="grafana",
|
name=self.application_slug,
|
||||||
authorization_flow=authorization_flow,
|
authorization_flow=authorization_flow,
|
||||||
response_type=ResponseTypes.CODE,
|
response_type=ResponseTypes.CODE,
|
||||||
client_type=ClientTypes.CONFIDENTIAL,
|
client_type=ClientTypes.CONFIDENTIAL,
|
||||||
client_id=self.client_id,
|
client_id=self.client_id,
|
||||||
client_secret=self.client_secret,
|
client_secret=self.client_secret,
|
||||||
rsa_key=CertificateKeyPair.objects.first(),
|
rsa_key=CertificateKeyPair.objects.first(),
|
||||||
redirect_uris="http://localhost:3000/login/generic_oauth",
|
redirect_uris="http://localhost:9009/auth/callback",
|
||||||
)
|
)
|
||||||
provider.property_mappings.set(
|
provider.property_mappings.set(
|
||||||
ScopeMapping.objects.filter(
|
ScopeMapping.objects.filter(
|
||||||
|
@ -344,15 +248,17 @@ class TestProviderOAuth2OIDC(SeleniumTestCase):
|
||||||
)
|
)
|
||||||
provider.save()
|
provider.save()
|
||||||
app = Application.objects.create(
|
app = Application.objects.create(
|
||||||
name="Grafana", slug=APPLICATION_SLUG, provider=provider,
|
name=self.application_slug, slug=self.application_slug, provider=provider,
|
||||||
)
|
)
|
||||||
|
|
||||||
negative_policy = ExpressionPolicy.objects.create(
|
negative_policy = ExpressionPolicy.objects.create(
|
||||||
name="negative-static", expression="return False"
|
name="negative-static", expression="return False"
|
||||||
)
|
)
|
||||||
PolicyBinding.objects.create(target=app, policy=negative_policy, order=0)
|
PolicyBinding.objects.create(target=app, policy=negative_policy, order=0)
|
||||||
self.driver.get("http://localhost:3000")
|
|
||||||
self.driver.find_element(By.CLASS_NAME, "btn-service--oauth").click()
|
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").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(USER().username)
|
||||||
self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
|
self.driver.find_element(By.ID, "id_uid_field").send_keys(Keys.ENTER)
|
||||||
|
|
Reference in New Issue