root: init rust code with worker

[rissson]

What changed

This PR makes a Rust binary the main entrypoint for running authentik. Here's what's changed:

• ak worker now runs Rust. The worker processes aren't spawned from Python anymore, but directly from Rust. That means that each of them have to load Django, making their startup slower, but in turn we save on RAM usage since we don't have a top-level process that uses a huge amount of memory doing nothing.
  • only 1 worker is started first, which will run the migrations, and listen on a unix socket for healthchecks and metrics (more on that later)
  • all other workers are started once the first one finished, and will start processing tasks directly
  • the worker doesn't run healthcheck and metrics servers in Python anymore, but in Rust. That includes the worker statuses reporting to the database for display in metrics and the admin interface.
  • About worker statuses, they are done every 30 seconds, and queried without locking anymore. This means that it can take up to 1 minute for the status to be correct in the UI and in metrics, but avoids a bunch of issues such as dedicated a postgresql connection to only this use case, and error handling about lock retention.
  • ak apply_blueprint isn't run from the lifecycle/ak script anymore, but by the first worker on startup. This allows for a faster startup since we don't need to load Django an extra time. It also fixes the issue of it failing if migrations hadn't run yet.
• ak server also starts Rust, which in turns starts Go, which in turns starts Python. See below for a detailed breakdown of who is listening where for what
• the $TMPDIR/authentik-mode is not set from bash (lifecycle/ak) anymore, but from Rust, since we use that mode throughout the code to change various behaviours
• a new mode has been introduced, allinone, for when the worker and server run at the same time. This is aimed at dev setups only, and will not be advertised for now.
• ak healthcheck is now written in Rust
• Metrics are handled by Rust. This means that Python doesn't need to compute nor return metrics data from /-/metrics/ anymore, speeding up that request and freeing a slot for another one.
• For development, code reloading isn't handled by ./manage.py dev_server anymore, but by watchexec. This allows for a unified setup between Python, Rust and Go
• In the main container, the Golang build isn't doing cross-compile anymore, because we need to copy some of the Go tools to build aws-lc-rs-fips-sys.

What's new

A bunch of Rust code, which in some places is incomplete:

• src/axum/trace.rs: this is missing a few fields when logging requests. Fortunately, we only log metrics (in all cases) and healthcheck requests (for the worker). These extra fields will be added in follow-up PRs, mainly to reduce the size of this one, but they're written already in The Coup™ #18122
• src/axum/server.rs: we're missing a bunch of ways to accept requests, such as the proxy protocol. These will be added in later PRs for the same reasons as previously mentioned
• graceful shutdown is implemented where you'd expect (worker processes, server process, HTTP requests)

Other details

This adds 14MB to the final docker image

Listeners depending on the mode (/tmp is used as a shorthand for $TMPDIR):

server

• gunicorn: /tmp/authentik-core.sock
• go server: /tmp/authentik.sock for HTTP requests, listen.http, listen.https, /tmp/authentik-server-metrics.sock
• rust: listen.metrics, /tmp/authentik-metrics.sock for metrics requests

In this mode, only metrics go through the rust process. It goes like this:

• Rust grabs its own metrics (none for now)
• Rust grabs metrics from Go
• Go notifies gunicorn that the monitoring_set signal should be sent
• Rust grabs metrics from Python by importing prometheus_client itself
• metrics are returned

Healthchecks are handled as before, by gunicorn

worker

• first started worker: /tmp/authentik-worker.sock for healthchecks and metrics
• Rust: listen.http for healthchecks, listen.metrics and /tmp/authentik-metrics.sock for metrics

In this mode, everything goes through rust:

• healthchecks run some checks from Rust (db, whether worker processes are running) and then queries the first worker to make sure that its healthchecks pass
• metrics work somewhat similarly as the server:
  • Rust grabs its own metrics (none for now)
  • Rust notifies the first worker process that monitoring_set should be sent
  • Rust grabs metrics from Python by importing prometheus_client itself

allinone

This is a mix and match of the above:

• gunicorn: /tmp/authentik-core.sock
• go server: /tmp/authentik.sock for HTTP requests, listen.http, listen.https, /tmp/authentik-server-metrics.sock
• first started worker: /tmp/authentik-worker.sock for healthchecks and metrics
• rust: listen.metrics, /tmp/authentik-metrics.sock for metrics requests

