eliott/authentik
1
0
Fork 0
mirror of https://github.com/goauthentik/authentik synced 2026-04-25 17:15:26 +02:00
Code Issues Packages Projects Releases Wiki Activity

providers/oauth2: Automated OpenID Conformance tests (#14785)

Browse Source 
* some progress

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* somewhat working?

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* remove some previous debugging things

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* make it kinda work

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* revert more debugging stuff

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* make tests mostly work

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* smaller screenshots?

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* remove debugging

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* sleep a bit before checking again

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* cleanup, restart loop when we finished an operation

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* refactor conformance helper to requests (thanks chatgpt)

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* properly install subtests

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* maybe run in CI?

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* dont hardcode IP

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* fix network and cookie deletion

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* upload cert exports

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* test

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* attempt to switch to generated

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* make it work generated?

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* fix teardown

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* re-add implicit and fix?

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* Revert "re-add implicit and fix?"

This reverts commit 6a4d15fc22cf4b27ffa428be9ecc9a0e778961c6.

* Revert "fix teardown"

This reverts commit cb96b0cb988acedec1fe72ec437b68e2c38ed6b1.

* Revert "make it work generated?"

This reverts commit 4e29d2c5737ee9aaad6c0f4701caf7e0fb110e15.

* Revert "attempt to switch to generated"

This reverts commit 6f851e021d305a93be9cfbb4a9b6783231b6d7cf.

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* check authorize request param earlier

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* fix some

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* fix basic suite?

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* another actual fix; don't return access_token when using response_type id_token

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* add implicit test

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* add custom profile scope that includes standard scopes to return number of warnings

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* use actual timestamp

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* fix missing offline_access, use scoped issuer

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* only run basic+implicit for now, fix other tests

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* split up

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* fix offline_access tests

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* fix waiting for compete on error

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* fix duplicate artifact

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* fix artifact

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* 👀

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* fix typing

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* typing

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* fix implicit

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* don't wait for conformance test

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

* more disk space

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

---------

Signed-off-by: Jens Langhammer <jens@goauthentik.io>
This commit is contained in:
Jens L.
2025-12-22 20:21:22 +01:00
committed by GitHub
parent b66e4074e2
commit 3cd1a31365
12 changed files with 446 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
tests/e2e/compose.yml
View File

@@ -4,6 +4,8 @@ services:
shm_size: 2g
network_mode: host
restart: always
extra_hosts:
- "host.docker.internal:host-gateway"
mailpit:
image: docker.io/axllent/mailpit:v1.28.0
ports:

1
tests/e2e/utils.py
View File

@@ -177,6 +177,7 @@ class SeleniumTestCase(DockerTestCase, StaticLiveServerTestCase):
def _get_driver(self) -> WebDriver:
count = 0
opts = webdriver.ChromeOptions()
opts.accept_insecure_certs = True
opts.add_argument("--disable-search-engine-choice-screen")
# This breaks selenium when running remotely...?
# opts.set_capability("goog:loggingPrefs", {"browser": "ALL"})

0
tests/openid_conformance/__init__.py Normal file
View File

155
tests/openid_conformance/base.py Normal file
View File

@@ -0,0 +1,155 @@
from os import makedirs
from pathlib import Path
from time import sleep
from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions as ec
from authentik.blueprints.tests import apply_blueprint, reconcile_app
from authentik.providers.oauth2.models import OAuth2Provider
from tests.e2e.utils import SeleniumTestCase
from tests.openid_conformance.conformance import Conformance
class TestOpenIDConformance(SeleniumTestCase):
conformance: Conformance
@apply_blueprint(
"default/flow-default-authentication-flow.yaml",
"default/flow-default-invalidation-flow.yaml",
)
@apply_blueprint(
"default/flow-default-provider-authorization-implicit-consent.yaml",
"default/flow-default-provider-invalidation.yaml",
)
@apply_blueprint("system/providers-oauth2.yaml")
@reconcile_app("authentik_crypto")
@apply_blueprint("testing/oidc-conformance.yaml")
def setUp(self):
super().setUp()
makedirs(Path(__file__).parent / "exports", exist_ok=True)
provider_a = OAuth2Provider.objects.get(
client_id="4054d882aff59755f2f279968b97ce8806a926e1"
)
provider_b = OAuth2Provider.objects.get(
client_id="ad64aeaf1efe388ecf4d28fcc537e8de08bcae26"
)
self.test_plan_config = {
"alias": "authentik",
"description": "authentik",
"server": {
"discoveryUrl": self.url(
"authentik_providers_oauth2:provider-info",
application_slug="oidc-conformance-1",
),
},
"client": {
"client_id": "4054d882aff59755f2f279968b97ce8806a926e1",
"client_secret": provider_a.client_secret,
},
"client_secret_post": {
"client_id": "4054d882aff59755f2f279968b97ce8806a926e1",
"client_secret": provider_a.client_secret,
},
"client2": {
"client_id": "ad64aeaf1efe388ecf4d28fcc537e8de08bcae26",
"client_secret": provider_b.client_secret,
},
"consent": {},
}
self.test_variant = {
"server_metadata": "discovery",
"client_registration": "static_client",
}
def run_test(self, test_name: str, test_plan_config: dict):
# Create a Conformance instance...
self.conformance = Conformance(f"https://{self.host}:8443/", None, verify_ssl=False)
test_plan = self.conformance.create_test_plan(
test_name,
test_plan_config,
self.test_variant,
)
plan_id = test_plan["id"]
for test in test_plan["modules"]:
with self.subTest(test["testModule"], **test["variant"]):
# Fetch name and variant of the next test to run
module_name = test["testModule"]
variant = test["variant"]
module_instance = self.conformance.create_test_from_plan_with_variant(
plan_id, module_name, variant
)
module_id = module_instance["id"]
self.run_single_test(module_id)
self.conformance.wait_for_state(module_id, ["FINISHED"], timeout=self.wait_timeout)
sleep(2)
self.conformance.export_html(plan_id, Path(__file__).parent / "exports")
def run_single_test(self, module_id: str):
"""Process instructions for a single test, navigate to browser URLs and take screenshots"""
tested_browser_url = 0
uploaded_image = 0
cleared_cookies = False
while True:
# Fetch all info
test_status = self.conformance.get_test_status(module_id)
test_log = self.conformance.get_test_log(module_id)
test_info = self.conformance.get_module_info(module_id)
# Check status early, if we're finished already we don't want to do anything extra
if test_info["status"] in ["INTERRUPTED", "FINISHED"]:
return
# Check if we need to clear cookies - tests only indicates this in their written summary
# so this check is a bit brittle
if "cookies" in test_info["summary"] and not cleared_cookies:
# Navigate to our origin to delete cookies in the right context
self.driver.get(self.url("authentik_api:user-me") + "?format=json")
self.driver.delete_all_cookies()
cleared_cookies = True
# Check if we need deal with any browser URLs
browser_urls = test_status.get("browser", {}).get("urls", [])
if len(browser_urls) > tested_browser_url:
self.do_browser(browser_urls[tested_browser_url])
tested_browser_url += 1
continue
# Check if we need to upload any items
upload_items = [x for x in test_log if "upload" in x]
if len(upload_items) > uploaded_image:
screenshot = self.get_screenshot()
self.conformance.upload_image(
module_id, upload_items[uploaded_image]["upload"], screenshot
)
sleep(3)
uploaded_image += 1
continue
sleep(0.1)
def get_screenshot(self):
"""Get a screenshot, but resize the window first so we don't exceed 500kb"""
self.driver.set_window_size(800, 600)
screenshot = f"data:image/jpeg;base64,{self.driver.get_screenshot_as_base64()}"
self.driver.maximize_window()
return screenshot
def do_browser(self, url):
"""For any specific OpenID Conformance test, execute the operations required"""
self.driver.get(url)
should_expect_completion = False
if "if/flow/default-authentication-flow" in self.driver.current_url:
self.logger.debug("Logging in")
self.login()
should_expect_completion = True
if "prompt=consent" in url or "offline_access" in url:
self.logger.debug("Authorizing")
self.wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, "ak-flow-executor")))
sleep(1)
flow_executor = self.get_shadow_root("ak-flow-executor")
consent_stage = self.get_shadow_root("ak-stage-consent", flow_executor)
consent_stage.find_element(
By.CSS_SELECTOR,
"[type=submit]",
).click()
should_expect_completion = True
if should_expect_completion:
self.wait.until(ec.presence_of_element_located((By.CSS_SELECTOR, "#complete")))

