flows: return correct status code on error

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

.github/actions/setup/action.yml

@@ -21,7 +21,7 @@ runs:
         sudo apt-get install --no-install-recommends -y libpq-dev openssl libxmlsec1-dev pkg-config gettext libkrb5-dev krb5-kdc krb5-user krb5-admin-server
     - name: Install uv
       if: ${{ contains(inputs.dependencies, 'python') }}
-      uses: astral-sh/setup-uv@ed21f2f24f8dd64503750218de024bcf64c7250a # v5
+      uses: astral-sh/setup-uv@681c641aba71e4a1c380be3ab5e12ad51f415867 # v5
       with:
         enable-cache: true
     - name: Setup python

.github/actions/test-results/action.yml

@@ -8,15 +8,15 @@ inputs:
 runs:
   using: "composite"
   steps:
-    - uses: codecov/codecov-action@5a1091511ad55cbe89839c7260b706298ca349f7 # v5
+    - uses: codecov/codecov-action@671740ac38dd9b0130fbe1cec585b89eea48d3de # v5
       with:
         flags: ${{ inputs.flags }}
         use_oidc: true
-    - uses: codecov/test-results-action@47f89e9acb64b76debcd5ea40642d25a4adced9f # v1
+    - uses: codecov/codecov-action@671740ac38dd9b0130fbe1cec585b89eea48d3de # v5
       with:
         flags: ${{ inputs.flags }}
-        file: unittest.xml
         use_oidc: true
+        report_type: test_results
     - name: PostgreSQL Logs
       shell: bash
       run: |

.github/pull_request_template.md

@@ -2,6 +2,10 @@
 👋 Hi there! Welcome.
 
 Please check the Contributing guidelines: https://docs.goauthentik.io/docs/developer-docs/#how-can-i-contribute
+
+⚠️ IMPORTANT: Make sure you are opening this PR from a FEATURE BRANCH, not from your main branch!
+If you opened this PR from your main branch, please close it and create a new feature branch instead.
+For more information, see: https://docs.goauthentik.io/developer-docs/contributing/#always-use-feature-branches
 -->
 
 ## Details

.github/workflows/_reusable-docker-build-single.yml

@@ -73,14 +73,12 @@ jobs:
           mkdir -p ./gen-ts-api
           mkdir -p ./gen-go-api
       - name: Setup node
-        if: ${{ !inputs.release }}
         uses: actions/setup-node@395ad3262231945c25e8478fd5baf05154b1d79f # v5
         with:
           node-version-file: web/package.json
           cache: "npm"
           cache-dependency-path: web/package-lock.json
       - name: generate ts client
-        if: ${{ !inputs.release }}
         run: make gen-client-ts
       - name: Build Docker Image
         uses: docker/build-push-action@263435318d21b8e681c14492fe198d362a7d2c83 # v6

.github/workflows/api-ts-publish.yml

@@ -46,7 +46,7 @@ jobs:
         run: |
           export VERSION=`node -e 'console.log(require("../gen-ts-api/package.json").version)'`
           npm i @goauthentik/api@$VERSION
-      - uses: peter-evans/create-pull-request@22a9089034f40e5a961c8808d113e2c98fb63676 # v7
+      - uses: peter-evans/create-pull-request@98357b18bf14b5342f975ff684046ec3b2a07725 # v7
         id: cpr
         with:
           token: ${{ steps.generate_token.outputs.token }}

.github/workflows/ci-api-docs.yml

@@ -41,7 +41,7 @@ jobs:
       - working-directory: website/
         name: Install Dependencies
         run: npm ci
-      - uses: actions/cache@0057852bfaa89a56745cba8c7296529d2fc39830 # v4
+      - uses: actions/cache@9255dc7a253b0ccc959486e2bca901246202afeb # v4
         with:
           path: |
             ${{ github.workspace }}/website/api/.docusaurus
@@ -55,7 +55,7 @@ jobs:
         env:
           NODE_ENV: production
         run: npm run build -w api
-      - uses: actions/upload-artifact@330a01c490aca151604b8cf639adc76d48f6c5d4 # v4
+      - uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v4
         with:
           name: api-docs
           path: website/api/build
@@ -67,7 +67,7 @@ jobs:
       - build
     steps:
       - uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v5
-      - uses: actions/download-artifact@018cc2cf5baa6db3ef3c5f8a56943fffe632ef53 # v5
+      - uses: actions/download-artifact@37930b1c2abaa49bbe596cd826c3c89aef350131 # v5
         with:
           name: api-docs
           path: website/api/build

.github/workflows/ci-main.yml

@@ -201,7 +201,7 @@ jobs:
         run: |
           docker compose -f tests/e2e/docker-compose.yml up -d --quiet-pull
       - id: cache-web
-        uses: actions/cache@0057852bfaa89a56745cba8c7296529d2fc39830 # v4
+        uses: actions/cache@9255dc7a253b0ccc959486e2bca901246202afeb # v4
         with:
           path: web/dist
           key: ${{ runner.os }}-web-${{ hashFiles('web/package-lock.json', 'package-lock.json', 'web/src/**', 'web/packages/sfe/src/**') }}-b

.github/workflows/gen-image-compress.yml

@@ -42,7 +42,7 @@ jobs:
         with:
           GITHUB_TOKEN: ${{ steps.generate_token.outputs.token }}
           compressOnly: ${{ github.event_name != 'pull_request' }}
-      - uses: peter-evans/create-pull-request@22a9089034f40e5a961c8808d113e2c98fb63676 # v7
+      - uses: peter-evans/create-pull-request@98357b18bf14b5342f975ff684046ec3b2a07725 # v7
         if: "${{ github.event_name != 'pull_request' && steps.compress.outputs.markdown != '' }}"
         id: cpr
         with:

.github/workflows/gen-update-webauthn-mds.yml

@@ -26,7 +26,7 @@ jobs:
       - name: Setup authentik env
         uses: ./.github/actions/setup
       - run: uv run ak update_webauthn_mds
-      - uses: peter-evans/create-pull-request@22a9089034f40e5a961c8808d113e2c98fb63676 # v7
+      - uses: peter-evans/create-pull-request@98357b18bf14b5342f975ff684046ec3b2a07725 # v7
         id: cpr
         with:
           token: ${{ steps.generate_token.outputs.token }}

.github/workflows/packages-npm-publish.yml

@@ -40,7 +40,7 @@ jobs:
           registry-url: "https://registry.npmjs.org"
       - name: Get changed files
         id: changed-files
-        uses: tj-actions/changed-files@24d32ffd492484c1d75e0c0b894501ddb9d30d62 # 24d32ffd492484c1d75e0c0b894501ddb9d30d62
+        uses: tj-actions/changed-files@e0021407031f5be11a464abee9a0776171c79891 # 24d32ffd492484c1d75e0c0b894501ddb9d30d62
         with:
           files: |
             ${{ matrix.package }}/package.json

.github/workflows/release-branch-off.yml

@@ -73,7 +73,7 @@ jobs:
       - name: Bump version
         run: "make bump version=${{ inputs.next_version }}.0-rc1"
       - name: Create pull request
-        uses: peter-evans/create-pull-request@22a9089034f40e5a961c8808d113e2c98fb63676 # v7
+        uses: peter-evans/create-pull-request@98357b18bf14b5342f975ff684046ec3b2a07725 # v7
         with:
           token: ${{ steps.generate_token.outputs.token }}
           branch: release-bump-${{ inputs.next_version }}

.github/workflows/release-tag.yml

@@ -49,8 +49,12 @@ jobs:
   test:
     name: Pre-release test
     runs-on: ubuntu-latest
+    needs:
+      - check-inputs
     steps:
       - uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v5
+        with:
+          ref: "version-${{ needs.check-inputs.outputs.major_version }}"
       - run: make test-docker
   bump-authentik:
     name: Bump authentik version
@@ -130,7 +134,7 @@ jobs:
           sed -E -i 's/[0-9]{4}\.[0-9]{1,2}\.[0-9]+$/${{ inputs.version }}/' charts/authentik/Chart.yaml
           ./scripts/helm-docs.sh
       - name: Create pull request
-        uses: peter-evans/create-pull-request@22a9089034f40e5a961c8808d113e2c98fb63676 # v7
+        uses: peter-evans/create-pull-request@98357b18bf14b5342f975ff684046ec3b2a07725 # v7
         with:
           token: "${{ steps.app-token.outputs.token }}"
           branch: bump-${{ inputs.version }}
@@ -185,7 +189,7 @@ jobs:
             '.stable.version = $version | .stable.changelog = $changelog | .stable.changelog_url = $changelog_url' version.json > version.new.json
           mv version.new.json version.json
       - name: Create pull request
-        uses: peter-evans/create-pull-request@22a9089034f40e5a961c8808d113e2c98fb63676 # v7
+        uses: peter-evans/create-pull-request@98357b18bf14b5342f975ff684046ec3b2a07725 # v7
         with:
           token: "${{ steps.app-token.outputs.token }}"
           branch: bump-${{ inputs.version }}

.github/workflows/translation-extract-compile.yml

@@ -44,7 +44,7 @@ jobs:
           make web-check-compile
       - name: Create Pull Request
         if: ${{ github.event_name != 'pull_request' }}
-        uses: peter-evans/create-pull-request@22a9089034f40e5a961c8808d113e2c98fb63676 # v7
+        uses: peter-evans/create-pull-request@98357b18bf14b5342f975ff684046ec3b2a07725 # v7
         with:
           token: ${{ steps.generate_token.outputs.token }}
           branch: extract-compile-backend-translation

CODEOWNERS

@@ -28,8 +28,10 @@ packages/django-channels-postgres       @goauthentik/backend
 packages/django-postgres-cache          @goauthentik/backend
 packages/django-dramatiq-postgres       @goauthentik/backend
 # Web packages
-packages/package.json                   @goauthentik/backend @goauthentik/frontend
-packages/package-lock.json              @goauthentik/backend @goauthentik/frontend
+package.json                            @goauthentik/frontend
+package-lock.json                       @goauthentik/frontend
+packages/package.json                   @goauthentik/frontend
+packages/package-lock.json              @goauthentik/frontend
 packages/docusaurus-config              @goauthentik/frontend
 packages/esbuild-plugin-live-reload     @goauthentik/frontend
 packages/eslint-config                  @goauthentik/frontend

Dockerfile

@@ -26,7 +26,7 @@ RUN npm run build && \
     npm run build:sfe
 
 # Stage 2: Build go proxy
-FROM --platform=${BUILDPLATFORM} docker.io/library/golang:1.25.5-trixie@sha256:4f9d98ebaa759f776496d850e0439c48948d587b191fc3949b5f5e4667abef90 AS go-builder
+FROM --platform=${BUILDPLATFORM} docker.io/library/golang:1.25.5-trixie@sha256:8e8f9c84609b6005af0a4a8227cee53d6226aab1c6dcb22daf5aeeb8b05480e1 AS go-builder
 
 ARG TARGETOS
 ARG TARGETARCH
@@ -76,7 +76,7 @@ RUN --mount=type=secret,id=GEOIPUPDATE_ACCOUNT_ID \
     /bin/sh -c "GEOIPUPDATE_LICENSE_KEY_FILE=/run/secrets/GEOIPUPDATE_LICENSE_KEY /usr/bin/entry.sh || echo 'Failed to get GeoIP database, disabling'; exit 0"
 
 # Stage 4: Download uv
-FROM ghcr.io/astral-sh/uv:0.9.16@sha256:ae9ff79d095a61faf534a882ad6378e8159d2ce322691153d68d2afac7422840 AS uv
+FROM ghcr.io/astral-sh/uv:0.9.18@sha256:5713fa8217f92b80223bc83aac7db36ec80a84437dbc0d04bbc659cae030d8c9 AS uv
 # Stage 5: Base python image
 FROM ghcr.io/goauthentik/fips-python:3.13.9-slim-trixie-fips@sha256:700fc8c1e290bd14e5eaca50b1d8e8c748c820010559cbfb4c4f8dfbe2c4c9ff AS python-base
 

Makefile

@@ -9,6 +9,13 @@ NPM_VERSION = $(shell python -m scripts.generate_semver)
 PY_SOURCES = authentik packages tests scripts lifecycle .github
 DOCKER_IMAGE ?= "authentik:test"
 
+UNAME_S := $(shell uname -s)
+ifeq ($(UNAME_S),Darwin)
+	SED_INPLACE = sed -i ''
+else
+	SED_INPLACE = sed -i
+endif
+
 GEN_API_TS = gen-ts-api
 GEN_API_PY = gen-py-api
 GEN_API_GO = gen-go-api
@@ -119,8 +126,8 @@ bump:  ## Bump authentik version. Usage: make bump version=20xx.xx.xx
 ifndef version
 	$(error Usage: make bump version=20xx.xx.xx )
 endif
-	sed -i 's/^version = ".*"/version = "$(version)"/' pyproject.toml
-	sed -i 's/^VERSION = ".*"/VERSION = "$(version)"/' authentik/__init__.py
+	$(SED_INPLACE) 's/^version = ".*"/version = "$(version)"/' pyproject.toml
+	$(SED_INPLACE) 's/^VERSION = ".*"/VERSION = "$(version)"/' authentik/__init__.py
 	$(MAKE) gen-build gen-compose aws-cfn
 	npm version --no-git-tag-version --allow-same-version $(version)
 	cd ${PWD}/web && npm version --no-git-tag-version --allow-same-version $(version)
@@ -155,8 +162,8 @@ gen-diff:  ## (Release) generate the changelog diff between the current schema a
 		/local/schema-old.yml \
 		/local/schema.yml
 	rm schema-old.yml
-	sed -i 's/{/{/g' diff.md
-	sed -i 's/}/}/g' diff.md
+	$(SED_INPLACE) 's/{/{/g' diff.md
+	$(SED_INPLACE) 's/}/}/g' diff.md
 	npx prettier --write diff.md
 
 gen-clean-ts:  ## Remove generated API client for TypeScript

authentik/__init__.py

@@ -3,7 +3,7 @@
 from functools import lru_cache
 from os import environ
 
-VERSION = "2025.12.0-rc1"
+VERSION = "2026.2.0-rc1"
 ENV_GIT_HASH_KEY = "GIT_BUILD_HASH"
 
 

authentik/admin/api/version.py

@@ -37,7 +37,7 @@ class VersionSerializer(PassiveSerializer):
 
     def get_version_latest(self, _) -> str:
         """Get latest version from cache"""
-        if get_current_tenant().schema_name == get_public_schema_name():
+        if get_current_tenant().schema_name != get_public_schema_name():
             return authentik_version()
         version_in_cache = cache.get(VERSION_CACHE_KEY)
         if not version_in_cache:  # pragma: no cover

authentik/admin/files/api.py

@@ -240,7 +240,9 @@ class FileUsedByView(APIView):
             for field in fields:
                 q |= Q(**{field: params.get("name")})
 