For metrics, the setup is the same as the server, as sending monitoring_set only on to gunicorn is sufficient.

For healthchecks, the setup is the same as the server, and we don't check the workers health at any point. This is fine since this is a development mode only. Once the server is written in Rust as well, we can improve this.

---

Closes #18292

[netlify]

✅ Deploy Preview for authentik-docs ready!

Name	Link
🔨 Latest commit	9f16d60
🔍 Latest deploy log	https://app.netlify.com/projects/authentik-docs/deploys/69bc42382c94e100098033f0
😎 Deploy Preview	https://deploy-preview-21018--authentik-docs.netlify.app
📱 Preview on mobile	Toggle QR Code... Use your smartphone camera to open QR code link.

---

To edit notification comments on pull requests, go to your Netlify project configuration.

[codecov]

❌ 7 Tests Failed:

Tests completed	Failed	Passed	Skipped
3124	7	3117	2

View the top 3 failed test(s) by shortest run time

tests.e2e.test_provider_oidc_implicit.TestProviderOAuth2OIDCImplicit::test_authorization_denied

Stack Traces | 3.08s run time

self = <unittest.case._Outcome object at 0x7f3b9d699310>
test_case = <tests.e2e.test_provider_oidc_implicit.TestProviderOAuth2OIDCImplicit testMethod=test_authorization_denied>
subTest = False

    @contextlib.contextmanager
    def testPartExecutor(self, test_case, subTest=False):
        old_success = self.success
        self.success = True
        try:
>           yield

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:58: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc_implicit.TestProviderOAuth2OIDCImplicit testMethod=test_authorization_denied>
result = <TestCaseFunction test_authorization_denied>

    def run(self, result=None):
        if result is None:
            result = self.defaultTestResult()
            startTestRun = getattr(result, 'startTestRun', None)
            stopTestRun = getattr(result, 'stopTestRun', None)
            if startTestRun is not None:
                startTestRun()
        else:
            stopTestRun = None
    
        result.startTest(self)
        try:
            testMethod = getattr(self, self._testMethodName)
            if (getattr(self.__class__, "__unittest_skip__", False) or
                getattr(testMethod, "__unittest_skip__", False)):
                # If the class or method was skipped.
                skip_why = (getattr(self.__class__, '__unittest_skip_why__', '')
                            or getattr(testMethod, '__unittest_skip_why__', ''))
                _addSkip(result, self, skip_why)
                return result
    
            expecting_failure = (
                getattr(self, "__unittest_expecting_failure__", False) or
                getattr(testMethod, "__unittest_expecting_failure__", False)
            )
            outcome = _Outcome(result)
            start_time = time.perf_counter()
            try:
                self._outcome = outcome
    
                with outcome.testPartExecutor(self):
>                   self._callSetUp()

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:665: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc_implicit.TestProviderOAuth2OIDCImplicit testMethod=test_authorization_denied>

    def _callSetUp(self):
>       self.setUp()

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:612: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc_implicit.TestProviderOAuth2OIDCImplicit testMethod=test_authorization_denied>

    def setUp(self):
        self.client_id = generate_id()
        self.client_secret = generate_key()
        self.application_slug = "test"
>       super().setUp()

tests/e2e/test_provider_oidc_implicit.py:39: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc_implicit.TestProviderOAuth2OIDCImplicit testMethod=test_authorization_denied>

    def setUp(self):
        if IS_CI:
            print("::group::authentik Logs", file=stderr)
        apps.get_app_config("authentik_tenants").ready()
        self.wait_timeout = 60
        self.logger = get_logger()
>       self.driver = self._get_driver()
                      ^^^^^^^^^^^^^^^^^^