215
tests/openid_conformance/conformance.py Normal file
View File

@@ -0,0 +1,215 @@
import re
import zipfile
from json import dumps
from os import devnull
from pathlib import Path
from time import sleep, time
from typing import Any, cast
from requests import RequestException
from requests.adapters import HTTPAdapter
from requests.sessions import default_headers
from urllib3.util.retry import Retry
from authentik.lib.utils.http import get_http_session
class ConformanceException(Exception):
"""Exception in conformance testing"""
class Conformance:
HTTP_OK = 200
HTTP_CREATED = 201
def __init__(self, api_url_base: str, api_token: str | None, verify_ssl: bool):
self.api_url_base = api_url_base
self.session = get_http_session()
self.session.verify = verify_ssl
retries = Retry(
total=5,
backoff_factor=1,
status_forcelist=[500, 502, 503, 504],
allowed_methods=["GET", "POST"],
)
self.session.mount("https://", HTTPAdapter(max_retries=retries))
self.session.mount("http://", HTTPAdapter(max_retries=retries))
self.session.headers.update({"Content-Type": "application/json"})
if api_token is not None:
self.session.headers.update({"Authorization": f"Bearer {api_token}"})
def get_all_test_modules(self) -> list[dict[str, Any]]:
url = f"{self.api_url_base}/api/runner/available"
response = self.session.get(url)
if response.status_code != Conformance.HTTP_OK:
raise ConformanceException(
f"get_all_test_modules failed - HTTP {response.status_code} {response.text}"
)
return cast(list[dict[str, Any]], response.json())
def get_test_status(self, module_id: str) -> dict[str, Any]:
url = f"{self.api_url_base}/api/runner/{module_id}"
response = self.session.get(url)
if response.status_code != Conformance.HTTP_OK:
raise ConformanceException(
f"get_test_status failed - HTTP {response.status_code} {response.text}"
)
return cast(dict[str, Any], response.json())
def export_html(self, plan_id: str, path: Path | str) -> Path:
for _ in range(5):
url = f"{self.api_url_base}/api/plan/exporthtml/{plan_id}"
try:
with self.session.get(url, stream=True) as response:
if response.status_code != Conformance.HTTP_OK:
raise ConformanceException(
f"exporthtml failed - HTTP {response.status_code} {response.text}"
)
cd = response.headers.get("content-disposition", "")
local_filename = Path(re.findall('filename="(.+)"', cd)[0])
full_path = Path(path) / local_filename
with open(full_path, "wb") as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
zip_file = zipfile.ZipFile(full_path)
ret = zip_file.testzip()
if ret is not None:
raise ConformanceException(f"export_html returned corrupt zip file: {ret}")
return full_path
except RequestException as e:
print(f"requests {url} exception {e} caught - retrying")
sleep(1)
raise ConformanceException(f"export_html for {plan_id} failed even after retries")
def create_certification_package(
self,
plan_id: str,
conformance_pdf_path: Path | str,
rp_logs_zip_path: Path | str | None = None,
output_zip_directory: Path | str = "./",
) -> None:
with (
open(conformance_pdf_path, "rb") as cert_pdf,
open(rp_logs_zip_path, "rb") if rp_logs_zip_path else open(devnull, "rb") as rp_logs,
):
files = {
"certificationOfConformancePdf": cert_pdf,
"clientSideData": rp_logs,
}
headers = default_headers()
headers.pop("Content-Type", None)
url = f"{self.api_url_base}/api/plan/{plan_id}/certificationpackage"
response = self.session.post(url, files=files, headers=headers)
if response.status_code != Conformance.HTTP_OK:
raise ConformanceException(
f"certificationpackage failed - HTTP {response.status_code} {response.text}"
)
cd = response.headers.get("content-disposition", "")
local_filename = Path(re.findall('filename="(.+)"', cd)[0])
full_path = Path(output_zip_directory) / local_filename
with open(full_path, "wb") as f:
f.write(response.content)
print(f"Certification package zip for plan id {plan_id} written to {full_path}")
def create_test_plan(
self, name: str, configuration: dict[str, Any], variant: dict[str, Any] | None = None
) -> dict[str, Any]:
url = f"{self.api_url_base}/api/plan"
payload = {"planName": name}
if variant is not None:
payload["variant"] = dumps(variant)
response = self.session.post(url, params=payload, data=dumps(configuration))
if response.status_code != Conformance.HTTP_CREATED:
raise ConformanceException(
f"create_test_plan failed - HTTP {response.status_code} {response.text}"
)
return cast(dict[str, Any], response.json())
def create_test(self, test_name: str, configuration: dict[str, Any]) -> dict[str, Any]:
url = f"{self.api_url_base}/api/runner"
payload = {"test": test_name}
response = self.session.post(url, params=payload, data=dumps(configuration))
if response.status_code != Conformance.HTTP_CREATED:
raise ConformanceException(
f"create_test failed - HTTP {response.status_code} {response.text}"
)
return cast(dict[str, Any], response.json())
def create_test_from_plan(self, plan_id: str, test_name: str) -> dict[str, Any]:
url = f"{self.api_url_base}/api/runner"
payload = {"test": test_name, "plan": plan_id}
response = self.session.post(url, params=payload)
if response.status_code != Conformance.HTTP_CREATED:
raise ConformanceException(
f"create_test_from_plan failed - HTTP {response.status_code} {response.text}"
)
return cast(dict[str, Any], response.json())
def create_test_from_plan_with_variant(
self, plan_id: str, test_name: str, variant: dict[str, Any]
) -> dict[str, Any]:
url = f"{self.api_url_base}/api/runner"
payload = {"test": test_name, "plan": plan_id}
if variant is not None:
payload["variant"] = dumps(variant)
response = self.session.post(url, params=payload)
if response.status_code != Conformance.HTTP_CREATED:
raise ConformanceException(
"create_test_from_plan_with_variant failed - "
f"HTTP {response.status_code} {response.text}"
)
return cast(dict[str, Any], response.json())
def get_module_info(self, module_id: str) -> dict[str, Any]:
url = f"{self.api_url_base}/api/info/{module_id}"
response = self.session.get(url)
if response.status_code != Conformance.HTTP_OK:
raise ConformanceException(
f"get_module_info failed - HTTP {response.status_code} {response.text}"
)
return cast(dict[str, Any], response.json())
def get_test_log(self, module_id: str) -> list[dict[str, Any]]:
url = f"{self.api_url_base}/api/log/{module_id}"
response = self.session.get(url)
if response.status_code != Conformance.HTTP_OK:
raise ConformanceException(
f"get_test_log failed - HTTP {response.status_code} {response.text}"
)
return cast(list[dict[str, Any]], response.json())
def upload_image(self, log_id: str, placeholder: str, data: Any) -> None:
url = f"{self.api_url_base}/api/log/{log_id}/images/{placeholder}"
response = self.session.post(url, data=data, headers={"Content-Type": "text/plain"})
if response.status_code != Conformance.HTTP_OK:
raise ConformanceException(
f"upload_image failed - HTTP {response.status_code} {response.text}"
)
def start_test(self, module_id: str) -> dict[str, Any]:
url = f"{self.api_url_base}/api/runner/{module_id}"
response = self.session.post(url)
if response.status_code != Conformance.HTTP_OK:
raise ConformanceException(
f"start_test failed - HTTP {response.status_code} {response.text}"
)
return cast(dict[str, Any], response.json())
def wait_for_state(self, module_id: str, required_states: list[str], timeout: int = 240) -> str:
timeout_at = time() + timeout
while time() < timeout_at:
info = self.get_module_info(module_id)
status: str | None = info.get("status")
if status in required_states:
return status
if status == "INTERRUPTED":
raise ConformanceException(f"Test module {module_id} has moved to INTERRUPTED")
sleep(1)
raise ConformanceException(
f"Timed out waiting for test module {module_id} "
f"to be in one of states: {required_states}"
)

10
tests/openid_conformance/test_basic.py Normal file
View File

@@ -0,0 +1,10 @@
from tests.e2e.utils import retry
from tests.openid_conformance.base import TestOpenIDConformance
class TestOpenIDConformanceBasic(TestOpenIDConformance):
@retry()
def test_oidcc_basic_certification_test(self):
test_plan_name = "oidcc-basic-certification-test-plan"
self.run_test(test_plan_name, self.test_plan_config)

10
tests/openid_conformance/test_implicit.py Normal file
View File

@@ -0,0 +1,10 @@
from tests.e2e.utils import retry
from tests.openid_conformance.base import TestOpenIDConformance
class TestOpenIDConformanceImplicit(TestOpenIDConformance):
@retry()
def test_oidcc_implicit_certification_test_plan(self):
test_plan_name = "oidcc-implicit-certification-test-plan"
self.run_test(test_plan_name, self.test_plan_config)