-            objs = get_objects_for_user(request.user, f"{app}.view_{model_name}", model)
+            objs = get_objects_for_user(
+                request.user, f"{app}.view_{model_name}", model.objects.all()
+            )
             objs = objs.filter(q)
             for obj in objs:
                 serializer = UsedBySerializer(

authentik/admin/files/backends/base.py

@@ -1,10 +1,13 @@
-from collections.abc import Generator, Iterator
+from collections.abc import Callable, Generator, Iterator
+from typing import cast
 
+from django.core.cache import cache
 from django.http.request import HttpRequest
 from structlog.stdlib import get_logger
 
 from authentik.admin.files.usage import FileUsage
 
+CACHE_PREFIX = "goauthentik.io/admin/files"
 LOGGER = get_logger()
 
 
@@ -53,13 +56,19 @@ class Backend:
         """
         raise NotImplementedError
 
-    def file_url(self, name: str, request: HttpRequest | None = None) -> str:
+    def file_url(
+        self,
+        name: str,
+        request: HttpRequest | None = None,
+        use_cache: bool = True,
+    ) -> str:
         """
         Get URL for accessing the file.
 
         Args:
             file_path: Relative file path
             request: Optional Django HttpRequest for fully qualifed URL building
+            use_cache: whether to retrieve the URL from cache
 
         Returns:
             URL to access the file (may be relative or absolute depending on backend)
@@ -132,3 +141,22 @@ class ManageableBackend(Backend):
             True if file exists, False otherwise
         """
         raise NotImplementedError
+
+    def _cache_get_or_set(
+        self,
+        name: str,
+        request: HttpRequest | None,
+        default: Callable[[str, HttpRequest | None], str],
+        timeout: int,
+    ) -> str:
+        timeout_ignore = 60
+        timeout = int(timeout * 0.67)
+        if timeout < timeout_ignore:
+            timeout = 0
+
+        request_key = "None"
+        if request is not None:
+            request_key = f"{request.build_absolute_uri('/')}"
+        cache_key = f"{CACHE_PREFIX}/{self.name}/{self.usage}/{request_key}/{name}"
+
+        return cast(str, cache.get_or_set(cache_key, lambda: default(name, request), timeout))

authentik/admin/files/backends/file.py

@@ -63,7 +63,12 @@ class FileBackend(ManageableBackend):
                 rel_path = full_path.relative_to(self.base_path)
                 yield str(rel_path)
 
-    def file_url(self, name: str, request: HttpRequest | None = None) -> str:
+    def file_url(
+        self,
+        name: str,
+        request: HttpRequest | None = None,
+        use_cache: bool = True,
+    ) -> str:
         """Get URL for accessing the file."""
         expires_in = timedelta_from_string(
             CONFIG.get(
@@ -72,21 +77,28 @@ class FileBackend(ManageableBackend):
             )
         )
 
-        prefix = CONFIG.get("web.path", "/")[:-1]
-        path = f"{self.usage.value}/{connection.schema_name}/{name}"
-        token = jwt.encode(
-            payload={
-                "path": path,
-                "exp": now() + expires_in,
-                "nbf": now() - timedelta(seconds=15),
-            },
-            key=sha256(f"{settings.SECRET_KEY}:{self.usage}".encode()).hexdigest(),
-            algorithm="HS256",
-        )
-        url = f"{prefix}/files/{path}?token={token}"
-        if request is None:
-            return url
-        return request.build_absolute_uri(url)
+        def _file_url(name: str, request: HttpRequest | None) -> str:
+            prefix = CONFIG.get("web.path", "/")[:-1]
+            path = f"{self.usage.value}/{connection.schema_name}/{name}"
+            token = jwt.encode(
+                payload={
+                    "path": path,
+                    "exp": now() + expires_in,
+                    "nbf": now() - timedelta(seconds=15),
+                },
+                key=sha256(f"{settings.SECRET_KEY}:{self.usage}".encode()).hexdigest(),
+                algorithm="HS256",
+            )
+            url = f"{prefix}/files/{path}?token={token}"
+            if request is None:
+                return url
+            return request.build_absolute_uri(url)
+
+        if use_cache:
+            timeout = int(expires_in.total_seconds())
+            return self._cache_get_or_set(name, request, _file_url, timeout)
+        else:
+            return _file_url(name, request)
 
     def save_file(self, name: str, content: bytes) -> None:
         """Save file to local filesystem."""

authentik/admin/files/backends/passthrough.py

@@ -38,6 +38,11 @@ class PassthroughBackend(Backend):
         """External files cannot be listed."""
         yield from []
 
-    def file_url(self, name: str, request: HttpRequest | None = None) -> str:
+    def file_url(
+        self,
+        name: str,
+        request: HttpRequest | None = None,
+        use_cache: bool = True,
+    ) -> str:
         """Return the URL as-is for passthrough files."""
         return name

authentik/admin/files/backends/s3.py

@@ -130,44 +130,57 @@ class S3Backend(ManageableBackend):
                 if rel_path:  # Skip if it's just the directory itself
                     yield rel_path
 
-    def file_url(self, name: str, request: HttpRequest | None = None) -> str:
+    def file_url(
+        self,
+        name: str,
+        request: HttpRequest | None = None,
+        use_cache: bool = True,
+    ) -> str:
         """Generate presigned URL for file access."""
         use_https = CONFIG.get_bool(
             f"storage.{self.usage.value}.{self.name}.secure_urls",
             CONFIG.get_bool(f"storage.{self.name}.secure_urls", True),
         )
 
-        params = {
-            "Bucket": self.bucket_name,
-            "Key": f"{self.base_path}/{name}",
-        }
+        expires_in = int(
+            timedelta_from_string(
+                CONFIG.get(
+                    f"storage.{self.usage.value}.{self.name}.url_expiry",
+                    CONFIG.get(f"storage.{self.name}.url_expiry", "minutes=15"),
+                )
+            ).total_seconds()
+        )
 
-        expires_in = timedelta_from_string(
-            CONFIG.get(
-                f"storage.{self.usage.value}.{self.name}.url_expiry",
-                CONFIG.get(f"storage.{self.name}.url_expiry", "minutes=15"),
+        def _file_url(name: str, request: HttpRequest | None) -> str:
+            params = {
+                "Bucket": self.bucket_name,
+                "Key": f"{self.base_path}/{name}",
+            }
+
+            url = self.client.generate_presigned_url(
+                "get_object",
+                Params=params,
+                ExpiresIn=expires_in,
+                HttpMethod="GET",
             )
-        )
 
-        url = self.client.generate_presigned_url(
-            "get_object",
-            Params=params,
-            ExpiresIn=expires_in.total_seconds(),
-            HttpMethod="GET",
-        )
+            # Support custom domain for S3-compatible storage (so not AWS)
+            # Well, can't you do custom domains on AWS as well?
+            custom_domain = CONFIG.get(
+                f"storage.{self.usage.value}.{self.name}.custom_domain",
+                CONFIG.get(f"storage.{self.name}.custom_domain", None),
+            )
+            if custom_domain:
+                parsed = urlsplit(url)
+                scheme = "https" if use_https else "http"
+                url = f"{scheme}://{custom_domain}{parsed.path}?{parsed.query}"
 
-        # Support custom domain for S3-compatible storage (so not AWS)
-        # Well, can't you do custom domains on AWS as well?
-        custom_domain = CONFIG.get(
-            f"storage.{self.usage.value}.{self.name}.custom_domain",
-            CONFIG.get(f"storage.{self.name}.custom_domain", None),
-        )
-        if custom_domain:
-            parsed = urlsplit(url)
-            scheme = "https" if use_https else "http"
-            url = f"{scheme}://{custom_domain}{parsed.path}?{parsed.query}"
+            return url
 
-        return url
+        if use_cache:
+            return self._cache_get_or_set(name, request, _file_url, expires_in)
+        else:
+            return _file_url(name, request)
 
     def save_file(self, name: str, content: bytes) -> None:
         """Save file to S3."""

authentik/admin/files/backends/static.py

@@ -44,7 +44,12 @@ class StaticBackend(Backend):
                     if file_path.is_file() and (file_path.suffix in STATIC_FILE_EXTENSIONS):
                         yield f"{STATIC_PATH_PREFIX}/dist/{dir}/{file_path.name}"
 
-    def file_url(self, name: str, request: HttpRequest | None = None) -> str:
+    def file_url(
+        self,
+        name: str,
+        request: HttpRequest | None = None,
+        use_cache: bool = True,
+    ) -> str:
         """Get URL for static file."""
         prefix = CONFIG.get("web.path", "/")[:-1]
         url = f"{prefix}{name}"

authentik/admin/files/backends/tests/test_s3_backend.py

@@ -1,10 +1,13 @@
+from unittest import skipUnless
+
 from django.test import TestCase
 
-from authentik.admin.files.tests.utils import FileTestS3BackendMixin
+from authentik.admin.files.tests.utils import FileTestS3BackendMixin, s3_test_server_available
 from authentik.admin.files.usage import FileUsage
 from authentik.lib.config import CONFIG
 
 
+@skipUnless(s3_test_server_available(), "S3 test server not available")
 class TestS3Backend(FileTestS3BackendMixin, TestCase):
     """Test S3 backend functionality"""
 

authentik/admin/files/manager.py

@@ -70,6 +70,7 @@ class FileManager:
         self,
         name: str | None,
         request: HttpRequest | Request | None = None,
+        use_cache: bool = True,
     ) -> str:
         """
         Get URL for accessing the file.

authentik/admin/files/tests/test_manager.py

@@ -1,10 +1,16 @@
 """Test file service layer"""
 
+from unittest import skipUnless
+
 from django.http import HttpRequest
 from django.test import TestCase
 
 from authentik.admin.files.manager import FileManager
-from authentik.admin.files.tests.utils import FileTestFileBackendMixin, FileTestS3BackendMixin
+from authentik.admin.files.tests.utils import (
+    FileTestFileBackendMixin,
+    FileTestS3BackendMixin,
+    s3_test_server_available,
+)
 from authentik.admin.files.usage import FileUsage
 from authentik.lib.config import CONFIG
 
@@ -81,6 +87,7 @@ class TestResolveFileUrlFileBackend(FileTestFileBackendMixin, TestCase):
         self.assertEqual(result, "http://example.com/files/media/public/test.png")
 
 
+@skipUnless(s3_test_server_available(), "S3 test server not available")
 class TestResolveFileUrlS3Backend(FileTestS3BackendMixin, TestCase):
     @CONFIG.patch("storage.media.s3.custom_domain", "s3.test:8080/test")
     @CONFIG.patch("storage.media.s3.secure_urls", False)

authentik/admin/files/tests/utils.py

@@ -1,11 +1,26 @@
 import shutil
+import socket
 from tempfile import mkdtemp
+from urllib.parse import urlparse
 
 from authentik.admin.files.backends.s3 import S3Backend
 from authentik.admin.files.usage import FileUsage
 from authentik.lib.config import CONFIG, UNSET
 from authentik.lib.generators import generate_id
 
+S3_TEST_ENDPOINT = "http://localhost:8020"
+
+
+def s3_test_server_available() -> bool:
+    """Check if the S3 test server is reachable."""
+
+    parsed = urlparse(S3_TEST_ENDPOINT)
+    try:
+        with socket.create_connection((parsed.hostname, parsed.port), timeout=2):
+            return True
+    except OSError:
+        return False
+
 
 class FileTestFileBackendMixin:
     def setUp(self):
@@ -57,7 +72,7 @@ class FileTestS3BackendMixin:
         for key in s3_config_keys:
             self.original_media_s3_settings[key] = CONFIG.get(f"storage.media.s3.{key}", UNSET)
         self.media_s3_bucket_name = f"authentik-test-{generate_id(10)}".lower()
-        CONFIG.set("storage.media.s3.endpoint", "http://localhost:8020")
+        CONFIG.set("storage.media.s3.endpoint", S3_TEST_ENDPOINT)
         CONFIG.set("storage.media.s3.access_key", "accessKey1")
         CONFIG.set("storage.media.s3.secret_key", "secretKey1")
         CONFIG.set("storage.media.s3.bucket_name", self.media_s3_bucket_name)
@@ -70,7 +85,7 @@ class FileTestS3BackendMixin:
         for key in s3_config_keys:
             self.original_reports_s3_settings[key] = CONFIG.get(f"storage.reports.s3.{key}", UNSET)
         self.reports_s3_bucket_name = f"authentik-test-{generate_id(10)}".lower()
-        CONFIG.set("storage.reports.s3.endpoint", "http://localhost:8020")
+        CONFIG.set("storage.reports.s3.endpoint", S3_TEST_ENDPOINT)
         CONFIG.set("storage.reports.s3.access_key", "accessKey1")
         CONFIG.set("storage.reports.s3.secret_key", "secretKey1")
         CONFIG.set("storage.reports.s3.bucket_name", self.reports_s3_bucket_name)

authentik/api/pagination.py

@@ -13,6 +13,13 @@ class Pagination(pagination.PageNumberPagination):
     page_query_param = "page"
     page_size_query_param = "page_size"
 
+    def get_page_size(self, request):
+        if self.page_size_query_param in request.query_params:
+            page_size = super().get_page_size(request)
+            if page_size is not None:
+                return min(super().get_page_size(request), request.tenant.pagination_max_page_size)
+        return request.tenant.pagination_default_page_size
+
     def get_paginated_response(self, data):
         previous_page_number = 0
         if self.page.has_previous():

authentik/blueprints/apps.py

@@ -1,10 +1,12 @@
 """authentik Blueprints app"""
 
+import traceback
 from collections.abc import Callable
 from importlib import import_module
 from inspect import ismethod
 
 from django.apps import AppConfig
+from django.conf import settings
 from django.db import DatabaseError, InternalError, ProgrammingError
 from dramatiq.broker import get_broker
 from structlog.stdlib import BoundLogger, get_logger
@@ -44,8 +46,21 @@ class ManagedAppConfig(AppConfig):
                 module_name = f"{self.name}.{rel_module}"
                 import_module(module_name)
                 self.logger.info("Imported related module", module=module_name)
-            except ModuleNotFoundError:
-                pass
+            except ModuleNotFoundError as exc:
+                if settings.DEBUG:
+                    # This is a heuristic for determining whether the exception was caused
+                    # "directly" by the `import_module` call or whether the initial import
+                    # succeeded and a later import (within the existing module) failed.
+                    # 1. <the calling function>
+                    # 2. importlib.import_module
+                    # 3. importlib._bootstrap._gcd_import
+                    # 4. importlib._bootstrap._find_and_load
+                    # 5. importlib._bootstrap._find_and_load_unlocked
+                    STACK_LENGTH_HEURISTIC = 5
+
+                    stack_length = len(traceback.extract_tb(exc.__traceback__))
+                    if stack_length > STACK_LENGTH_HEURISTIC:
+                        raise
 
         import_relative("checks")
         import_relative("tasks")

authentik/blueprints/management/commands/make_blueprint_schema.py

@@ -5,6 +5,7 @@ from typing import Any
 
 from django.core.management.base import BaseCommand, no_translations
 from django.db.models import Model, fields
+from django.db.models.fields.related import OneToOneField
 from drf_jsonschema_serializer.convert import converter, field_to_converter
 from rest_framework.fields import Field, JSONField, UUIDField
 from rest_framework.relations import PrimaryKeyRelatedField
@@ -32,6 +33,8 @@ class PrimaryKeyRelatedFieldConverter:
     def convert(self, field: PrimaryKeyRelatedField):
         model: Model = field.queryset.model
         pk_field = model._meta.pk
+        if isinstance(pk_field, OneToOneField):
+            pk_field = pk_field.related_fields[0][1]
         if isinstance(pk_field, fields.UUIDField):
             return {"type": "string", "format": "uuid"}
         return {"type": "integer"}

authentik/core/api/applications.py

@@ -4,7 +4,8 @@ from collections.abc import Iterator
 from copy import copy
 
 from django.core.cache import cache
-from django.db.models import QuerySet
+from django.db.models import Case, QuerySet
+from django.db.models.expressions import When
 from django.shortcuts import get_object_or_404
 from django.utils.translation import gettext as _
 from drf_spectacular.types import OpenApiTypes
@@ -22,6 +23,7 @@ from authentik.api.pagination import Pagination
 from authentik.blueprints.v1.importer import SERIALIZER_CONTEXT_BLUEPRINT
 from authentik.core.api.providers import ProviderSerializer
 from authentik.core.api.used_by import UsedByMixin
+from authentik.core.api.users import UserSerializer
 from authentik.core.api.utils import ModelSerializer
 from authentik.core.models import Application, User
 from authentik.events.logs import LogEventSerializer, capture_logs
@@ -55,9 +57,21 @@ class ApplicationSerializer(ModelSerializer):
     def get_launch_url(self, app: Application) -> str | None:
         """Allow formatting of launch URL"""
         user = None
+        user_data = None
+
         if "request" in self.context:
             user = self.context["request"].user
-        return app.get_launch_url(user)
+
+        # Cache serialized user data to avoid N+1 when formatting launch URLs
+        # for multiple applications. UserSerializer accesses user.ak_groups which
+        # would otherwise trigger a query for each application.
+        if user is not None:
+            if "_cached_user_data" not in self.context:
+                # Prefetch groups to avoid N+1
+                self.context["_cached_user_data"] = UserSerializer(instance=user).data
+            user_data = self.context["_cached_user_data"]
+
+        return app.get_launch_url(user, user_data=user_data)
 
     def validate_slug(self, slug: str) -> str:
         if slug in Application.reserved_slugs:
@@ -150,8 +164,23 @@ class ApplicationViewSet(UsedByMixin, ModelViewSet):
                 applications.append(application)
         return applications
 
+    def _expand_applications(self, applications: list[Application]) -> QuerySet[Application]:
+        """
+        Re-fetch with proper prefetching for serialization
+        Cached applications don't have prefetched relationships, causing N+1 queries
+        during serialization when get_provider() is called
+        """
+        if not applications:
+            return self.get_queryset().none()
+        pks = [app.pk for app in applications]
+        return (
+            self.get_queryset()
+            .filter(pk__in=pks)
+            .order_by(Case(*[When(pk=pk, then=pos) for pos, pk in enumerate(pks)]))
+        )
+
     def _filter_applications_with_launch_url(
-        self, paginated_apps: Iterator[Application]
+        self, paginated_apps: QuerySet[Application]
     ) -> list[Application]:
         applications = []
         for app in paginated_apps:
@@ -254,6 +283,8 @@ class ApplicationViewSet(UsedByMixin, ModelViewSet):
             except ValueError as exc:
                 raise ValidationError from exc
             allowed_applications = self._get_allowed_applications(paginated_apps, user=for_user)
+            allowed_applications = self._expand_applications(allowed_applications)
+
             serializer = self.get_serializer(allowed_applications, many=True)
             return self.get_paginated_response(serializer.data)
 
@@ -272,6 +303,7 @@ class ApplicationViewSet(UsedByMixin, ModelViewSet):
                     allowed_applications,
                     timeout=86400,
                 )