tests/e2e/utils.py:80: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc_implicit.TestProviderOAuth2OIDCImplicit testMethod=test_authorization_denied>

    def _get_driver(self) -> WebDriver:
        count = 0
        opts = webdriver.ChromeOptions()
        opts.accept_insecure_certs = True
        opts.add_argument("--disable-search-engine-choice-screen")
        opts.add_extension(self._get_chrome_extension())
        # This breaks selenium when running remotely...?
        # opts.set_capability("goog:loggingPrefs", {"browser": "ALL"})
        opts.add_experimental_option(
            "prefs",
            {
                "profile.password_manager_leak_detection": False,
            },
        )
        while count < RETRIES:
            try:
                driver = webdriver.Remote(
                    command_executor="http://localhost:4444/wd/hub",
                    options=opts,
                )
                driver.maximize_window()
                return driver
            except WebDriverException as exc:
                self.logger.warning("Failed to setup webdriver", exc=exc)
                count += 1
>       raise ValueError(f"Webdriver failed after {RETRIES}.")
E       ValueError: Webdriver failed after 3.

tests/e2e/utils.py:111: ValueError

tests.e2e.test_provider_oidc.TestProviderOAuth2OIDC::test_authorization_denied

Stack Traces | 3.61s run time

self = <unittest.case._Outcome object at 0x7f3b9e794050>
test_case = <tests.e2e.test_provider_oidc.TestProviderOAuth2OIDC testMethod=test_authorization_denied>
subTest = False

    @contextlib.contextmanager
    def testPartExecutor(self, test_case, subTest=False):
        old_success = self.success
        self.success = True
        try:
>           yield

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:58: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc.TestProviderOAuth2OIDC testMethod=test_authorization_denied>
result = <TestCaseFunction test_authorization_denied>

    def run(self, result=None):
        if result is None:
            result = self.defaultTestResult()
            startTestRun = getattr(result, 'startTestRun', None)
            stopTestRun = getattr(result, 'stopTestRun', None)
            if startTestRun is not None:
                startTestRun()
        else:
            stopTestRun = None
    
        result.startTest(self)
        try:
            testMethod = getattr(self, self._testMethodName)
            if (getattr(self.__class__, "__unittest_skip__", False) or
                getattr(testMethod, "__unittest_skip__", False)):
                # If the class or method was skipped.
                skip_why = (getattr(self.__class__, '__unittest_skip_why__', '')
                            or getattr(testMethod, '__unittest_skip_why__', ''))
                _addSkip(result, self, skip_why)
                return result
    
            expecting_failure = (
                getattr(self, "__unittest_expecting_failure__", False) or
                getattr(testMethod, "__unittest_expecting_failure__", False)
            )
            outcome = _Outcome(result)
            start_time = time.perf_counter()
            try:
                self._outcome = outcome
    
                with outcome.testPartExecutor(self):
>                   self._callSetUp()

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:665: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc.TestProviderOAuth2OIDC testMethod=test_authorization_denied>

    def _callSetUp(self):
>       self.setUp()

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:612: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc.TestProviderOAuth2OIDC testMethod=test_authorization_denied>

    def setUp(self):
        self.client_id = generate_id()
        self.client_secret = generate_key()
        self.application_slug = generate_id()
>       super().setUp()

tests/e2e/test_provider_oidc.py:39: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc.TestProviderOAuth2OIDC testMethod=test_authorization_denied>

    def setUp(self):
        if IS_CI:
            print("::group::authentik Logs", file=stderr)
        apps.get_app_config("authentik_tenants").ready()
        self.wait_timeout = 60
        self.logger = get_logger()
>       self.driver = self._get_driver()
                      ^^^^^^^^^^^^^^^^^^

tests/e2e/utils.py:80: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc.TestProviderOAuth2OIDC testMethod=test_authorization_denied>

    def _get_driver(self) -> WebDriver:
        count = 0
        opts = webdriver.ChromeOptions()
        opts.accept_insecure_certs = True
        opts.add_argument("--disable-search-engine-choice-screen")
        opts.add_extension(self._get_chrome_extension())
        # This breaks selenium when running remotely...?
        # opts.set_capability("goog:loggingPrefs", {"browser": "ALL"})
        opts.add_experimental_option(
            "prefs",
            {
                "profile.password_manager_leak_detection": False,
            },
        )
        while count < RETRIES:
            try:
                driver = webdriver.Remote(
                    command_executor="http://localhost:4444/wd/hub",
                    options=opts,
                )
                driver.maximize_window()
                return driver
            except WebDriverException as exc:
                self.logger.warning("Failed to setup webdriver", exc=exc)
                count += 1
>       raise ValueError(f"Webdriver failed after {RETRIES}.")
E       ValueError: Webdriver failed after 3.

tests/e2e/utils.py:111: ValueError

tests.e2e.test_provider_oidc.TestProviderOAuth2OIDC::test_authorization_consent_implied

Stack Traces | 3.82s run time

self = <unittest.case._Outcome object at 0x7f3b9e7f5370>
test_case = <tests.e2e.test_provider_oidc.TestProviderOAuth2OIDC testMethod=test_authorization_consent_implied>
subTest = False

    @contextlib.contextmanager
    def testPartExecutor(self, test_case, subTest=False):
        old_success = self.success
        self.success = True
        try:
>           yield

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:58: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc.TestProviderOAuth2OIDC testMethod=test_authorization_consent_implied>
result = <TestCaseFunction test_authorization_consent_implied>

    def run(self, result=None):
        if result is None:
            result = self.defaultTestResult()
            startTestRun = getattr(result, 'startTestRun', None)
            stopTestRun = getattr(result, 'stopTestRun', None)
            if startTestRun is not None:
                startTestRun()
        else:
            stopTestRun = None
    
        result.startTest(self)
        try:
            testMethod = getattr(self, self._testMethodName)
            if (getattr(self.__class__, "__unittest_skip__", False) or
                getattr(testMethod, "__unittest_skip__", False)):
                # If the class or method was skipped.
                skip_why = (getattr(self.__class__, '__unittest_skip_why__', '')
                            or getattr(testMethod, '__unittest_skip_why__', ''))
                _addSkip(result, self, skip_why)
                return result
    
            expecting_failure = (
                getattr(self, "__unittest_expecting_failure__", False) or
                getattr(testMethod, "__unittest_expecting_failure__", False)
            )
            outcome = _Outcome(result)
            start_time = time.perf_counter()
            try:
                self._outcome = outcome
    
                with outcome.testPartExecutor(self):
>                   self._callSetUp()

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:665: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc.TestProviderOAuth2OIDC testMethod=test_authorization_consent_implied>

    def _callSetUp(self):
>       self.setUp()

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:612: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc.TestProviderOAuth2OIDC testMethod=test_authorization_consent_implied>

    def setUp(self):
        self.client_id = generate_id()
        self.client_secret = generate_key()
        self.application_slug = generate_id()
>       super().setUp()

tests/e2e/test_provider_oidc.py:39: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc.TestProviderOAuth2OIDC testMethod=test_authorization_consent_implied>

    def setUp(self):
        if IS_CI:
            print("::group::authentik Logs", file=stderr)
        apps.get_app_config("authentik_tenants").ready()
        self.wait_timeout = 60
        self.logger = get_logger()
>       self.driver = self._get_driver()
                      ^^^^^^^^^^^^^^^^^^

tests/e2e/utils.py:80: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc.TestProviderOAuth2OIDC testMethod=test_authorization_consent_implied>

    def _get_driver(self) -> WebDriver:
        count = 0
        opts = webdriver.ChromeOptions()
        opts.accept_insecure_certs = True
        opts.add_argument("--disable-search-engine-choice-screen")
        opts.add_extension(self._get_chrome_extension())
        # This breaks selenium when running remotely...?
        # opts.set_capability("goog:loggingPrefs", {"browser": "ALL"})
        opts.add_experimental_option(
            "prefs",
            {
                "profile.password_manager_leak_detection": False,
            },
        )
        while count < RETRIES:
            try:
                driver = webdriver.Remote(
                    command_executor="http://localhost:4444/wd/hub",
                    options=opts,
                )
                driver.maximize_window()
                return driver
            except WebDriverException as exc:
                self.logger.warning("Failed to setup webdriver", exc=exc)
                count += 1
>       raise ValueError(f"Webdriver failed after {RETRIES}.")
E       ValueError: Webdriver failed after 3.

tests/e2e/utils.py:111: ValueError

tests.e2e.test_provider_oidc.TestProviderOAuth2OIDC::test_authorization_consent_explicit

Stack Traces | 4.17s run time