+        allowed_applications = self._expand_applications(allowed_applications)
 
         if only_with_launch_url == "true":
             allowed_applications = self._filter_applications_with_launch_url(allowed_applications)

authentik/core/api/groups.py

@@ -33,6 +33,16 @@ from authentik.endpoints.connectors.agent.auth import AgentAuth
 from authentik.rbac.api.roles import RoleSerializer
 from authentik.rbac.decorators import permission_required
 
+PARTIAL_USER_SERIALIZER_MODEL_FIELDS = [
+    "pk",
+    "username",
+    "name",
+    "is_active",
+    "last_login",
+    "email",
+    "attributes",
+]
+
 
 class PartialUserSerializer(ModelSerializer):
     """Partial User Serializer, does not include child relations."""
@@ -42,16 +52,7 @@ class PartialUserSerializer(ModelSerializer):
 
     class Meta:
         model = User
-        fields = [
-            "pk",
-            "username",
-            "name",
-            "is_active",
-            "last_login",
-            "email",
-            "attributes",
-            "uid",
-        ]
+        fields = PARTIAL_USER_SERIALIZER_MODEL_FIELDS + ["uid"]
 
 
 class RelatedGroupSerializer(ModelSerializer):
@@ -262,7 +263,14 @@ class GroupViewSet(UsedByMixin, ModelViewSet):
         base_qs = Group.objects.all().prefetch_related("roles")
 
         if self.serializer_class(context={"request": self.request})._should_include_users:
-            base_qs = base_qs.prefetch_related("users")
+            # Only fetch fields needed by PartialUserSerializer to reduce DB load and instantiation
+            # time
+            base_qs = base_qs.prefetch_related(
+                Prefetch(
+                    "users",
+                    queryset=User.objects.all().only(*PARTIAL_USER_SERIALIZER_MODEL_FIELDS),
+                )
+            )
         else:
             base_qs = base_qs.prefetch_related(
                 Prefetch("users", queryset=User.objects.all().only("id"))

authentik/core/api/tokens.py

@@ -4,7 +4,6 @@ from typing import Any
 
 from django.utils.timezone import now
 from drf_spectacular.utils import OpenApiResponse, extend_schema
-from guardian.shortcuts import get_anonymous_user
 from rest_framework.decorators import action
 from rest_framework.exceptions import ValidationError
 from rest_framework.fields import CharField
@@ -145,12 +144,6 @@ class TokenViewSet(UsedByMixin, ModelViewSet):
     owner_field = "user"
     rbac_allow_create_without_perm = True
 
-    def get_queryset(self):
-        user = self.request.user if self.request else get_anonymous_user()
-        if user.is_superuser:
-            return super().get_queryset()
-        return super().get_queryset().filter(user=user.pk)
-
     def perform_create(self, serializer: TokenSerializer):
         if not self.request.user.is_superuser:
             instance = serializer.save(

authentik/core/api/users.py

@@ -86,6 +86,7 @@ from authentik.flows.models import FlowToken
 from authentik.flows.planner import PLAN_CONTEXT_PENDING_USER, FlowPlanner
 from authentik.flows.views.executor import QS_KEY_TOKEN
 from authentik.lib.avatars import get_avatar
+from authentik.lib.utils.reflection import ConditionalInheritance
 from authentik.rbac.api.roles import RoleSerializer
 from authentik.rbac.decorators import permission_required
 from authentik.rbac.models import Role, get_permission_choices
@@ -484,7 +485,11 @@ class UsersFilter(FilterSet):
         ]
 
 
-class UserViewSet(UsedByMixin, ModelViewSet):
+class UserViewSet(
+    ConditionalInheritance("authentik.enterprise.reports.api.reports.ExportMixin"),
+    UsedByMixin,
+    ModelViewSet,
+):
     """User Viewset"""
 
     queryset = User.objects.none()

authentik/core/migrations/0056_user_roles.py

@@ -18,10 +18,9 @@ def migrate_object_permissions(apps: Apps, schema_editor: BaseDatabaseSchemaEdit
     RoleModelPermission = apps.get_model("guardian", "RoleModelPermission")
 
     def get_role_for_user_id(user_id: int) -> Role:
-        name = f"ak-managed-role--user-{user_id}"
+        name = f"ak-migrated-role--user-{user_id}"
         role, created = Role.objects.using(db_alias).get_or_create(
             name=name,
-            managed=name,
         )
         if created:
             role.users.add(user_id)
@@ -32,11 +31,10 @@ def migrate_object_permissions(apps: Apps, schema_editor: BaseDatabaseSchemaEdit
         if not role:
             # Every django group should already have a role, so this should never happen.
             # But let's be nice.
-            name = f"ak-managed-role--group-{group_id}"
+            name = f"ak-migrated-role--group-{group_id}"
             role, created = Role.objects.using(db_alias).get_or_create(
                 group_id=group_id,
                 name=name,
-                managed=name,
             )
             if created:
                 role.group_id = group_id

authentik/core/models.py

@@ -152,7 +152,7 @@ class AttributesMixin(models.Model):
     @classmethod
     def update_or_create_attributes(
         cls, query: dict[str, Any], properties: dict[str, Any]
-    ) -> tuple[models.Model, bool]:
+    ) -> tuple[Self, bool]:
         """Same as django's update_or_create but correctly updates attributes by merging dicts"""
         instance = cls.objects.filter(**query).first()
         if not instance:
@@ -658,6 +658,10 @@ class ApplicationQuerySet(QuerySet):
         qs = self.select_related("provider")
         for subclass in Provider.objects.get_queryset()._get_subclasses_recurse(Provider):
             qs = qs.select_related(f"provider__{subclass}")
+            # Also prefetch/select through each subclass path to ensure casted instances have access
+            qs = qs.prefetch_related(f"provider__{subclass}__property_mappings")
+            qs = qs.select_related(f"provider__{subclass}__application")
+            qs = qs.select_related(f"provider__{subclass}__backchannel_application")
         return qs
 
 
@@ -709,8 +713,15 @@ class Application(SerializerModel, PolicyBindingModel):
 
         return get_file_manager(FileUsage.MEDIA).file_url(self.meta_icon)
 
-    def get_launch_url(self, user: Optional["User"] = None) -> str | None:
-        """Get launch URL if set, otherwise attempt to get launch URL based on provider."""
+    def get_launch_url(
+        self, user: Optional["User"] = None, user_data: dict | None = None
+    ) -> str | None:
+        """Get launch URL if set, otherwise attempt to get launch URL based on provider.
+
+        Args:
+            user: User instance for formatting the URL
+            user_data: Pre-serialized user data to avoid re-serialization (performance optimization)
+        """
         from authentik.core.api.users import UserSerializer
 
         url = None
@@ -720,7 +731,10 @@ class Application(SerializerModel, PolicyBindingModel):
             url = provider.launch_url
         if user and url:
             try:
-                return url % UserSerializer(instance=user).data
+                # Use pre-serialized data if available, otherwise serialize now
+                if user_data is None:
+                    user_data = UserSerializer(instance=user).data
+                return url % user_data
             except Exception as exc:  # noqa
                 LOGGER.warning("Failed to format launch url", exc=exc)
                 return url

authentik/core/sessions.py

@@ -34,18 +34,12 @@ class SessionStore(SessionBase):
 
     def _get_session_from_db(self):
         try:
-            return (
-                self.model.objects.select_related(
-                    "authenticatedsession",
-                    "authenticatedsession__user",
-                )
-                .prefetch_related(
-                    "authenticatedsession__user__groups",
-                )
-                .get(
-                    session_key=self.session_key,
-                    expires__gt=timezone.now(),
-                )
+            return self.model.objects.select_related(
+                "authenticatedsession",
+                "authenticatedsession__user",
+            ).get(
+                session_key=self.session_key,
+                expires__gt=timezone.now(),
             )
         except (self.model.DoesNotExist, SuspiciousOperation) as exc:
             if isinstance(exc, SuspiciousOperation):
@@ -54,18 +48,12 @@ class SessionStore(SessionBase):
 
     async def _aget_session_from_db(self):
         try:
-            return (
-                await self.model.objects.select_related(
-                    "authenticatedsession",
-                    "authenticatedsession__user",
-                )
-                .prefetch_related(
-                    "authenticatedsession__user__groups",
-                )
-                .aget(
-                    session_key=self.session_key,
-                    expires__gt=timezone.now(),
-                )
+            return await self.model.objects.select_related(
+                "authenticatedsession",
+                "authenticatedsession__user",
+            ).aget(
+                session_key=self.session_key,
+                expires__gt=timezone.now(),
             )
         except (self.model.DoesNotExist, SuspiciousOperation) as exc:
             if isinstance(exc, SuspiciousOperation):

authentik/core/tests/test_models.py

@@ -39,7 +39,7 @@ def source_tester_factory(test_model: type[Source]) -> Callable:
     def tester(self: TestModels):
         model_class = None
         if test_model._meta.abstract:
-            model_class = [x for x in test_model.__bases__ if issubclass(x, Source)][0]()
+            return
         else:
             model_class = test_model()
         model_class.slug = "test"

authentik/core/tests/test_token_api.py

@@ -183,16 +183,16 @@ class TestTokenAPI(APITestCase):
         self.assertEqual(len(body["results"]), 1)
         self.assertEqual(body["results"][0]["identifier"], token_should.identifier)
 
-    def test_list_admin(self):
-        """Test Token List (Test with admin auth)"""
+    def test_list_with_permission(self):
+        """Test Token List (Test with `view_token` permission)"""
         Token.objects.all().delete()
-        self.client.force_login(self.admin)
         token_should: Token = Token.objects.create(
             identifier="test", expiring=False, user=self.user
         )
         token_should_not: Token = Token.objects.create(
             identifier="test-2", expiring=False, user=get_anonymous_user()
         )
+        self.user.assign_perms_to_managed_role("authentik_core.view_token")
         response = self.client.get(reverse("authentik_api:token-list"))
         body = loads(response.content)
         self.assertEqual(len(body["results"]), 2)

authentik/crypto/api.py

@@ -1,7 +1,5 @@
 """Crypto API Views"""
 
-from datetime import datetime
-
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives.serialization import load_pem_private_key
 from cryptography.x509 import load_pem_x509_certificate
@@ -15,14 +13,12 @@ from drf_spectacular.utils import (
     OpenApiParameter,
     OpenApiResponse,
     extend_schema,
-    extend_schema_field,
 )
 from rest_framework.decorators import action
 from rest_framework.exceptions import ValidationError
 from rest_framework.fields import (
     CharField,
     ChoiceField,
-    DateTimeField,
     IntegerField,
     SerializerMethodField,
 )
@@ -51,59 +47,15 @@ LOGGER = get_logger()
 class CertificateKeyPairSerializer(ModelSerializer):
     """CertificateKeyPair Serializer"""
 
-    fingerprint_sha256 = SerializerMethodField()
-    fingerprint_sha1 = SerializerMethodField()
-
-    cert_expiry = SerializerMethodField()
-    cert_subject = SerializerMethodField()
     private_key_available = SerializerMethodField()
-    key_type = SerializerMethodField()
 
     certificate_download_url = SerializerMethodField()
     private_key_download_url = SerializerMethodField()
 
-    @property
-    def _should_include_details(self) -> bool:
-        request: Request = self.context.get("request", None)
-        if not request:
-            return True
-        return str(request.query_params.get("include_details", "true")).lower() == "true"
-
-    def get_fingerprint_sha256(self, instance: CertificateKeyPair) -> str | None:
-        "Get certificate Hash (SHA256)"
-        if not self._should_include_details:
-            return None
-        return instance.fingerprint_sha256
-
-    def get_fingerprint_sha1(self, instance: CertificateKeyPair) -> str | None:
-        "Get certificate Hash (SHA1)"
-        if not self._should_include_details:
-            return None
-        return instance.fingerprint_sha1
-
-    def get_cert_expiry(self, instance: CertificateKeyPair) -> datetime | None:
-        "Get certificate expiry"
-        if not self._should_include_details:
-            return None
-        return DateTimeField().to_representation(instance.certificate.not_valid_after_utc)
-
-    def get_cert_subject(self, instance: CertificateKeyPair) -> str | None:
-        """Get certificate subject as full rfc4514"""
-        if not self._should_include_details:
-            return None
-        return instance.certificate.subject.rfc4514_string()
-
     def get_private_key_available(self, instance: CertificateKeyPair) -> bool:
         """Show if this keypair has a private key configured or not"""
         return instance.key_data != "" and instance.key_data is not None
 
-    @extend_schema_field(ChoiceField(choices=KeyType.choices, allow_null=True))
-    def get_key_type(self, instance: CertificateKeyPair) -> str | None:
-        """Get the key algorithm type from the certificate's public key"""
-        if not self._should_include_details:
-            return None
-        return instance.key_type
-
     def get_certificate_download_url(self, instance: CertificateKeyPair) -> str:
         """Get URL to download certificate"""
         return (
@@ -175,6 +127,11 @@ class CertificateKeyPairSerializer(ModelSerializer):
             "managed": {"read_only": True},
             "key_data": {"write_only": True},
             "certificate_data": {"write_only": True},
+            "fingerprint_sha256": {"read_only": True},
+            "fingerprint_sha1": {"read_only": True},
+            "cert_expiry": {"read_only": True},
+            "cert_subject": {"read_only": True},
+            "key_type": {"read_only": True},
         }
 
 
@@ -216,17 +173,12 @@ class CertificateKeyPairFilter(FilterSet):
         return queryset.exclude(key_data__exact="")
 
     def filter_key_type(self, queryset, name, value):  # pragma: no cover
-        """Filter certificates by key type using the public key from the certificate"""
+        """Filter certificates by key type using the stored database field"""
         if not value:
             return queryset
 
         # value is a list of KeyType enum values from MultipleChoiceFilter
-        filtered_pks = []
-        for cert in queryset:
-            if cert.key_type in value:
-                filtered_pks.append(cert.pk)
-
-        return queryset.filter(pk__in=filtered_pks)
+        return queryset.filter(key_type__in=value)
 
     class Meta:
         model = CertificateKeyPair
@@ -263,7 +215,6 @@ class CertificateKeyPairViewSet(UsedByMixin, ModelViewSet):
                     "Can be specified multiple times (e.g. '?key_type=rsa&key_type=ec')"
                 ),
             ),
-            OpenApiParameter("include_details", bool, default=True),
         ]
     )
     def list(self, request, *args, **kwargs):