self = <unittest.case._Outcome object at 0x7f3b9d487e30>
test_case = <tests.e2e.test_provider_oidc.TestProviderOAuth2OIDC testMethod=test_authorization_consent_explicit>
subTest = False

    @contextlib.contextmanager
    def testPartExecutor(self, test_case, subTest=False):
        old_success = self.success
        self.success = True
        try:
>           yield

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:58: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc.TestProviderOAuth2OIDC testMethod=test_authorization_consent_explicit>
result = <TestCaseFunction test_authorization_consent_explicit>

    def run(self, result=None):
        if result is None:
            result = self.defaultTestResult()
            startTestRun = getattr(result, 'startTestRun', None)
            stopTestRun = getattr(result, 'stopTestRun', None)
            if startTestRun is not None:
                startTestRun()
        else:
            stopTestRun = None
    
        result.startTest(self)
        try:
            testMethod = getattr(self, self._testMethodName)
            if (getattr(self.__class__, "__unittest_skip__", False) or
                getattr(testMethod, "__unittest_skip__", False)):
                # If the class or method was skipped.
                skip_why = (getattr(self.__class__, '__unittest_skip_why__', '')
                            or getattr(testMethod, '__unittest_skip_why__', ''))
                _addSkip(result, self, skip_why)
                return result
    
            expecting_failure = (
                getattr(self, "__unittest_expecting_failure__", False) or
                getattr(testMethod, "__unittest_expecting_failure__", False)
            )
            outcome = _Outcome(result)
            start_time = time.perf_counter()
            try:
                self._outcome = outcome
    
                with outcome.testPartExecutor(self):
>                   self._callSetUp()

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:665: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc.TestProviderOAuth2OIDC testMethod=test_authorization_consent_explicit>

    def _callSetUp(self):
>       self.setUp()

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:612: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc.TestProviderOAuth2OIDC testMethod=test_authorization_consent_explicit>

    def setUp(self):
        self.client_id = generate_id()
        self.client_secret = generate_key()
        self.application_slug = generate_id()
>       super().setUp()

tests/e2e/test_provider_oidc.py:39: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc.TestProviderOAuth2OIDC testMethod=test_authorization_consent_explicit>

    def setUp(self):
        if IS_CI:
            print("::group::authentik Logs", file=stderr)
        apps.get_app_config("authentik_tenants").ready()
        self.wait_timeout = 60
        self.logger = get_logger()
>       self.driver = self._get_driver()
                      ^^^^^^^^^^^^^^^^^^

tests/e2e/utils.py:80: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc.TestProviderOAuth2OIDC testMethod=test_authorization_consent_explicit>

    def _get_driver(self) -> WebDriver:
        count = 0
        opts = webdriver.ChromeOptions()
        opts.accept_insecure_certs = True
        opts.add_argument("--disable-search-engine-choice-screen")
        opts.add_extension(self._get_chrome_extension())
        # This breaks selenium when running remotely...?
        # opts.set_capability("goog:loggingPrefs", {"browser": "ALL"})
        opts.add_experimental_option(
            "prefs",
            {
                "profile.password_manager_leak_detection": False,
            },
        )
        while count < RETRIES:
            try:
                driver = webdriver.Remote(
                    command_executor="http://localhost:4444/wd/hub",
                    options=opts,
                )
                driver.maximize_window()
                return driver
            except WebDriverException as exc:
                self.logger.warning("Failed to setup webdriver", exc=exc)
                count += 1
>       raise ValueError(f"Webdriver failed after {RETRIES}.")
E       ValueError: Webdriver failed after 3.

tests/e2e/utils.py:111: ValueError

tests.e2e.test_provider_oidc.TestProviderOAuth2OIDC::test_redirect_uri_error

Stack Traces | 4.37s run time

self = <unittest.case._Outcome object at 0x7f3b9e7f5ae0>
test_case = <tests.e2e.test_provider_oidc.TestProviderOAuth2OIDC testMethod=test_redirect_uri_error>
subTest = False

    @contextlib.contextmanager
    def testPartExecutor(self, test_case, subTest=False):
        old_success = self.success
        self.success = True
        try:
>           yield

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:58: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc.TestProviderOAuth2OIDC testMethod=test_redirect_uri_error>
result = <TestCaseFunction test_redirect_uri_error>

    def run(self, result=None):
        if result is None:
            result = self.defaultTestResult()
            startTestRun = getattr(result, 'startTestRun', None)
            stopTestRun = getattr(result, 'stopTestRun', None)
            if startTestRun is not None:
                startTestRun()
        else:
            stopTestRun = None
    
        result.startTest(self)
        try:
            testMethod = getattr(self, self._testMethodName)
            if (getattr(self.__class__, "__unittest_skip__", False) or
                getattr(testMethod, "__unittest_skip__", False)):
                # If the class or method was skipped.
                skip_why = (getattr(self.__class__, '__unittest_skip_why__', '')
                            or getattr(testMethod, '__unittest_skip_why__', ''))
                _addSkip(result, self, skip_why)
                return result
    
            expecting_failure = (
                getattr(self, "__unittest_expecting_failure__", False) or
                getattr(testMethod, "__unittest_expecting_failure__", False)
            )
            outcome = _Outcome(result)
            start_time = time.perf_counter()
            try:
                self._outcome = outcome
    
                with outcome.testPartExecutor(self):
>                   self._callSetUp()

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:665: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc.TestProviderOAuth2OIDC testMethod=test_redirect_uri_error>

    def _callSetUp(self):
>       self.setUp()

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:612: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc.TestProviderOAuth2OIDC testMethod=test_redirect_uri_error>

    def setUp(self):
        self.client_id = generate_id()
        self.client_secret = generate_key()
        self.application_slug = generate_id()
>       super().setUp()

tests/e2e/test_provider_oidc.py:39: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc.TestProviderOAuth2OIDC testMethod=test_redirect_uri_error>

    def setUp(self):
        if IS_CI:
            print("::group::authentik Logs", file=stderr)
        apps.get_app_config("authentik_tenants").ready()
        self.wait_timeout = 60
        self.logger = get_logger()
>       self.driver = self._get_driver()
                      ^^^^^^^^^^^^^^^^^^

tests/e2e/utils.py:80: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc.TestProviderOAuth2OIDC testMethod=test_redirect_uri_error>

    def _get_driver(self) -> WebDriver:
        count = 0
        opts = webdriver.ChromeOptions()
        opts.accept_insecure_certs = True
        opts.add_argument("--disable-search-engine-choice-screen")
        opts.add_extension(self._get_chrome_extension())
        # This breaks selenium when running remotely...?
        # opts.set_capability("goog:loggingPrefs", {"browser": "ALL"})
        opts.add_experimental_option(
            "prefs",
            {
                "profile.password_manager_leak_detection": False,
            },
        )
        while count < RETRIES:
            try:
                driver = webdriver.Remote(
                    command_executor="http://localhost:4444/wd/hub",
                    options=opts,
                )
                driver.maximize_window()
                return driver
            except WebDriverException as exc:
                self.logger.warning("Failed to setup webdriver", exc=exc)
                count += 1
>       raise ValueError(f"Webdriver failed after {RETRIES}.")
E       ValueError: Webdriver failed after 3.

tests/e2e/utils.py:111: ValueError

tests.e2e.test_provider_oidc_implicit.TestProviderOAuth2OIDCImplicit::test_authorization_consent_explicit

Stack Traces | 40.7s run time

self = <unittest.case._Outcome object at 0x7f3b9b271480>
test_case = <tests.e2e.test_provider_oidc_implicit.TestProviderOAuth2OIDCImplicit testMethod=test_authorization_consent_explicit>
subTest = False

    @contextlib.contextmanager
    def testPartExecutor(self, test_case, subTest=False):
        old_success = self.success
        self.success = True
        try:
>           yield

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:58: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc_implicit.TestProviderOAuth2OIDCImplicit testMethod=test_authorization_consent_explicit>
result = <TestCaseFunction test_authorization_consent_explicit>

    def run(self, result=None):
        if result is None:
            result = self.defaultTestResult()
            startTestRun = getattr(result, 'startTestRun', None)
            stopTestRun = getattr(result, 'stopTestRun', None)
            if startTestRun is not None:
                startTestRun()
        else:
            stopTestRun = None
    
        result.startTest(self)
        try:
            testMethod = getattr(self, self._testMethodName)
            if (getattr(self.__class__, "__unittest_skip__", False) or
                getattr(testMethod, "__unittest_skip__", False)):
                # If the class or method was skipped.
                skip_why = (getattr(self.__class__, '__unittest_skip_why__', '')
                            or getattr(testMethod, '__unittest_skip_why__', ''))
                _addSkip(result, self, skip_why)
                return result
    
            expecting_failure = (
                getattr(self, "__unittest_expecting_failure__", False) or
                getattr(testMethod, "__unittest_expecting_failure__", False)
            )
            outcome = _Outcome(result)
            start_time = time.perf_counter()
            try:
                self._outcome = outcome
    
                with outcome.testPartExecutor(self):
>                   self._callSetUp()

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:665: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc_implicit.TestProviderOAuth2OIDCImplicit testMethod=test_authorization_consent_explicit>

    def _callSetUp(self):
>       self.setUp()

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:612: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc_implicit.TestProviderOAuth2OIDCImplicit testMethod=test_authorization_consent_explicit>

    def setUp(self):
        self.client_id = generate_id()
        self.client_secret = generate_key()
        self.application_slug = "test"
>       super().setUp()

tests/e2e/test_provider_oidc_implicit.py:39: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc_implicit.TestProviderOAuth2OIDCImplicit testMethod=test_authorization_consent_explicit>

    def setUp(self):
        if IS_CI:
            print("::group::authentik Logs", file=stderr)
        apps.get_app_config("authentik_tenants").ready()
        self.wait_timeout = 60
        self.logger = get_logger()
>       self.driver = self._get_driver()
                      ^^^^^^^^^^^^^^^^^^

tests/e2e/utils.py:80: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc_implicit.TestProviderOAuth2OIDCImplicit testMethod=test_authorization_consent_explicit>

    def _get_driver(self) -> WebDriver:
        count = 0
        opts = webdriver.ChromeOptions()
        opts.accept_insecure_certs = True
        opts.add_argument("--disable-search-engine-choice-screen")
        opts.add_extension(self._get_chrome_extension())
        # This breaks selenium when running remotely...?
        # opts.set_capability("goog:loggingPrefs", {"browser": "ALL"})
        opts.add_experimental_option(
            "prefs",
            {
                "profile.password_manager_leak_detection": False,
            },
        )
        while count < RETRIES:
            try:
                driver = webdriver.Remote(
                    command_executor="http://localhost:4444/wd/hub",
                    options=opts,
                )
                driver.maximize_window()
                return driver
            except WebDriverException as exc:
                self.logger.warning("Failed to setup webdriver", exc=exc)
                count += 1
>       raise ValueError(f"Webdriver failed after {RETRIES}.")
E       ValueError: Webdriver failed after 3.

tests/e2e/utils.py:111: ValueError

tests.e2e.test_provider_oidc_implicit.TestProviderOAuth2OIDCImplicit::test_authorization_consent_implied

Stack Traces | 102s run time

self = <unittest.case._Outcome object at 0x7f3ba41b2270>
test_case = <tests.e2e.test_provider_oidc_implicit.TestProviderOAuth2OIDCImplicit testMethod=test_authorization_consent_implied>
subTest = False

    @contextlib.contextmanager
    def testPartExecutor(self, test_case, subTest=False):
        old_success = self.success
        self.success = True
        try:
>           yield

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:58: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc_implicit.TestProviderOAuth2OIDCImplicit testMethod=test_authorization_consent_implied>
result = <TestCaseFunction test_authorization_consent_implied>

    def run(self, result=None):
        if result is None:
            result = self.defaultTestResult()
            startTestRun = getattr(result, 'startTestRun', None)
            stopTestRun = getattr(result, 'stopTestRun', None)
            if startTestRun is not None:
                startTestRun()
        else:
            stopTestRun = None
    
        result.startTest(self)
        try:
            testMethod = getattr(self, self._testMethodName)
            if (getattr(self.__class__, "__unittest_skip__", False) or
                getattr(testMethod, "__unittest_skip__", False)):
                # If the class or method was skipped.
                skip_why = (getattr(self.__class__, '__unittest_skip_why__', '')
                            or getattr(testMethod, '__unittest_skip_why__', ''))
                _addSkip(result, self, skip_why)
                return result
    
            expecting_failure = (
                getattr(self, "__unittest_expecting_failure__", False) or
                getattr(testMethod, "__unittest_expecting_failure__", False)
            )
            outcome = _Outcome(result)
            start_time = time.perf_counter()
            try:
                self._outcome = outcome
    
                with outcome.testPartExecutor(self):