authentik/crypto/migrations/0006_certificatekeypair_cert_expiry_and_more.py

@@ -0,0 +1,117 @@
+# Generated by Django 5.2.9 on 2025-12-09 06:22
+
+from hashlib import md5
+
+from cryptography.hazmat.backends import default_backend
+from cryptography.x509 import load_pem_x509_certificate
+from django.db import migrations, models
+
+from authentik.crypto.signals import extract_certificate_metadata
+
+
+def backfill_certificate_metadata(apps, schema_editor):  # noqa: ARG001
+    """Backfill certificate metadata and kid for existing records."""
+
+    CertificateKeyPair = apps.get_model("authentik_crypto", "CertificateKeyPair")
+
+    for cert in CertificateKeyPair.objects.all():
+        updated_fields = []
+
+        if cert.certificate_data:
+            try:
+                certificate = load_pem_x509_certificate(
+                    cert.certificate_data.encode("utf-8"), default_backend()
+                )
+                metadata = extract_certificate_metadata(certificate)
+
+                cert.key_type = metadata["key_type"]
+                cert.cert_expiry = metadata["cert_expiry"]
+                cert.cert_subject = metadata["cert_subject"]
+                cert.fingerprint_sha256 = metadata["fingerprint_sha256"]
+                cert.fingerprint_sha1 = metadata["fingerprint_sha1"]
+                updated_fields.extend(
+                    [
+                        "key_type",
+                        "cert_expiry",
+                        "cert_subject",
+                        "fingerprint_sha256",
+                        "fingerprint_sha1",
+                    ]
+                )
+            except (ValueError, TypeError, AttributeError):
+                pass
+
+        # Backfill kid with MD5 for backwards compatibility
+        if cert.key_data:
+            cert.kid = md5(cert.key_data.encode("utf-8"), usedforsecurity=False).hexdigest()
+            updated_fields.append("kid")
+
+        if updated_fields:
+            cert.save(update_fields=updated_fields)
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("authentik_crypto", "0005_alter_certificatekeypair_options"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="certificatekeypair",
+            name="cert_expiry",
+            field=models.DateTimeField(blank=True, help_text="Certificate expiry date", null=True),
+        ),
+        migrations.AddField(
+            model_name="certificatekeypair",
+            name="cert_subject",
+            field=models.TextField(
+                blank=True, help_text="Certificate subject as RFC4514 string", null=True
+            ),
+        ),
+        migrations.AddField(
+            model_name="certificatekeypair",
+            name="fingerprint_sha1",
+            field=models.CharField(
+                blank=True,
+                help_text="SHA1 fingerprint of the certificate",
+                max_length=59,
+                null=True,
+            ),
+        ),
+        migrations.AddField(
+            model_name="certificatekeypair",
+            name="fingerprint_sha256",
+            field=models.CharField(
+                blank=True,
+                help_text="SHA256 fingerprint of the certificate",
+                max_length=95,
+                null=True,
+            ),
+        ),
+        migrations.AddField(
+            model_name="certificatekeypair",
+            name="key_type",
+            field=models.CharField(
+                blank=True,
+                choices=[
+                    ("rsa", "RSA"),
+                    ("ec", "Elliptic Curve"),
+                    ("dsa", "DSA"),
+                    ("ed25519", "Ed25519"),
+                    ("ed448", "Ed448"),
+                ],
+                help_text="Key algorithm type detected from the certificate's public key",
+                max_length=16,
+                null=True,
+            ),
+        ),
+        migrations.AddField(
+            model_name="certificatekeypair",
+            name="kid",
+            field=models.CharField(
+                blank=True, help_text="Key ID generated from private key", max_length=128, null=True
+            ),
+        ),
+        migrations.RunPython(backfill_certificate_metadata, migrations.RunPython.noop),
+    ]

authentik/crypto/models.py

@@ -1,7 +1,8 @@
 """authentik crypto models"""
 
+from base64 import urlsafe_b64encode
 from binascii import hexlify
-from hashlib import md5
+from hashlib import md5, sha512
 from ssl import PEM_FOOTER, PEM_HEADER
 from textwrap import wrap
 from uuid import uuid4
@@ -47,6 +48,39 @@ def fingerprint_sha256(cert: Certificate) -> str:
     return hexlify(cert.fingerprint(hashes.SHA256()), ":").decode("utf-8")
 
 
+def detect_key_type(certificate: Certificate) -> str | None:
+    """Detect the key algorithm type by parsing the certificate's public key"""
+    try:
+        public_key = certificate.public_key()
+        if isinstance(public_key, RSAPublicKey):
+            return KeyType.RSA
+        if isinstance(public_key, EllipticCurvePublicKey):
+            return KeyType.EC
+        if isinstance(public_key, DSAPublicKey):
+            return KeyType.DSA
+        if isinstance(public_key, Ed25519PublicKey):
+            return KeyType.ED25519
+        if isinstance(public_key, Ed448PublicKey):
+            return KeyType.ED448
+    except (ValueError, TypeError, AttributeError) as exc:
+        LOGGER.warning("Failed to detect key type", exc=exc)
+    return None
+
+
+def generate_key_id(key_data: str) -> str:
+    """Generate Key ID using SHA512 + urlsafe_b64encode."""
+    if not key_data:
+        return ""
+    return urlsafe_b64encode(sha512(key_data.encode("utf-8")).digest()).decode("utf-8").rstrip("=")
+
+
+def generate_key_id_legacy(key_data: str) -> str:
+    """Generate Key ID using MD5 (legacy format for backwards compatibility)."""
+    if not key_data:
+        return ""
+    return md5(key_data.encode("utf-8")).hexdigest()  # nosec
+
+
 class CertificateKeyPair(SerializerModel, ManagedModel, CreatedUpdatedModel):
     """CertificateKeyPair that can be used for signing or encrypting if `key_data`
     is set, otherwise it can be used to verify remote data."""
@@ -62,6 +96,41 @@ class CertificateKeyPair(SerializerModel, ManagedModel, CreatedUpdatedModel):
         blank=True,
         default="",
     )
+    key_type = models.CharField(
+        max_length=16,
+        choices=KeyType.choices,
+        null=True,
+        blank=True,
+        help_text=_("Key algorithm type detected from the certificate's public key"),
+    )
+    cert_expiry = models.DateTimeField(
+        null=True,
+        blank=True,
+        help_text=_("Certificate expiry date"),
+    )
+    cert_subject = models.TextField(
+        null=True,
+        blank=True,
+        help_text=_("Certificate subject as RFC4514 string"),
+    )
+    fingerprint_sha256 = models.CharField(
+        max_length=95,
+        null=True,
+        blank=True,
+        help_text=_("SHA256 fingerprint of the certificate"),
+    )
+    fingerprint_sha1 = models.CharField(
+        max_length=59,
+        null=True,
+        blank=True,
+        help_text=_("SHA1 fingerprint of the certificate"),
+    )
+    kid = models.CharField(
+        max_length=128,
+        null=True,
+        blank=True,
+        help_text=_("Key ID generated from private key"),
+    )
 
     _cert: Certificate | None = None
     _private_key: PrivateKeyTypes | None = None
@@ -106,41 +175,6 @@ class CertificateKeyPair(SerializerModel, ManagedModel, CreatedUpdatedModel):
                 return None
         return self._private_key
 
-    @property
-    def fingerprint_sha256(self) -> str:
-        """Get SHA256 Fingerprint of certificate_data"""
-        return fingerprint_sha256(self.certificate)
-
-    @property
-    def fingerprint_sha1(self) -> str:
-        """Get SHA1 Fingerprint of certificate_data"""
-        return hexlify(self.certificate.fingerprint(hashes.SHA1()), ":").decode("utf-8")  # nosec
-
-    @property
-    def kid(self):
-        """Get Key ID used for JWKS"""
-        return (
-            md5(self.key_data.encode("utf-8"), usedforsecurity=False).hexdigest()
-            if self.key_data
-            else ""
-        )  # nosec
-
-    @property
-    def key_type(self) -> str | None:
-        """Get the key algorithm type from the certificate's public key"""
-        public_key = self.certificate.public_key()
-        if isinstance(public_key, RSAPublicKey):
-            return KeyType.RSA
-        if isinstance(public_key, EllipticCurvePublicKey):
-            return KeyType.EC
-        if isinstance(public_key, DSAPublicKey):
-            return KeyType.DSA
-        if isinstance(public_key, Ed25519PublicKey):
-            return KeyType.ED25519
-        if isinstance(public_key, Ed448PublicKey):
-            return KeyType.ED448
-        return None
-
     def __str__(self) -> str:
         return f"Certificate-Key Pair {self.name}"
 

authentik/crypto/signals.py

@@ -0,0 +1,70 @@
+"""authentik crypto signals"""
+
+from binascii import hexlify
+from datetime import datetime
+from ssl import CertificateError
+
+from cryptography.hazmat.primitives import hashes
+from cryptography.x509 import Certificate
+from django.db.models.signals import pre_save
+from django.dispatch import receiver
+from structlog.stdlib import get_logger
+
+from authentik.crypto.models import (
+    CertificateKeyPair,
+    detect_key_type,
+    fingerprint_sha256,
+    generate_key_id,
+    generate_key_id_legacy,
+)
+
+LOGGER = get_logger()
+
+
+def extract_certificate_metadata(certificate: Certificate) -> dict[str, str | datetime]:
+    """Extract all metadata fields from a certificate."""
+    metadata = {}
+
+    try:
+        metadata["key_type"] = detect_key_type(certificate)
+        metadata["cert_expiry"] = certificate.not_valid_after_utc
+        metadata["cert_subject"] = certificate.subject.rfc4514_string()
+        metadata["fingerprint_sha256"] = fingerprint_sha256(certificate)
+        metadata["fingerprint_sha1"] = hexlify(
+            certificate.fingerprint(hashes.SHA1()), ":"  # nosec
+        ).decode("utf-8")
+    except (ValueError, TypeError, AttributeError) as exc:
+        raise CertificateError(f"Invalid certificate metadata: {exc}") from exc
+
+    return metadata
+
+
+@receiver(pre_save, sender="authentik_crypto.CertificateKeyPair")
+def certificate_key_pair_pre_save(
+    sender: type[CertificateKeyPair], instance: CertificateKeyPair, **_
+):
+    """Automatically populate certificate metadata fields before saving"""
+
+    # Only extract metadata if certificate_data is present
+    if not instance.certificate_data:
+        return
+
+    try:
+        metadata = extract_certificate_metadata(instance.certificate)
+    except (CertificateError, ValueError, TypeError, AttributeError) as exc:
+        LOGGER.warning("Failed to extract certificate metadata", exc=exc)
+        return
+
+    instance.key_type = metadata["key_type"]
+    instance.cert_expiry = metadata["cert_expiry"]
+    instance.cert_subject = metadata["cert_subject"]
+    instance.fingerprint_sha256 = metadata["fingerprint_sha256"]
+    instance.fingerprint_sha1 = metadata["fingerprint_sha1"]
+
+    # Generate kid if not set, or regenerate if key_data has changed
+    # Preserve existing kid (MD5 or SHA512) if it matches the current key_data
+    if instance.key_data:
+        new_kid = generate_key_id(instance.key_data)
+        legacy_kid = generate_key_id_legacy(instance.key_data)
+        if instance.kid not in (new_kid, legacy_kid):
+            instance.kid = new_kid

authentik/crypto/tests.py

@@ -20,7 +20,7 @@ from authentik.core.tests.utils import (
 )
 from authentik.crypto.api import CertificateKeyPairSerializer
 from authentik.crypto.builder import CertificateBuilder
-from authentik.crypto.models import CertificateKeyPair
+from authentik.crypto.models import CertificateKeyPair, generate_key_id, generate_key_id_legacy
 from authentik.crypto.tasks import MANAGED_DISCOVERED, certificate_discovery
 from authentik.lib.config import CONFIG
 from authentik.lib.generators import generate_id, generate_key
@@ -173,21 +173,24 @@ class TestCrypto(APITestCase):
         self.assertEqual(api_cert["fingerprint_sha1"], cert.fingerprint_sha1)
         self.assertEqual(api_cert["fingerprint_sha256"], cert.fingerprint_sha256)
 
-    def test_list_without_details(self):
-        """Test API List (no details)"""
+    def test_list_always_includes_details(self):
+        """Test API List always includes certificate details"""
         cert = create_test_cert()
         self.client.force_login(create_test_admin_user())
         response = self.client.get(
             reverse(
                 "authentik_api:certificatekeypair-list",
             ),
-            data={"name": cert.name, "include_details": False},
+            data={"name": cert.name},
         )
         self.assertEqual(response.status_code, 200)
         body = loads(response.content.decode())
         api_cert = [x for x in body["results"] if x["name"] == cert.name][0]
-        self.assertEqual(api_cert["fingerprint_sha1"], None)
-        self.assertEqual(api_cert["fingerprint_sha256"], None)
+        # All details should now always be included
+        self.assertEqual(api_cert["fingerprint_sha1"], cert.fingerprint_sha1)
+        self.assertEqual(api_cert["fingerprint_sha256"], cert.fingerprint_sha256)
+        self.assertIsNotNone(api_cert["cert_expiry"])
+        self.assertIsNotNone(api_cert["cert_subject"])
 
     def test_certificate_download(self):
         """Test certificate export (download)"""
@@ -426,3 +429,114 @@ class TestCrypto(APITestCase):
             self.assertEqual(
                 1, final_count, "Should not create duplicate cert for same private key"
             )
+
+    def test_metadata_extraction_with_cert_and_key(self):
+        """Test that metadata is extracted when creating keypair with certificate and key"""
+        cert = create_test_cert()
+
+        # Verify all metadata fields are populated
+        self.assertIsNotNone(cert.key_type)
+        self.assertIsNotNone(cert.cert_expiry)
+        self.assertIsNotNone(cert.cert_subject)
+        self.assertIsNotNone(cert.fingerprint_sha256)
+        self.assertIsNotNone(cert.fingerprint_sha1)
+
+        # Verify kid is generated using SHA512 for new records
+        self.assertIsNotNone(cert.kid)
+        self.assertEqual(cert.kid, generate_key_id(cert.key_data))
+
+    def test_metadata_extraction_without_key(self):
+        """Test that metadata is extracted when creating keypair without private key"""
+        builder = CertificateBuilder(generate_id())
+        builder.build(subject_alt_names=[], validity_days=3)
+
+        # Create keypair with only certificate, no key
+        cert = CertificateKeyPair.objects.create(
+            name=generate_id(),
+            certificate_data=builder.certificate,
+            key_data="",
+        )
+
+        # Verify certificate metadata fields are populated
+        self.assertIsNotNone(cert.key_type)
+        self.assertIsNotNone(cert.cert_expiry)
+        self.assertIsNotNone(cert.cert_subject)
+        self.assertIsNotNone(cert.fingerprint_sha256)
+        self.assertIsNotNone(cert.fingerprint_sha1)
+
+        # Verify kid is empty when no key_data
+        self.assertEqual(cert.kid, None)
+
+    def test_metadata_extraction_invalid_cert(self):
+        """Test that invalid certificate data doesn't crash, just skips metadata"""
+        cert = CertificateKeyPair.objects.create(
+            name=generate_id(),
+            certificate_data="invalid certificate data",
+            key_data="",
+        )
+
+        # Verify metadata fields are None for invalid cert
+        self.assertIsNone(cert.key_type)
+        self.assertIsNone(cert.cert_expiry)
+        self.assertIsNone(cert.cert_subject)
+        self.assertIsNone(cert.fingerprint_sha256)
+        self.assertIsNone(cert.fingerprint_sha1)
+        self.assertIsNone(cert.kid)
+
+    def test_kid_legacy_preservation(self):
+        """Test that legacy MD5 kid is preserved when key_data hasn't changed"""
+        cert = create_test_cert()
+
+        # Simulate a legacy MD5 kid (as if backfilled from old system)
+        legacy_kid = generate_key_id_legacy(cert.key_data)
+        CertificateKeyPair.objects.filter(pk=cert.pk).update(kid=legacy_kid)
+        cert.refresh_from_db()
+        self.assertEqual(cert.kid, legacy_kid)
+
+        # Save the cert again (e.g., name change) - kid should be preserved
+        cert.name = generate_id()
+        cert.save()
+        cert.refresh_from_db()
+
+        self.assertEqual(cert.kid, legacy_kid)
+
+    def test_kid_regenerated_on_key_change(self):
+        """Test that kid is regenerated when key_data changes"""
+        cert = create_test_cert()
+        original_kid = cert.kid
+
+        # Generate a new key and update the keypair
+        builder = CertificateBuilder(generate_id())
+        builder.build(subject_alt_names=[], validity_days=3)
+
+        cert.key_data = builder.private_key
+        cert.certificate_data = builder.certificate
+        cert.save()
+        cert.refresh_from_db()
+
+        # Kid should be regenerated for the new key
+        self.assertNotEqual(cert.kid, original_kid)
+        self.assertEqual(cert.kid, generate_key_id(cert.key_data))
+
+    def test_kid_regenerated_on_key_change_from_legacy(self):
+        """Test that kid is regenerated from legacy MD5 when key_data changes"""
+        cert = create_test_cert()
+
+        # Simulate a legacy MD5 kid
+        legacy_kid = generate_key_id_legacy(cert.key_data)
+        CertificateKeyPair.objects.filter(pk=cert.pk).update(kid=legacy_kid)
+        cert.refresh_from_db()
+        self.assertEqual(cert.kid, legacy_kid)
+
+        # Generate a new key and update the keypair
+        builder = CertificateBuilder(generate_id())
+        builder.build(subject_alt_names=[], validity_days=3)
+
+        cert.key_data = builder.private_key
+        cert.certificate_data = builder.certificate
+        cert.save()
+        cert.refresh_from_db()
+
+        # Kid should now be SHA512 for the new key
+        self.assertNotEqual(cert.kid, legacy_kid)
+        self.assertEqual(cert.kid, generate_key_id(cert.key_data))

authentik/endpoints/models.py

@@ -175,7 +175,7 @@ class Connector(ScheduledModel, SerializerModel):
         ]
 
 
-class DeviceAccessGroup(PolicyBindingModel):
+class DeviceAccessGroup(SerializerModel, PolicyBindingModel):
 
     name = models.TextField(unique=True)
 

authentik/enterprise/reports/api/reports.py

@@ -0,0 +1,129 @@
+from django.contrib.contenttypes.models import ContentType
+from django.db.models import QuerySet
+from django.urls import reverse
+from drf_spectacular.utils import extend_schema
+from rest_framework import mixins
+from rest_framework.decorators import action
+from rest_framework.fields import CharField
+from rest_framework.permissions import BasePermission
+from rest_framework.request import Request
+from rest_framework.response import Response
+from rest_framework.viewsets import GenericViewSet
+
+from authentik.core.api.utils import ModelSerializer
+from authentik.core.models import User
+from authentik.enterprise.api import EnterpriseRequiredMixin
+from authentik.enterprise.reports.models import DataExport
+from authentik.enterprise.reports.tasks import generate_export
+from authentik.rbac.permissions import HasPermission
+
+
+class RequestedBySerializer(ModelSerializer):
+    class Meta:
+        model = User
+        fields = ("pk", "username")
+
+
+class ContentTypeSerializer(ModelSerializer):
+    app_label = CharField(read_only=True)
+    model = CharField(read_only=True)
+
+    class Meta:
+        model = ContentType
+        fields = ("id", "app_label", "model")
+
+
+class DataExportSerializer(EnterpriseRequiredMixin, ModelSerializer):
+    requested_by = RequestedBySerializer(read_only=True)
+    content_type = ContentTypeSerializer(read_only=True)
+
+    class Meta:
+        model = DataExport
+        fields = (
+            "id",
+            "requested_by",
+            "requested_on",
+            "content_type",
+            "query_params",
+            "file_url",
+            "completed",
+        )
+        read_only_fields = (
+            "id",
+            "requested_by",
+            "requested_on",
+            "content_type",
+            "file_url",
+            "completed",
+        )
+
+
+class DataExportViewSet(
+    mixins.RetrieveModelMixin, mixins.DestroyModelMixin, mixins.ListModelMixin, GenericViewSet
+):
+    queryset = DataExport.objects.all()
+    serializer_class = DataExportSerializer
+    owner_field = "requested_by"
+    ordering_fields = ["completed", "requested_by", "requested_on", "content_type__model"]
+    ordering = ["-requested_on"]
+    search_fields = ["requested_by__username", "content_type__model"]
+
+    def get_queryset(self) -> QuerySet[DataExport]:
+        """Limit to exports of content types the user has view permission on"""
+        qs = super().get_queryset()
+        permitted_cts = []
+        for ct in ContentType.objects.filter(
+            id__in=qs.values_list("content_type_id", flat=True).distinct()
+        ):
+            model = ct.model_class()
+            if model is None:
+                continue
+            perm = f"{ct.app_label}.view_{ct.model}"
+            if self.request.user.has_perm(perm):
+                permitted_cts.append(ct)
+        return qs.filter(content_type__in=permitted_cts)
+
+
+class ExportMixin:
+    @extend_schema(
+        request=None,
+        parameters=[],
+        responses={201: DataExportSerializer},
+        filters=True,
+    )
+    @action(
+        detail=False,
+        methods=["POST"],
+        permission_classes=[HasPermission("authentik_reports.add_dataexport")],
+    )
+    def export(self: GenericViewSet, request: Request) -> Response:
+        """
+        Create a data export for this data type. Note that the export is generated asynchronously:
+        this method returns a `DataExport` object that will initially have `completed=false` as well
+        as the permanent URL to that object in the `Location` header.
+        You can poll that URL until `completed=true`, at which point the `file_url` property will
+        contain a URL to download
+        """
+
+        s = DataExportSerializer(data={"query_params": request.query_params.dict()})
+        s.is_valid(raise_exception=True)
+        export = s.save(
+            requested_by=request.user,
+            content_type=ContentType.objects.get_for_model(self.queryset.model),
+        )
+        generate_export.send(export.id)
+
+        set = export.serializer(instance=export)
+
+        return Response(
+            set.data,
+            status=201,
+            headers={"Location": reverse("authentik_api:dataexport-detail", args=[export.id])},
+        )
+
+    def get_permissions(self: GenericViewSet) -> list[BasePermission]:
+        perms = super().get_permissions()
+        if self.action == "export":
+            model = self.get_queryset().model
+            perms.append(HasPermission(f"{model._meta.app_label}.view_{model._meta.model_name}")())
+        return perms

authentik/enterprise/reports/apps.py

@@ -0,0 +1,8 @@
+from authentik.enterprise.apps import EnterpriseConfig
+
+
+class ReportsConfig(EnterpriseConfig):
+    name = "authentik.enterprise.reports"
+    label = "authentik_reports"
+    verbose_name = "authentik Enterprise.Reports"
+    default = True

authentik/enterprise/reports/migrations/0001_initial.py

@@ -0,0 +1,48 @@
+# Generated by Django 5.2.8 on 2025-12-02 17:19
+
+import authentik.admin.files.fields
+import django.db.models.deletion
+import uuid
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+        ("contenttypes", "0002_remove_content_type_name"),
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="DataExport",
+            fields=[
+                ("id", models.UUIDField(default=uuid.uuid4, primary_key=True, serialize=False)),
+                ("requested_on", models.DateTimeField(auto_now_add=True)),
+                ("query_params", models.JSONField()),
+                ("file", authentik.admin.files.fields.FileField(blank=True)),
+                ("completed", models.BooleanField(default=False)),
+                (
+                    "content_type",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE, to="contenttypes.contenttype"
+                    ),
+                ),
+                (
+                    "requested_by",
+                    models.ForeignKey(
+                        null=True,
+                        on_delete=django.db.models.deletion.SET_NULL,
+                        to=settings.AUTH_USER_MODEL,
+                    ),
+                ),
+            ],
+            options={
+                "verbose_name": "Data Export",
+                "verbose_name_plural": "Data Exports",
+            },
+        ),
+    ]

authentik/enterprise/reports/models.py

@@ -0,0 +1,123 @@
+import csv
+import io
+from uuid import uuid4
+
+from django.contrib.contenttypes.models import ContentType
+from django.db import models
+from django.utils.translation import gettext as _
+from rest_framework.serializers import Serializer
+from rest_framework.viewsets import ModelViewSet
+
+from authentik.admin.files.fields import FileField
+from authentik.admin.files.manager import get_file_manager
+from authentik.admin.files.usage import FileUsage
+from authentik.core.models import User
+from authentik.enterprise.reports.utils import MockRequest
+from authentik.events.models import Event, EventAction, Notification, NotificationSeverity
+from authentik.lib.models import SerializerModel
+from authentik.lib.utils.db import chunked_queryset
+from authentik.tenants.utils import get_current_tenant
+
+
+class DataExport(SerializerModel):
+    id = models.UUIDField(primary_key=True, default=uuid4)
+    requested_by = models.ForeignKey(User, null=True, on_delete=models.SET_NULL)
+    requested_on = models.DateTimeField(auto_now_add=True)
+    content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE)
+    query_params = models.JSONField()
+    file = FileField(blank=True)
+    completed = models.BooleanField(default=False)
+
+    class Meta:
+        verbose_name = _("Data Export")
+        verbose_name_plural = _("Data Exports")
+
+    @property
+    def serializer(self) -> type[Serializer]:
+        """Get serializer for this model"""
+        from authentik.enterprise.reports.api.reports import DataExportSerializer
+
+        return DataExportSerializer
+
+    def generate(self) -> None:
+        if self.completed:
+            raise AssertionError("Data export must only be generated once")
+
+        model_class = self.content_type.model_class()
+        model_verbose_name = model_class._meta.verbose_name
+        model_verbose_name_plural = model_class._meta.verbose_name_plural
+
+        queryset = chunked_queryset(self.get_queryset())
+
+        serializer = self.get_serializer_class()(
+            context={"request": self._get_request()}, instance=queryset, many=True
+        )
+        self.file = f"{model_verbose_name_plural.lower()}_{self.id}.csv"
+
+        with get_file_manager(FileUsage.REPORTS).save_file_stream(self.file) as f:
+            with io.TextIOWrapper(f, encoding="utf-8", newline="") as text:
+                writer = csv.writer(text)
+                fields = [field.label for field in serializer.child.fields.values()]
+                writer.writerow(fields)
+                for record in queryset:
+                    data = serializer.child.to_representation(record).values()
+                    writer.writerow(data)
+        self.completed = True
+        self.save()
+
+        message = _(f"{model_verbose_name} export generated successfully")
+        e = Event.new(
+            EventAction.EXPORT_READY,
+            message=message,
+            export=self,
+        ).set_user(self.requested_by)
+        e.save()
+        Notification.objects.create(
+            event=e,
+            severity=NotificationSeverity.NOTICE,
+            body=message,
+            hyperlink=self.file_url,
+            hyperlink_label=_("Download"),
+            user=self.requested_by,
+        )
+
+    @property
+    def file_url(self) -> str:
+        return get_file_manager(FileUsage.REPORTS).file_url(self.file)
+
+    def _get_request(self) -> MockRequest:
+        return MockRequest(
+            user=self.requested_by, query_params=self.query_params, tenant=get_current_tenant()
+        )
+
+    def get_queryset(self) -> models.QuerySet:
+        request = self._get_request()
+        viewset = self.get_viewset()
+        viewset.request = request
+        queryset = viewset.get_queryset()
+        queryset = viewset.filter_queryset(queryset)
+
+        return queryset
+
+    def get_viewset(self) -> ModelViewSet:
+        from authentik.core.api.users import UserViewSet
+        from authentik.events.api.events import EventViewSet
+
+        model = (self.content_type.app_label, self.content_type.model)
+        if model == ("authentik_core", "user"):
+            return UserViewSet()
+        elif model == ("authentik_events", "event"):
+            return EventViewSet()
+        raise NotImplementedError(f"Unsupported data export type {self.content_type.model}")
+
+    def get_serializer_class(self) -> type[Serializer]:
+        from authentik.enterprise.reports.serializers import (
+            ExportEventSerializer,
+            ExportUserSerializer,
+        )
+
+        if self.content_type.model == "user":
+            return ExportUserSerializer
+        elif self.content_type.model == "event":
+            return ExportEventSerializer
+        return self.get_viewset().get_serializer_class()

authentik/enterprise/reports/serializers.py

@@ -0,0 +1,32 @@
+from rest_framework.fields import CharField, IntegerField, SerializerMethodField
+
+from authentik.core.api.users import UserSerializer
+from authentik.core.models import User
+from authentik.events.api.events import EventSerializer
+
+
+class ExportUserSerializer(UserSerializer):
+    """Serializer for exporting users"""
+
+    groups = SerializerMethodField(source="get_groups")
+
+    def get_groups(self, instance: User) -> str:
+        return ",".join([group.name for group in instance.ak_groups.all()])
+
+    class Meta(UserSerializer.Meta):
+        fields = [f for f in UserSerializer.Meta.fields if f != "groups_obj"] + ["groups"]
+
+
+class ExportEventSerializer(EventSerializer):
+    """Serializer for exporting events"""
+
+    user_pk = IntegerField(source="user.pk", read_only=True)
+    username = CharField(source="user.username", read_only=True)
+    email = CharField(source="user.email", read_only=True)
+
+    class Meta(EventSerializer.Meta):
+        fields = [f for f in EventSerializer.Meta.fields if f != "user"] + [
+            "user_pk",
+            "username",
+            "email",
+        ]

authentik/enterprise/reports/tasks.py

@@ -0,0 +1,10 @@
+from django.utils.translation import gettext_lazy as _
+from dramatiq import actor
+
+from authentik.enterprise.reports.models import DataExport
+
+
+@actor(description=_("Generate data export."))
+def generate_export(export_id: int):
+    export = DataExport.objects.get(id=export_id)
+    export.generate()

authentik/enterprise/reports/tests/test_api.py

@@ -0,0 +1,53 @@
+from django.contrib.contenttypes.models import ContentType
+from django.urls import reverse
+from rest_framework.test import APITestCase
+
+from authentik.core.models import User
+from authentik.core.tests.utils import create_test_admin_user
+from authentik.enterprise.reports.tests.utils import patch_license
+from authentik.events.models import Event
+
+
+@patch_license
+class TestExportAPI(APITestCase):
+    def setUp(self) -> None:
+        self.user = create_test_admin_user()
+        self.client.force_login(self.user)
+
+    def test_create_user_export(self):
+        """Test User export endpoint"""
+        response = self.client.post(
+            reverse("authentik_api:user-export"),
+        )
+        self.assertEqual(response.status_code, 201)
+        self.assertEqual(
+            response.headers["Location"],
+            reverse("authentik_api:dataexport-detail", kwargs={"pk": response.data["id"]}),
+        )
+        self.assertEqual(response.data["requested_by"]["pk"], self.user.pk)
+        self.assertEqual(response.data["completed"], False)
+        self.assertEqual(response.data["file_url"], "")
+        self.assertEqual(response.data["query_params"], {})
+        self.assertEqual(
+            response.data["content_type"]["id"],
+            ContentType.objects.get_for_model(User).id,
+        )
+
+    def test_create_event_export(self):
+        """Test Event export endpoint"""
+        response = self.client.post(
+            reverse("authentik_api:event-export"),
+        )
+        self.assertEqual(response.status_code, 201)
+        self.assertEqual(
+            response.headers["Location"],
+            reverse("authentik_api:dataexport-detail", kwargs={"pk": response.data["id"]}),
+        )
+        self.assertEqual(response.data["requested_by"]["pk"], self.user.pk)
+        self.assertEqual(response.data["completed"], False)
+        self.assertEqual(response.data["file_url"], "")
+        self.assertEqual(response.data["query_params"], {})
+        self.assertEqual(
+            response.data["content_type"]["id"],
+            ContentType.objects.get_for_model(Event).id,
+        )