>                   self._callSetUp()

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:665: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc_implicit.TestProviderOAuth2OIDCImplicit testMethod=test_authorization_consent_implied>

    def _callSetUp(self):
>       self.setUp()

.../hostedtoolcache/Python/3.14.3........./x64/lib/python3.14/unittest/case.py:612: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc_implicit.TestProviderOAuth2OIDCImplicit testMethod=test_authorization_consent_implied>

    def setUp(self):
        self.client_id = generate_id()
        self.client_secret = generate_key()
        self.application_slug = "test"
>       super().setUp()

tests/e2e/test_provider_oidc_implicit.py:39: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc_implicit.TestProviderOAuth2OIDCImplicit testMethod=test_authorization_consent_implied>

    def setUp(self):
        if IS_CI:
            print("::group::authentik Logs", file=stderr)
        apps.get_app_config("authentik_tenants").ready()
        self.wait_timeout = 60
        self.logger = get_logger()
>       self.driver = self._get_driver()
                      ^^^^^^^^^^^^^^^^^^

tests/e2e/utils.py:80: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <tests.e2e.test_provider_oidc_implicit.TestProviderOAuth2OIDCImplicit testMethod=test_authorization_consent_implied>

    def _get_driver(self) -> WebDriver:
        count = 0
        opts = webdriver.ChromeOptions()
        opts.accept_insecure_certs = True
        opts.add_argument("--disable-search-engine-choice-screen")
        opts.add_extension(self._get_chrome_extension())
        # This breaks selenium when running remotely...?
        # opts.set_capability("goog:loggingPrefs", {"browser": "ALL"})
        opts.add_experimental_option(
            "prefs",
            {
                "profile.password_manager_leak_detection": False,
            },
        )
        while count < RETRIES:
            try:
                driver = webdriver.Remote(
                    command_executor="http://localhost:4444/wd/hub",
                    options=opts,
                )
                driver.maximize_window()
                return driver
            except WebDriverException as exc:
                self.logger.warning("Failed to setup webdriver", exc=exc)
                count += 1
>       raise ValueError(f"Webdriver failed after {RETRIES}.")
E       ValueError: Webdriver failed after 3.

tests/e2e/utils.py:111: ValueError

To view more test analytics, go to the Test Analytics Dashboard
_{📋 Got 3 mins? Take this short survey to help us improve Test Analytics.}

[netlify]

✅ Deploy Preview for authentik-integrations ready!

Name	Link
🔨 Latest commit	f93e7a7
🔍 Latest deploy log	https://app.netlify.com/projects/authentik-integrations/deploys/69c4234aeeb53f00099966bc
😎 Deploy Preview	https://deploy-preview-21018--authentik-integrations.netlify.app
📱 Preview on mobile	Toggle QR Code... Use your smartphone camera to open QR code link.

---

To edit notification comments on pull requests, go to your Netlify project configuration.

[netlify]

✅ Deploy Preview for authentik-storybook ready!

Name	Link
🔨 Latest commit	f93e7a7
🔍 Latest deploy log	https://app.netlify.com/projects/authentik-storybook/deploys/69c4234a347ef40008f46ff4
😎 Deploy Preview	https://deploy-preview-21018--authentik-storybook.netlify.app
📱 Preview on mobile	Toggle QR Code... Use your smartphone camera to open QR code link.

---

To edit notification comments on pull requests, go to your Netlify project configuration.

[tanberry]

A couple of tweaks....

[dominic-r]

can we also update https://next.goauthentik.io/core/architecture/ ?

[rissson]

can we also update next.goauthentik.io/core/architecture ?

it's technically still correct