authentik/enterprise/reports/tests/test_event_export.py

@@ -0,0 +1,29 @@
+from django.contrib.contenttypes.models import ContentType
+from django.test.testcases import TestCase
+
+from authentik.core.tests.utils import create_test_user
+from authentik.enterprise.reports.models import DataExport
+from authentik.enterprise.reports.tests.utils import patch_license
+from authentik.events.models import Event, EventAction
+
+
+@patch_license
+class TestEventExport(TestCase):
+    def setUp(self) -> None:
+        self.user = create_test_user()
+        self.user.assign_perms_to_managed_role("authentik_events.view_event")
+
+        self.e1 = Event.new(EventAction.LOGIN, user=self.user)
+        self.e1.save()
+        self.e2 = Event.new(EventAction.LOGIN_FAILED, user=self.user)
+        self.e2.save()
+
+    def test_type_filter(self):
+        export = DataExport.objects.create(
+            content_type=ContentType.objects.get_for_model(Event),
+            requested_by=self.user,
+            query_params={"actions": [EventAction.LOGIN]},
+        )
+        records = list(export.get_queryset())
+        self.assertEqual(len(records), 1)
+        self.assertEqual(records[0], self.e1)

authentik/enterprise/reports/tests/test_permissions.py

@@ -0,0 +1,80 @@
+from django.urls import reverse
+from rest_framework.test import APITestCase
+
+from authentik.core.tests.utils import create_test_user
+from authentik.enterprise.reports.tests.utils import patch_license
+
+
+@patch_license
+class TestExportPermissions(APITestCase):
+    def setUp(self) -> None:
+        self.user = create_test_user()
+        self.client.force_login(self.user)
+
+    def test_export_without_permission(self):
+        """Test User export endpoint without permission"""
+        response = self.client.post(reverse("authentik_api:user-export"))
+        self.assertEqual(response.status_code, 403)
+
+    def test_export_only_user_permission(self):
+        """Test User export endpoint with only view_user permission"""
+        self.user.assign_perms_to_managed_role("authentik_core.view_user")
+        response = self.client.post(reverse("authentik_api:user-export"))
+        self.assertEqual(response.status_code, 403)
+
+    def test_export_with_permission(self):
+        """Test User export endpoint with view_user and add_dataexport permission"""
+        self.user.assign_perms_to_managed_role("authentik_core.view_user")
+        self.user.assign_perms_to_managed_role("authentik_reports.add_dataexport")
+        response = self.client.post(reverse("authentik_api:user-export"))
+        self.assertEqual(response.status_code, 201)
+
+    def test_export_access(self):
+        """Test that data export access is restricted to the user who created it"""
+        self.user.assign_perms_to_managed_role("authentik_core.view_user")
+        self.user.assign_perms_to_managed_role("authentik_reports.add_dataexport")
+        response = self.client.post(reverse("authentik_api:user-export"))
+        self.assertEqual(response.status_code, 201)
+        export_url = reverse("authentik_api:dataexport-detail", kwargs={"pk": response.data["id"]})
+        response = self.client.get(export_url)
+        self.assertEqual(response.status_code, 200)
+        other_user = create_test_user()
+        other_user.assign_perms_to_managed_role("authentik_core.view_user")
+        other_user.assign_perms_to_managed_role("authentik_reports.add_dataexport")
+        self.client.logout()
+        self.client.force_login(other_user)
+        response = self.client.get(export_url)
+        self.assertEqual(response.status_code, 404)
+
+    def test_export_access_no_datatype_permission(self):
+        """Test that data export access requires view permission on the data type"""
+        self.user.assign_perms_to_managed_role("authentik_core.view_user")
+        self.user.assign_perms_to_managed_role("authentik_reports.add_dataexport")
+        self.user.assign_perms_to_managed_role("authentik_reports.view_dataexport")
+        response = self.client.post(reverse("authentik_api:user-export"))
+        self.assertEqual(response.status_code, 201)
+        export_url = reverse("authentik_api:dataexport-detail", kwargs={"pk": response.data["id"]})
+
+        response = self.client.get(export_url)
+        self.assertEqual(response.status_code, 200)
+
+        self.user.remove_perms_from_managed_role("authentik_core.view_user")
+        response = self.client.get(export_url)
+        self.assertEqual(response.status_code, 404)
+
+        response = self.client.get(reverse("authentik_api:dataexport-list"))
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(len(response.data["results"]), 0)
+
+    def test_export_access_owner(self):
+        self.user.assign_perms_to_managed_role("authentik_core.view_user")
+        self.user.assign_perms_to_managed_role("authentik_reports.add_dataexport")
+        response = self.client.post(reverse("authentik_api:user-export"))
+        self.assertEqual(response.status_code, 201)
+        export_url = reverse("authentik_api:dataexport-detail", kwargs={"pk": response.data["id"]})
+        response = self.client.get(export_url)
+        self.assertEqual(response.status_code, 200)
+
+        self.user.remove_perms_from_managed_role("authentik_core.view_user")
+        response = self.client.get(export_url)
+        self.assertEqual(response.status_code, 404)

authentik/enterprise/reports/tests/test_schema.py

@@ -0,0 +1,48 @@
+from django.test.testcases import TestCase
+from drf_spectacular.generators import SchemaGenerator
+
+from authentik.enterprise.reports.tests.utils import patch_license
+
+
+@patch_license
+class TestSchemaMatch(TestCase):
+    def setUp(self) -> None:
+        generator = SchemaGenerator()
+        self.schema = generator.get_schema(request=None, public=True)
+
+    def _index_params_by_name(self, parameters):
+        result = {}
+        for p in parameters or []:
+            if p.get("in") != "query":
+                continue
+            schema = p.get("schema", {})
+            result[p["name"]] = {
+                "required": p.get("required", False),
+                "type": schema.get("type"),
+                "format": schema.get("format"),
+                "enum": tuple(schema.get("enum", [])),
+            }
+        return result
+
+    def _find_operation_by_operation_id(self, operation_id):
+        for path_item in self.schema.get("paths", {}).values():
+            for operation in path_item.values():
+                if isinstance(operation, dict) and operation.get("operationId") == operation_id:
+                    return operation
+        raise AssertionError(f"operationId '{operation_id}' not found in schema")
+
+    def _get_op_params(self, operation_id):
+        operation = self._find_operation_by_operation_id(operation_id)
+        return self._index_params_by_name(operation.get("parameters", []))
+
+    def test_user_export_action_query_params_match_list(self):
+        list_params = self._get_op_params("core_users_list")
+        del list_params["include_groups"]  # Not applicable for export
+        del list_params["include_roles"]  # Not applicable for export
+        export_params = self._get_op_params("core_users_export_create")
+        self.assertDictEqual(list_params, export_params)
+
+    def test_event_export_action_query_params_match_list(self):
+        list_params = self._get_op_params("events_events_list")
+        export_params = self._get_op_params("events_events_export_create")
+        self.assertDictEqual(list_params, export_params)

authentik/enterprise/reports/tests/test_user_export.py

@@ -0,0 +1,75 @@
+import csv
+
+from django.contrib.contenttypes.models import ContentType
+from django.test.testcases import TestCase
+
+from authentik.admin.files.tests.utils import FileTestFileBackendMixin
+from authentik.core.models import User
+from authentik.core.tests.utils import create_test_user
+from authentik.enterprise.reports.models import DataExport
+from authentik.enterprise.reports.tests.utils import patch_license
+
+
+@patch_license
+class TestUserExport(FileTestFileBackendMixin, TestCase):
+    def setUp(self) -> None:
+        super().setUp()
+
+        self.u1 = create_test_user(username="a")
+        self.u1.assign_perms_to_managed_role("authentik_core.view_user")
+        self.u2 = create_test_user(username="b", path="abcd")
+        self.u1.assign_perms_to_managed_role("authentik_core.view_user")
+
+    def _read_export(self, filename):
+        with open(f"{self.reports_backend_path}/reports/public/{filename}") as f:
+            reader = csv.DictReader(f)
+            return list(reader)
+
+    def test_generate_user_export(self):
+        export = DataExport.objects.create(
+            content_type=ContentType.objects.get_for_model(User),
+            requested_by=self.u1,
+            query_params={"email": str(self.u1.email)},
+        )
+        export.generate()
+
+        self.assertEqual(export.completed, True)
+        data = self._read_export(export.file)
+        self.assertEqual(len(data), 1)
+        self.assertEqual(data[0]["Username"], self.u1.username)
+
+    def test_path_filter(self):
+        export = DataExport.objects.create(
+            content_type=ContentType.objects.get_for_model(User),
+            requested_by=self.u1,
+            query_params={"path": str(self.u2.path)},
+        )
+        records = list(export.get_queryset())
+        self.assertEqual(len(records), 1)
+        self.assertEqual(records[0], self.u2)
+
+    def test_search_filter(self):
+        export = DataExport.objects.create(
+            content_type=ContentType.objects.get_for_model(User),
+            requested_by=self.u1,
+            query_params={"search": f'username = "{self.u2.username}"'},
+        )
+        records = list(export.get_queryset())
+        self.assertEqual(len(records), 1)
+        self.assertEqual(records[0], self.u2)
+
+    def test_ordering(self):
+        export = DataExport.objects.create(
+            content_type=ContentType.objects.get_for_model(User),
+            requested_by=self.u1,
+            query_params={"ordering": "-username"},
+        )
+        records = list(export.get_queryset())
+        self.assertGreaterEqual(records[0].username, records[-1].username)
+        export = DataExport.objects.create(
+            content_type=ContentType.objects.get_for_model(User),
+            requested_by=self.u1,
+            query_params={"ordering": "username"},
+        )
+        records = list(export.get_queryset())
+        self.assertLess(records[0].username, records[-1].username)

authentik/enterprise/reports/tests/utils.py

@@ -0,0 +1,6 @@
+from unittest.mock import MagicMock, patch
+
+patch_license = patch(
+    "authentik.enterprise.models.LicenseUsageStatus.is_valid",
+    MagicMock(return_value=True),
+)

authentik/enterprise/reports/urls.py

@@ -0,0 +1,7 @@
+"""API URLs"""
+
+from authentik.enterprise.reports.api.reports import DataExportViewSet
+
+api_urlpatterns = [
+    ("reports/exports", DataExportViewSet),
+]

authentik/enterprise/reports/utils.py

@@ -0,0 +1,11 @@
+from dataclasses import dataclass
+
+from authentik.core.models import User
+from authentik.tenants.models import Tenant
+
+
+@dataclass
+class MockRequest:
+    user: User
+    query_params: dict[str, str]
+    tenant: Tenant

authentik/enterprise/settings.py

@@ -9,6 +9,7 @@ TENANT_APPS = [
     "authentik.enterprise.providers.radius",
     "authentik.enterprise.providers.scim",
     "authentik.enterprise.providers.ssf",
+    "authentik.enterprise.reports",
     "authentik.enterprise.search",
     "authentik.enterprise.stages.authenticator_endpoint_gdtc",
     "authentik.enterprise.stages.mtls",

authentik/events/api/events.py

@@ -21,6 +21,7 @@ from rest_framework.viewsets import ModelViewSet
 from authentik.core.api.object_types import TypeCreateSerializer
 from authentik.core.api.utils import ModelSerializer, PassiveSerializer
 from authentik.events.models import Event, EventAction
+from authentik.lib.utils.reflection import ConditionalInheritance
 
 
 class EventVolumeSerializer(PassiveSerializer):
@@ -116,7 +117,9 @@ class EventsFilter(django_filters.FilterSet):
         fields = ["action", "client_ip", "username"]
 
 
-class EventViewSet(ModelViewSet):
+class EventViewSet(
+    ConditionalInheritance("authentik.enterprise.reports.api.reports.ExportMixin"), ModelViewSet
+):
     """Event Read-Only Viewset"""
 
     queryset = Event.objects.all()

authentik/events/api/notifications.py

@@ -29,6 +29,8 @@ class NotificationSerializer(ModelSerializer):
             "pk",
             "severity",
             "body",
+            "hyperlink",
+            "hyperlink_label",
             "created",
             "event",
             "seen",

authentik/events/migrations/0014_notification_hyperlink_notification_hyperlink_label_and_more.py

@@ -0,0 +1,59 @@
+# Generated by Django 5.2.9 on 2025-12-05 10:24
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("authentik_events", "0013_delete_systemtask"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="notification",
+            name="hyperlink",
+            field=models.TextField(blank=True, max_length=4096, null=True),
+        ),
+        migrations.AddField(
+            model_name="notification",
+            name="hyperlink_label",
+            field=models.TextField(blank=True, null=True),
+        ),
+        migrations.AlterField(
+            model_name="event",
+            name="action",
+            field=models.TextField(
+                choices=[
+                    ("login", "Login"),
+                    ("login_failed", "Login Failed"),
+                    ("logout", "Logout"),
+                    ("user_write", "User Write"),
+                    ("suspicious_request", "Suspicious Request"),
+                    ("password_set", "Password Set"),
+                    ("secret_view", "Secret View"),
+                    ("secret_rotate", "Secret Rotate"),
+                    ("invitation_used", "Invite Used"),
+                    ("authorize_application", "Authorize Application"),
+                    ("source_linked", "Source Linked"),
+                    ("impersonation_started", "Impersonation Started"),
+                    ("impersonation_ended", "Impersonation Ended"),
+                    ("flow_execution", "Flow Execution"),
+                    ("policy_execution", "Policy Execution"),
+                    ("policy_exception", "Policy Exception"),
+                    ("property_mapping_exception", "Property Mapping Exception"),
+                    ("system_task_execution", "System Task Execution"),
+                    ("system_task_exception", "System Task Exception"),
+                    ("system_exception", "System Exception"),
+                    ("configuration_error", "Configuration Error"),
+                    ("model_created", "Model Created"),
+                    ("model_updated", "Model Updated"),
+                    ("model_deleted", "Model Deleted"),
+                    ("email_sent", "Email Sent"),
+                    ("update_available", "Update Available"),
+                    ("export_ready", "Export Ready"),
+                    ("custom_", "Custom Prefix"),
+                ]
+            ),
+        ),
+    ]

authentik/events/models.py

@@ -117,6 +117,8 @@ class EventAction(models.TextChoices):
 
     UPDATE_AVAILABLE = "update_available"
 
+    EXPORT_READY = "export_ready"
+
     CUSTOM_PREFIX = "custom_"
 
 
@@ -261,6 +263,14 @@ class Event(SerializerModel, ExpiringModel):
             return self.context["message"]
         return f"{self.action}: {self.context}"
 
+    @property
+    def hyperlink(self) -> str | None:
+        return self.context.get("hyperlink")
+
+    @property
+    def hyperlink_label(self) -> str | None:
+        return self.context.get("hyperlink_label")
+
     def __str__(self) -> str:
         return f"Event action={self.action} user={self.user} context={self.context}"
 
@@ -479,6 +489,11 @@ class NotificationTransport(TasksModel, SerializerModel):
             context["key_value"]["event_user_username"] = notification.event.user.get(
                 "username", None
             )
+        if notification.hyperlink:
+            context["link"] = {
+                "target": notification.hyperlink,
+                "label": notification.hyperlink_label,
+            }
         if notification.event:
             context["title"] += notification.event.action
             for key, value in notification.event.context.items():
@@ -532,6 +547,8 @@ class Notification(SerializerModel):
     uuid = models.UUIDField(primary_key=True, editable=False, default=uuid4)
     severity = models.TextField(choices=NotificationSeverity.choices)
     body = models.TextField()
+    hyperlink = models.TextField(blank=True, null=True, max_length=4096)
+    hyperlink_label = models.TextField(blank=True, null=True)
     created = models.DateTimeField(auto_now_add=True)
     event = models.ForeignKey(Event, on_delete=models.SET_NULL, null=True, blank=True)
     seen = models.BooleanField(default=False)

authentik/events/tasks.py

@@ -110,7 +110,12 @@ def notification_transport(transport_pk: int, event_pk: str, user_pk: int, trigg
     if not trigger:
         return
     notification = Notification(
-        severity=trigger.severity, body=event.summary, event=event, user=user
+        severity=trigger.severity,
+        body=event.summary,
+        event=event,
+        user=user,
+        hyperlink=event.hyperlink,
+        hyperlink_label=event.hyperlink_label,
     )
     transport: NotificationTransport = NotificationTransport.objects.filter(pk=transport_pk).first()
     if not transport:

authentik/flows/api/stages.py

@@ -27,7 +27,7 @@ class StageSerializer(ModelSerializer, MetaNameSerializer):
     """Stage Serializer"""
 
     component = SerializerMethodField()
-    flow_set = FlowSetSerializer(many=True, required=False)
+    flow_set = FlowSetSerializer(many=True, required=False, read_only=True)
 
     def to_representation(self, instance: Stage):
         if isinstance(instance, Stage) and instance.is_in_memory:

authentik/flows/auth.py

@@ -0,0 +1,18 @@
+from typing import cast
+
+from django.contrib.auth.models import AnonymousUser
+from rest_framework.authentication import BaseAuthentication
+from rest_framework.request import Request
+
+from authentik.flows.planner import PLAN_CONTEXT_PENDING_USER, FlowPlan
+from authentik.flows.views.executor import SESSION_KEY_PLAN
+
+
+class FlowActive(BaseAuthentication):
+    """Authenticate requests when a flow is currently active"""
+
+    def authenticate(self, request: Request):
+        plan = cast(FlowPlan | None, request.session.get(SESSION_KEY_PLAN))
+        if not plan:
+            return None
+        return (plan.context.get(PLAN_CONTEXT_PENDING_USER, AnonymousUser()), plan)

authentik/flows/stage.py

@@ -249,7 +249,7 @@ class ChallengeStageView(StageView):
                 "f(ch): invalid challenge response",
                 errors=challenge_response.errors,
             )
-        return HttpChallengeResponse(challenge_response)
+        return HttpChallengeResponse(challenge_response, status=400)
 
 
 class AccessDeniedStage(ChallengeStageView):

authentik/flows/tests/__init__.py

@@ -48,6 +48,9 @@ class FlowTestCase(APITestCase):
             self.assertEqual(raw_response[key], expected)
         return raw_response
 
+    def get_flow_plan(self) -> FlowPlan | None:
+        return self.client.session.get(SESSION_KEY_PLAN)
+
     def assertStageRedirects(self, response: HttpResponse, to: str) -> dict[str, Any]:
         """Wrapper around assertStageResponse that checks for a redirect"""
         return self.assertStageResponse(response, component="xak-flow-redirect", to=to)

authentik/flows/tests/test_inspector.py

@@ -56,6 +56,7 @@ class TestFlowInspector(APITestCase):
                     "layout": "stacked",
                 },
                 "flow_designation": "authentication",
+                "passkey_challenge": None,
                 "password_fields": False,
                 "primary_action": "Log in",
                 "sources": [],

authentik/flows/views/executor.py

@@ -147,6 +147,8 @@ class FlowExecutorView(APIView):
                 token.delete()
         if not isinstance(plan, FlowPlan):
             return None
+        if existing_plan := self.request.session[SESSION_KEY_PLAN]:
+            plan.context.update(existing_plan.context)
         plan.context[PLAN_CONTEXT_IS_RESTORED] = token
         self._logger.debug("f(exec): restored flow plan from token", plan=plan)
         return plan
@@ -256,6 +258,11 @@ class FlowExecutorView(APIView):
                 serializers=challenge_types,
                 resource_type_field_name="component",
             ),
+            400: PolymorphicProxySerializer(
+                component_name="ChallengeTypes",
+                serializers=challenge_types,
+                resource_type_field_name="component",
+            ),
         },
         request=OpenApiTypes.NONE,
         parameters=[
@@ -303,6 +310,11 @@ class FlowExecutorView(APIView):
                 serializers=challenge_types,
                 resource_type_field_name="component",
             ),
+            400: PolymorphicProxySerializer(
+                component_name="ChallengeTypes",
+                serializers=challenge_types,
+                resource_type_field_name="component",
+            ),
         },
         request=PolymorphicProxySerializer(
             component_name="FlowChallengeResponse",

authentik/lib/config.py

@@ -301,8 +301,6 @@ class ConfigLoader:
             return {}
         try:
             b64decoded_str = base64.b64decode(config_value).decode("utf-8")
-            b64decoded_str = b64decoded_str.strip().lstrip("{").rstrip("}")
-            b64decoded_str = "{" + b64decoded_str + "}"
             return json.loads(b64decoded_str)
         except (JSONDecodeError, TypeError, ValueError) as exc:
             self.log(

authentik/lib/expression/evaluator.py

@@ -63,6 +63,7 @@ class BaseEvaluator:
             "ak_call_policy": self.expr_func_call_policy,
             "ak_create_event": self.expr_event_create,
             "ak_create_jwt": self.expr_create_jwt,
+            "ak_create_jwt_raw": self.expr_create_jwt_raw,
             "ak_is_group_member": BaseEvaluator.expr_is_group_member,
             "ak_logger": get_logger(self._filename).bind(),
             "ak_send_email": self.expr_send_email,
@@ -222,6 +223,16 @@ class BaseEvaluator:
         access_token.save()
         return access_token.token
 
+    def expr_create_jwt_raw(
+        self, provider: OAuth2Provider | str, validity: str = "seconds=60", **kwargs
+    ) -> str:
+        """Issue a JWT for a given provider with completely customized data"""
+        if not isinstance(provider, OAuth2Provider):
+            provider = OAuth2Provider.objects.get(name=provider)
+        kwargs["exp"] = int((now() + timedelta_from_string(validity)).timestamp())
+        kwargs["aud"] = provider.client_id
+        return provider.encode(kwargs)
+
     def expr_send_email(
         self,
         address: str | list[str],

authentik/lib/sync/incoming/models.py

@@ -0,0 +1,25 @@
+from django.db import models
+from django.utils.translation import gettext_lazy as _
+
+from authentik.core.models import Source
+from authentik.tasks.schedules.models import ScheduledModel
+
+
+class SyncOutgoingTriggerMode(models.TextChoices):
+    # Do not trigger outgoing syncs
+    NONE = "none"
+    # Trigger immediately after object changed
+    IMMEDIATE = "immediate"
+    # Trigger at the end of full sync
+    DEFERRED_END = "deferred_end"
+
+
+class IncomingSyncSource(ScheduledModel, Source):
+    sync_outgoing_trigger_mode = models.TextField(
+        choices=SyncOutgoingTriggerMode.choices,
+        default=SyncOutgoingTriggerMode.DEFERRED_END,
+        help_text=_("When to trigger sync for outgoing providers"),
+    )
+
+    class Meta:
+        abstract = True

authentik/lib/sync/outgoing/models.py

@@ -83,6 +83,10 @@ class OutgoingSyncProvider(ScheduledModel, Model):
     def sync_actor(self) -> Actor:
         raise NotImplementedError
 
+    def sync_dispatch(self) -> None:
+        for schedule in self.schedules:
+            schedule.send()
+
     @property
     def schedule_specs(self) -> list[ScheduleSpec]:
         return [

authentik/lib/sync/outgoing/signals.py

@@ -1,3 +1,6 @@
+from contextlib import contextmanager
+from contextvars import ContextVar
+
 from django.db.models import Model
 from django.db.models.signals import m2m_changed, post_save, pre_delete
 from dramatiq.actor import Actor
@@ -7,6 +10,23 @@ from authentik.lib.sync.outgoing.base import Direction
 from authentik.lib.sync.outgoing.models import OutgoingSyncProvider
 from authentik.lib.utils.reflection import class_to_path
 
+_CTX_INHIBIT_DISPATCH = ContextVar[bool](
+    "authentik_sync_outgoing_inhibit_dispatch",
+    default=False,
+)
+
+
+@contextmanager
+def sync_outgoing_inhibit_dispatch():
+    """
+    Prevent direct and m2m tasks from being dispatched when User/Group/membership change
+    """
+    _CTX_INHIBIT_DISPATCH.set(True)
+    try:
+        yield
+    finally:
+        _CTX_INHIBIT_DISPATCH.set(False)
+
 
 def register_signals(
     provider_type: type[OutgoingSyncProvider],
@@ -28,6 +48,8 @@ def register_signals(
         # This primarily happens during user login
         if sender == User and update_fields == {"last_login"}:
             return
+        if _CTX_INHIBIT_DISPATCH.get():
+            return
         if not provider_type.objects.exists():
             return
         task_sync_direct_dispatch.send(
@@ -41,6 +63,8 @@ def register_signals(
 
     def model_pre_delete(sender: type[Model], instance: User | Group, **_):
         """Pre-delete handler"""
+        if _CTX_INHIBIT_DISPATCH.get():
+            return
         if not provider_type.objects.exists():
             return
         task_sync_direct_dispatch.send(
@@ -58,6 +82,8 @@ def register_signals(
         """Sync group membership"""
         if action not in ["post_add", "post_remove"]:
             return
+        if _CTX_INHIBIT_DISPATCH.get():
+            return
         if not provider_type.objects.exists():
             return
         task_sync_m2m_dispatch.send(instance.pk, action, list(pk_set), reverse)

authentik/lib/tests/test_config.py

@@ -129,7 +129,7 @@ class TestConfig(TestCase):
         test_value = b' "foo": "bar"     '
         b64_value = base64.b64encode(test_value)
         config.set("foo", b64_value)
-        self.assertEqual(config.get_dict_from_b64_json("foo"), {"foo": "bar"})
+        self.assertEqual(config.get_dict_from_b64_json("foo"), {})
 
     def test_get_dict_from_b64_json_invalid(self):
         """Test get_dict_from_b64_json with invalid value"""

authentik/lib/tests/test_evaluator.py

@@ -80,6 +80,26 @@ class TestEvaluator(TestCase):
         )
         self.assertEqual(decoded["preferred_username"], user.username)
 
+    def test_expr_create_jwt_raw(self):
+        """Test expr_create_jwt_raw"""
+        rf = RequestFactory()
+        user = create_test_user()
+        provider = OAuth2Provider.objects.create(
+            name=generate_id(),
+            authorization_flow=create_test_flow(),
+        )
+        evaluator = BaseEvaluator(generate_id())
+        evaluator._context = {
+            "http_request": rf.get(reverse("authentik_core:root-redirect")),
+            "user": user,
+            "provider": provider.name,
+        }
+        jwt = evaluator.evaluate("return ak_create_jwt_raw(provider, foo='bar')")
+        decoded = decode(
+            jwt, provider.client_secret, algorithms=["HS256"], audience=provider.client_id
+        )
+        self.assertEqual(decoded["foo"], "bar")
+
     @patch("authentik.stages.email.tasks.send_mails")
     def test_expr_send_email_with_body(self, mock_send_mails):
         """Test ak_send_email with body parameter"""

authentik/lib/utils/db.py

@@ -26,4 +26,4 @@ def chunked_queryset(queryset: QuerySet, chunk_size: int = 1_000):
     for chunk in get_chunks(queryset):
         reset_queries()
         gc.collect()
-        yield from chunk.iterator()
+        yield from chunk.iterator(chunk_size=chunk_size)

authentik/outposts/models.py

@@ -86,7 +86,7 @@ class OutpostConfig:
 class OutpostModel(Model):
     """Base model for providers that need more objects than just themselves"""
 
-    def get_required_objects(self) -> Iterable[models.Model | str]:
+    def get_required_objects(self) -> Iterable[models.Model | str | tuple[str, models.Model]]:
         """Return a list of all required objects"""
         return [self]
 
@@ -332,41 +332,35 @@ class Outpost(ScheduledModel, SerializerModel, ManagedModel):
         """Create per-object and global permissions for outpost service-account"""
         # To ensure the user only has the correct permissions, we delete all of them and re-add
         # the ones the user needs
-        with transaction.atomic():
-            user.remove_all_perms_from_managed_role()
-            for model_or_perm in self.get_required_objects():
-                if isinstance(model_or_perm, models.Model):
-                    model_or_perm: models.Model
-                    code_name = (
-                        f"{model_or_perm._meta.app_label}.view_{model_or_perm._meta.model_name}"
-                    )
-                    try:
-                        user.assign_perms_to_managed_role(code_name, model_or_perm)
-                    except (Permission.DoesNotExist, AttributeError) as exc:
-                        LOGGER.warning(
-                            "permission doesn't exist",
-                            code_name=code_name,
-                            user=user,
-                            model=model_or_perm,
+        try:
+            with transaction.atomic():
+                user.remove_all_perms_from_managed_role()
+                for model_or_perm in self.get_required_objects():
+                    if isinstance(model_or_perm, models.Model):
+                        code_name = (
+                            f"{model_or_perm._meta.app_label}.view_{model_or_perm._meta.model_name}"
                         )
-                        Event.new(
-                            action=EventAction.SYSTEM_EXCEPTION,
-                            message=(
-                                "While setting the permissions for the service-account, a "
-                                "permission was not found: Check "
-                                "https://docs.goauthentik.io/troubleshooting/missing_permission"
-                            ),
-                        ).with_exception(exc).set_user(user).save()
-                else:
-                    app_label, perm = model_or_perm.split(".")
-                    permission = Permission.objects.filter(
-                        codename=perm,
-                        content_type__app_label=app_label,
-                    )
-                    if not permission.exists():
-                        LOGGER.warning("permission doesn't exist", perm=model_or_perm)
-                        continue
-                    user.assign_perms_to_managed_role(permission.first())
+                        user.assign_perms_to_managed_role(code_name, model_or_perm)
+                    elif isinstance(model_or_perm, tuple):
+                        perm, obj = model_or_perm
+                        user.assign_perms_to_managed_role(perm, obj)
+                    else:
+                        user.assign_perms_to_managed_role(model_or_perm)
+        except (Permission.DoesNotExist, AttributeError) as exc:
+            LOGGER.warning(
+                "permission doesn't exist",
+                code_name=code_name,
+                user=user,
+                model=model_or_perm,
+            )
+            Event.new(
+                action=EventAction.SYSTEM_EXCEPTION,
+                message=(
+                    "While setting the permissions for the service-account, a "
+                    "permission was not found: Check "
+                    "https://docs.goauthentik.io/troubleshooting/missing_permission"
+                ),
+            ).with_exception(exc).set_user(user).save()
         LOGGER.debug(
             "Updated service account's permissions",
             obj_perms=user.get_all_obj_perms_on_managed_role(),
@@ -431,7 +425,7 @@ class Outpost(ScheduledModel, SerializerModel, ManagedModel):
             Token.objects.filter(identifier=self.token_identifier).delete()
             return self.token
 
-    def get_required_objects(self) -> Iterable[models.Model | str]:
+    def get_required_objects(self) -> Iterable[models.Model | str | tuple[str, models.Model]]:
         """Get an iterator of all objects the user needs read access to"""
         objects: list[models.Model | str] = [
             self,
@@ -445,7 +439,9 @@ class Outpost(ScheduledModel, SerializerModel, ManagedModel):
         if self.managed:
             for brand in Brand.objects.filter(web_certificate__isnull=False):
                 objects.append(brand)
-                objects.append(brand.web_certificate)
+                objects.append(("view_certificatekeypair", brand.web_certificate))
+                objects.append(("view_certificatekeypair_certificate", brand.web_certificate))
+                objects.append(("view_certificatekeypair_key", brand.web_certificate))
         return objects
 
     def __str__(self) -> str:

authentik/outposts/tests/test_sa.py

@@ -51,10 +51,12 @@ class OutpostTests(TestCase):
         permissions = outpost.user.get_all_obj_perms_on_managed_role().order_by(
             "content_type__model"
         )
-        self.assertEqual(len(permissions), 3)
+        self.assertEqual(len(permissions), 5)
         self.assertEqual(permissions[0].object_pk, str(keypair.pk))
-        self.assertEqual(permissions[1].object_pk, str(outpost.pk))
-        self.assertEqual(permissions[2].object_pk, str(provider.pk))
+        self.assertEqual(permissions[1].object_pk, str(keypair.pk))
+        self.assertEqual(permissions[2].object_pk, str(keypair.pk))
+        self.assertEqual(permissions[3].object_pk, str(outpost.pk))
+        self.assertEqual(permissions[4].object_pk, str(provider.pk))
 
         # Remove provider from outpost, user should only have access to outpost
         outpost.providers.remove(provider)

authentik/policies/event_matcher/migrations/0024_alter_eventmatcherpolicy_action.py

@@ -0,0 +1,52 @@
+# Generated by Django 5.1.12 on 2025-10-02 12:59
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("authentik_policies_event_matcher", "0023_alter_eventmatcherpolicy_action_and_more"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="eventmatcherpolicy",
+            name="action",
+            field=models.TextField(
+                choices=[
+                    ("login", "Login"),
+                    ("login_failed", "Login Failed"),
+                    ("logout", "Logout"),
+                    ("user_write", "User Write"),
+                    ("suspicious_request", "Suspicious Request"),
+                    ("password_set", "Password Set"),
+                    ("secret_view", "Secret View"),
+                    ("secret_rotate", "Secret Rotate"),
+                    ("invitation_used", "Invite Used"),
+                    ("authorize_application", "Authorize Application"),
+                    ("source_linked", "Source Linked"),
+                    ("impersonation_started", "Impersonation Started"),
+                    ("impersonation_ended", "Impersonation Ended"),
+                    ("flow_execution", "Flow Execution"),
+                    ("policy_execution", "Policy Execution"),
+                    ("policy_exception", "Policy Exception"),
+                    ("property_mapping_exception", "Property Mapping Exception"),
+                    ("system_task_execution", "System Task Execution"),
+                    ("system_task_exception", "System Task Exception"),
+                    ("system_exception", "System Exception"),
+                    ("configuration_error", "Configuration Error"),
+                    ("model_created", "Model Created"),
+                    ("model_updated", "Model Updated"),
+                    ("model_deleted", "Model Deleted"),
+                    ("email_sent", "Email Sent"),
+                    ("update_available", "Update Available"),
+                    ("export_ready", "Export Ready"),
+                    ("custom_", "Custom Prefix"),
+                ],
+                default=None,
+                help_text="Match created events with this action type. When left empty, all action types will be matched.",
+                null=True,
+            ),
+        ),
+    ]

authentik/providers/ldap/models.py

@@ -93,11 +93,13 @@ class LDAPProvider(OutpostModel, BackchannelProvider):
     def __str__(self):
         return f"LDAP Provider {self.name}"
 
-    def get_required_objects(self) -> Iterable[models.Model | str]:
-        required_models = [self, "authentik_core.view_user", "authentik_core.view_group"]
+    def get_required_objects(self) -> Iterable[models.Model | str | tuple[str, models.Model]]:
+        required = [self, "authentik_core.view_user", "authentik_core.view_group"]
         if self.certificate is not None:
-            required_models.append(self.certificate)
-        return required_models
+            required.append(("view_certificatekeypair", self.certificate))
+            required.append(("view_certificatekeypair_certificate", self.certificate))
+            required.append(("view_certificatekeypair_key", self.certificate))
+        return required
 
     class Meta:
         verbose_name = _("LDAP Provider")

authentik/providers/proxy/models.py

@@ -179,11 +179,13 @@ class ProxyProvider(OutpostModel, OAuth2Provider):
     def __str__(self):
         return f"Proxy Provider {self.name}"
 
-    def get_required_objects(self) -> Iterable[models.Model | str]:
-        required_models = [self]
+    def get_required_objects(self) -> Iterable[models.Model | str | tuple[str, models.Model]]:
+        required = [self]
         if self.certificate is not None:
-            required_models.append(self.certificate)
-        return required_models
+            required.append(("view_certificatekeypair", self.certificate))
+            required.append(("view_certificatekeypair_certificate", self.certificate))
+            required.append(("view_certificatekeypair_key", self.certificate))
+        return required
 
     class Meta:
         verbose_name = _("Proxy Provider")

authentik/providers/proxy/tests.py

@@ -1,10 +1,14 @@
 """proxy provider tests"""
 
+from json import loads
+
 from django.urls import reverse
 from rest_framework.test import APITestCase
 
-from authentik.core.tests.utils import create_test_admin_user, create_test_flow
+from authentik.core.models import Application
+from authentik.core.tests.utils import create_test_admin_user, create_test_cert, create_test_flow
 from authentik.lib.generators import generate_id
+from authentik.outposts.models import Outpost, OutpostType
 from authentik.providers.oauth2.models import ClientTypes
 from authentik.providers.proxy.models import ProxyMode, ProxyProvider
 
@@ -127,3 +131,55 @@ class ProxyProviderTests(APITestCase):
         self.assertEqual(response.status_code, 200)
         provider: ProxyProvider = ProxyProvider.objects.get(name=name)
         self.assertEqual(provider.client_type, ClientTypes.CONFIDENTIAL)
+
+    def test_sa_fetch(self):
+        """Test fetching the outpost config as the service account"""
+        outpost = Outpost.objects.create(name=generate_id(), type=OutpostType.PROXY)
+        provider = ProxyProvider.objects.create(name=generate_id())
+        Application.objects.create(name=generate_id(), slug=generate_id(), provider=provider)
+        outpost.providers.add(provider)
+
+        res = self.client.get(
+            reverse("authentik_api:proxyprovideroutpost-list"),
+            HTTP_AUTHORIZATION=f"Bearer {outpost.token.key}",
+        )
+        body = loads(res.content)
+        self.assertEqual(body["pagination"]["count"], 1)
+
+    def test_sa_perms_cert(self):
+        """Test permissions to access a configured certificate"""
+        cert = create_test_cert()
+        outpost = Outpost.objects.create(name=generate_id(), type=OutpostType.PROXY)
+        provider = ProxyProvider.objects.create(name=generate_id(), certificate=cert)
+        Application.objects.create(name=generate_id(), slug=generate_id(), provider=provider)
+        outpost.providers.add(provider)
+
+        res = self.client.get(
+            reverse("authentik_api:proxyprovideroutpost-list"),
+            HTTP_AUTHORIZATION=f"Bearer {outpost.token.key}",
+        )
+        body = loads(res.content)
+        self.assertEqual(body["pagination"]["count"], 1)
+        cert_id = body["results"][0]["certificate"]
+        self.assertEqual(cert_id, str(cert.pk))
+
+        res = self.client.get(
+            reverse(
+                "authentik_api:certificatekeypair-view-certificate",
+                kwargs={
+                    "pk": cert_id,
+                },
+            ),
+            HTTP_AUTHORIZATION=f"Bearer {outpost.token.key}",
+        )
+        self.assertEqual(res.status_code, 200)
+        # res = self.client.get(
+        #     reverse(
+        #         "authentik_api:certificatekeypair-view-private-key",
+        #         kwargs={
+        #             "pk": cert_id,
+        #         },
+        #     ),
+        #     HTTP_AUTHORIZATION=f"Bearer {outpost.token.key}",
+        # )
+        # self.assertEqual(res.status_code, 200)

authentik/providers/radius/models.py

@@ -64,10 +64,12 @@ class RadiusProvider(OutpostModel, Provider):
 
         return RadiusProviderSerializer
 
-    def get_required_objects(self) -> Iterable[models.Model | str]:
+    def get_required_objects(self) -> Iterable[models.Model | str | tuple[str, models.Model]]:
         required = [self, "authentik_stages_mtls.pass_outpost_certificate"]
         if self.certificate is not None:
-            required.append(self.certificate)
+            required.append(("view_certificatekeypair", self.certificate))
+            required.append(("view_certificatekeypair_certificate", self.certificate))
+            required.append(("view_certificatekeypair_key", self.certificate))
         return required
 
     def __str__(self):

authentik/providers/saml/api/providers.py

@@ -313,7 +313,7 @@ class SAMLProviderViewSet(UsedByMixin, ModelViewSet):
             "multipart/form-data": SAMLProviderImportSerializer,
         },
         responses={
-            204: OpenApiResponse(description="Successfully imported provider"),
+            201: SAMLProviderSerializer,
             400: OpenApiResponse(description="Bad request"),
         },
     )
@@ -330,17 +330,18 @@ class SAMLProviderViewSet(UsedByMixin, ModelViewSet):
         file.seek(0)
         try:
             metadata = ServiceProviderMetadataParser().parse(file.read().decode())
-            metadata.to_provider(
+            provider = metadata.to_provider(
                 body.validated_data["name"],
                 body.validated_data["authorization_flow"],
                 body.validated_data["invalidation_flow"],
             )
+            # Return the created provider for use in workflows like the application wizard
+            return Response(SAMLProviderSerializer(provider).data, status=201)
         except ValueError as exc:  # pragma: no cover
             LOGGER.warning(str(exc))
             raise ValidationError(
                 _("Failed to import Metadata: {messages}".format_map({"messages": str(exc)})),
             ) from None
-        return Response(status=204)
 
     @permission_required(
         "authentik_providers_saml.view_samlprovider",

authentik/providers/saml/tests/test_api.py

@@ -145,6 +145,9 @@ class TestSAMLProviderAPI(APITestCase):
 
     def test_import_success(self):
         """Test metadata import (success case)"""
+        name = generate_id()
+        authorization_flow = create_test_flow(FlowDesignation.AUTHORIZATION)
+        invalidation_flow = create_test_flow(FlowDesignation.INVALIDATION)
         with TemporaryFile() as metadata:
             metadata.write(load_fixture("fixtures/simple.xml").encode())
             metadata.seek(0)
@@ -152,14 +155,18 @@ class TestSAMLProviderAPI(APITestCase):
                 reverse("authentik_api:samlprovider-import-metadata"),
                 {
                     "file": metadata,
-                    "name": generate_id(),
-                    "authorization_flow": create_test_flow(FlowDesignation.AUTHORIZATION).pk,
-                    "invalidation_flow": create_test_flow(FlowDesignation.INVALIDATION).pk,
+                    "name": name,
+                    "authorization_flow": authorization_flow.pk,
+                    "invalidation_flow": invalidation_flow.pk,
                 },
                 format="multipart",
             )
-        self.assertEqual(204, response.status_code)
-        # We don't test the actual object being created here, that has its own tests
+        self.assertEqual(201, response.status_code)
+        body = response.json()
+        self.assertIn("pk", body)
+        self.assertEqual(body["name"], name)
+        self.assertEqual(body["authorization_flow"], str(authorization_flow.pk))
+        self.assertEqual(body["invalidation_flow"], str(invalidation_flow.pk))
 
     def test_import_failed(self):
         """Test metadata import (invalid xml)"""

authentik/root/settings.py

@@ -207,7 +207,6 @@ SPECTACULAR_SETTINGS = {
 
 REST_FRAMEWORK = {
     "DEFAULT_PAGINATION_CLASS": "authentik.api.pagination.Pagination",
-    "PAGE_SIZE": 100,
     "DEFAULT_FILTER_BACKENDS": [
         "authentik.rbac.filters.ObjectFilter",
         "django_filters.rest_framework.DjangoFilterBackend",
@@ -260,7 +259,7 @@ MIDDLEWARE_FIRST = [
     "django_prometheus.middleware.PrometheusBeforeMiddleware",
 ]
 MIDDLEWARE = [
-    "django_tenants.middleware.default.DefaultTenantMiddleware",
+    "authentik.tenants.middleware.DefaultTenantMiddleware",
     "authentik.root.middleware.LoggingMiddleware",
     "authentik.root.middleware.ClientIPMiddleware",
     "authentik.stages.user_login.middleware.BoundSessionMiddleware",

authentik/sources/kerberos/api/source.py

@@ -44,6 +44,7 @@ class KerberosSourceSerializer(SourceSerializer):
             "spnego_keytab",
             "spnego_ccache",
             "password_login_update_internal_password",
+            "sync_outgoing_trigger_mode",
         ]
         extra_kwargs = {
             "sync_password": {"write_only": True},

authentik/sources/kerberos/migrations/0004_kerberossource_sync_outgoing_trigger_mode.py

@@ -0,0 +1,26 @@
+# Generated by Django 5.2.9 on 2025-12-08 13:45
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("authentik_sources_kerberos", "0003_migrate_userkerberossourceconnection_identifier"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="kerberossource",
+            name="sync_outgoing_trigger_mode",
+            field=models.TextField(
+                choices=[
+                    ("none", "None"),
+                    ("immediate", "Immediate"),
+                    ("deferred_end", "Deferred End"),
+                ],
+                default="deferred_end",
+                help_text="When to trigger sync for outgoing providers",
+            ),
+        ),
+    ]

authentik/sources/kerberos/models.py

@@ -22,15 +22,14 @@ from structlog.stdlib import get_logger
 from authentik.core.models import (
     GroupSourceConnection,
     PropertyMapping,
-    Source,
     UserSourceConnection,
     UserTypes,
 )
 from authentik.core.types import UILoginButton, UserSettingSerializer
 from authentik.flows.challenge import RedirectChallenge
+from authentik.lib.sync.incoming.models import IncomingSyncSource
 from authentik.lib.utils.time import fqdn_rand
 from authentik.tasks.schedules.common import ScheduleSpec
-from authentik.tasks.schedules.models import ScheduledModel
 
 LOGGER = get_logger()
 
@@ -46,7 +45,7 @@ class KAdminType(models.TextChoices):
     OTHER = "other"
 
 
-class KerberosSource(ScheduledModel, Source):
+class KerberosSource(IncomingSyncSource):
     """Federate Kerberos realm with authentik"""
 
     realm = models.TextField(help_text=_("Kerberos realm"), unique=True)

authentik/sources/kerberos/tasks.py

@@ -6,7 +6,11 @@ from dramatiq.actor import actor
 from structlog.stdlib import get_logger
 
 from authentik.lib.config import CONFIG
+from authentik.lib.sync.incoming.models import SyncOutgoingTriggerMode
 from authentik.lib.sync.outgoing.exceptions import StopSync
+from authentik.lib.sync.outgoing.models import OutgoingSyncProvider
+from authentik.lib.sync.outgoing.signals import sync_outgoing_inhibit_dispatch
+from authentik.lib.utils.reflection import all_subclasses
 from authentik.sources.kerberos.models import KerberosSource
 from authentik.sources.kerberos.sync import KerberosSync
 from authentik.tasks.middleware import CurrentTask
@@ -45,7 +49,15 @@ def kerberos_sync(pk: str):
                 )
                 return
             syncer = KerberosSync(source, self)
-            syncer.sync()
+            if source.sync_outgoing_trigger_mode == SyncOutgoingTriggerMode.IMMEDIATE:
+                syncer.sync()
+            else:
+                with sync_outgoing_inhibit_dispatch():
+                    syncer.sync()
+        if source.sync_outgoing_trigger_mode == SyncOutgoingTriggerMode.DEFERRED_END:
+            for outgoing_sync_provider_cls in all_subclasses(OutgoingSyncProvider):
+                for provider in outgoing_sync_provider_cls.objects.all():
+                    provider.sync_dispatch()
     except StopSync as exc:
         LOGGER.warning("Error syncing kerberos", exc=exc, source=source)
         self.error(exc)

authentik/sources/ldap/api.py

@@ -114,6 +114,7 @@ class LDAPSourceSerializer(SourceSerializer):
             "connectivity",
             "lookup_groups_from_user",
             "delete_not_found_objects",
+            "sync_outgoing_trigger_mode",
         ]
         extra_kwargs = {"bind_password": {"write_only": True}}