initial steps for concurrent execution

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

.github/actions/cherry-pick/action.yml

@@ -179,7 +179,7 @@ runs:
             fi
 
             # Create a unique branch name for the cherry-pick
-            CHERRY_PICK_BRANCH="cherry-pick-${PR_NUMBER}-to-${TARGET_BRANCH}"
+            CHERRY_PICK_BRANCH="cherry-pick/${PR_NUMBER}-to-${TARGET_BRANCH}"
 
             # Check if a cherry-pick PR already exists
             EXISTING_PR=$(gh pr list --head "$CHERRY_PICK_BRANCH" --json number --jq '.[0].number' 2>/dev/null || echo "")
@@ -201,7 +201,7 @@ runs:
               git push origin "$CHERRY_PICK_BRANCH"
 
               # Create PR for the cherry-pick
-              CHERRY_PICK_TITLE="$PR_TITLE (cherry-pick #$PR_NUMBER)"
+              CHERRY_PICK_TITLE="$PR_TITLE (cherry-pick #$PR_NUMBER to $TARGET_BRANCH)"
               CHERRY_PICK_BODY="Cherry-pick of #$PR_NUMBER to \`$TARGET_BRANCH\` branch.
 
         **Original PR:** #$PR_NUMBER
@@ -236,7 +236,7 @@ runs:
               git push origin "$CHERRY_PICK_BRANCH"
 
               # Create PR with conflict notice
-              CONFLICT_TITLE="$PR_TITLE (backport of #$PR_NUMBER)"
+              CONFLICT_TITLE="$PR_TITLE (cherry-pick #$PR_NUMBER to $TARGET_BRANCH)"
               CONFLICT_BODY="⚠️ **This cherry-pick has conflicts that require manual resolution.**
 
         Cherry-pick of #$PR_NUMBER to \`$TARGET_BRANCH\` branch.

.github/actions/setup/action.yml

@@ -8,9 +8,6 @@ inputs:
   postgresql_version:
     description: "Optional postgresql image tag"
     default: "16"
-  profiles:
-    description: "Extra profiles of supporting services to start"
-    default: ""
 
 runs:
   using: "composite"
@@ -58,13 +55,21 @@ runs:
       shell: bash
       run: |
         export PSQL_TAG=${{ inputs.postgresql_version }}
-        export COMPOSE_PROFILES=${{ inputs.profiles }}
         docker compose -f .github/actions/setup/docker-compose.yml up -d
         cd web && npm ci
     - name: Generate config
       if: ${{ contains(inputs.dependencies, 'python') }}
-      shell: bash
-      env:
-        PROFILES: ${{ inputs.profiles }}
+      shell: uv run python {0}
       run: |
-        uv run python3 ${{ github.action_path }}/ci_config.py
+        from authentik.lib.generators import generate_id
+        from yaml import safe_dump
+
+        with open("local.env.yml", "w") as _config:
+            safe_dump(
+                {
+                    "log_level": "debug",
+                    "secret_key": generate_id(),
+                },
+                _config,
+                default_flow_style=False,
+            )

.github/actions/setup/ci_config.py

@@ -1,18 +0,0 @@
-from os import getenv
-from typing import Any
-
-from yaml import safe_dump
-
-from authentik.lib.generators import generate_id
-
-config: dict[str, Any] = {
-    "log_level": "debug",
-    "secret_key": generate_id(),
-}
-
-profiles = getenv("PROFILES")
-if profiles and "postgres_replica" in profiles:
-    config["postgresql"] = {"read_replicas": {"0": {"host": "localhost", "port": 5433}}}
-
-with open("local.env.yml", "w") as _config:
-    safe_dump(config, _config, default_flow_style=False)

.github/actions/setup/docker-compose.yml

@@ -1,17 +1,9 @@
 services:
-  redis:
-    image: docker.io/library/redis:7
-    ports:
-      - 6379:6379
-    restart: always
-
-  postgres:
+  postgresql:
     image: docker.io/library/postgres:${PSQL_TAG:-16}
     volumes:
       - db-data:/var/lib/postgresql/data
-      - ./primary/00-replication.sql:/docker-entrypoint-initdb.d/00-replication.sql
-      - ./primary/01-replication-hba.sh:/docker-entrypoint-initdb.d/01-replication-hba.sh
-    command: postgres -c 'wal_level=replica' -c 'max_wal_senders=10' -c 'max_replication_slots=10' -c 'listen_addresses=*'
+    command: "-c log_statement=all"
     environment:
       POSTGRES_USER: authentik
       POSTGRES_PASSWORD: "EK-5jnKfjrGRm<77"
@@ -19,34 +11,12 @@ services:
     ports:
       - 5432:5432
     restart: always
-    healthcheck:
-      test: ["CMD-SHELL", "pg_isready -U $${POSTGRES_USER}"]
-      interval: 5s
-      timeout: 5s
-      retries: 5
-
-  postgres_replica:
-    profiles:
-      - postgres_replica
-    image: docker.io/library/postgres:${PSQL_TAG:-16}
-    environment:
-      POSTGRES_USER: authentik
-      POSTGRES_PASSWORD: "EK-5jnKfjrGRm<77"
-      POSTGRES_DB: authentik
+  redis:
+    image: docker.io/library/redis:7
     ports:
-      - "5433:5432"
-    volumes:
-      - db-data-replica:/var/lib/postgresql/data
-      - ./replica:/replica
-    command: /replica/start.sh
-    healthcheck:
-      test: ["CMD-SHELL", "pg_isready -U $${POSTGRES_USER}"]
-      interval: 5s
-      timeout: 5s
-      retries: 5
+      - 6379:6379
+    restart: always
 
 volumes:
   db-data:
     driver: local
-  db-data-replica:
-    driver: local

.github/actions/setup/primary/00-replication.sql

@@ -1,9 +0,0 @@
--- Create replication role if it doesn't exist
-DO $$ BEGIN
-    IF NOT EXISTS (SELECT FROM pg_catalog.pg_roles WHERE rolname = 'replica') THEN
-        CREATE ROLE replica WITH REPLICATION LOGIN PASSWORD 'EK-5jnKfjrGRm<77';
-    END IF;
-END $$;
-
--- Create replication slot if it doesn't exist
-SELECT pg_create_physical_replication_slot('replica_slot', true);

.github/actions/setup/primary/01-replication-hba.sh

@@ -1,3 +0,0 @@
-#!/bin/bash
-set -euxo pipefail
-echo "host replication all all scram-sha-256" >> /var/lib/postgresql/data/pg_hba.conf

.github/actions/setup/replica/start.sh

@@ -1,9 +0,0 @@
-#!/bin/bash
-set -euxo pipefail
-echo 'Waiting for primary to be ready...'
-while ! pg_isready -h postgres -p 5432 -U replica; do sleep 1; done;
-echo 'Primary is ready, starting replica...'
-rm -rf /var/lib/postgresql/data/* 2>/dev/null || true
-PGPASSWORD=${POSTGRES_PASSWORD} pg_basebackup -h postgres -U replica -D /var/lib/postgresql/data -Fp -Xs -R -P
-echo 'Replication setup complete, starting PostgreSQL...'
-docker-entrypoint.sh postgres

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

@@ -0,0 +1,28 @@
+name: "Process test results"
+description: Convert test results to JUnit, add them to GitHub Actions and codecov
+
+inputs:
+  flags:
+    description: Codecov flags
+
+runs:
+  using: "composite"
+  steps:
+    - uses: codecov/codecov-action@v5
+      with:
+        flags: ${{ inputs.flags }}
+        use_oidc: true
+    - uses: codecov/test-results-action@v1
+      with:
+        flags: ${{ inputs.flags }}
+        file: unittest.xml
+        use_oidc: true
+    - name: PostgreSQL Logs
+      shell: bash
+      run: |
+        if [[ $ACTIONS_RUNNER_DEBUG == 'true' || $ACTIONS_STEP_DEBUG == 'true' ]]; then
+          docker stop setup-postgresql-1
+          echo "::group::PostgreSQL Logs"
+          docker logs setup-postgresql-1
+          echo "::endgroup::"
+        fi

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

@@ -72,6 +72,13 @@ jobs:
         run: |
           mkdir -p ./gen-ts-api
           mkdir -p ./gen-go-api
+      - name: Setup node
+        if: ${{ !inputs.release }}
+        uses: actions/setup-node@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

.github/workflows/ci-main.yml

@@ -67,6 +67,7 @@ jobs:
       fail-fast: false
       matrix:
         psql:
+          - 15-alpine
           - 16-alpine
           - 17-alpine
         run_id: [1, 2, 3, 4, 5]
@@ -112,8 +113,12 @@ jobs:
           CI_TOTAL_RUNS: "5"
         run: |
           uv run make ci-test
+      - uses: ./.github/actions/test-results
+        if: ${{ always() }}
+        with:
+          flags: unit-migrate
   test-unittest:
-    name: test-unittest - PostgreSQL ${{ matrix.psql }} (${{ matrix.profiles }}) - Run ${{ matrix.run_id }}/5
+    name: test-unittest - PostgreSQL ${{ matrix.psql }} - Run ${{ matrix.run_id }}/5
     runs-on: ubuntu-latest
     timeout-minutes: 20
     needs: test-make-seed
@@ -121,11 +126,9 @@ jobs:
       fail-fast: false
       matrix:
         psql:
+          - 15-alpine
           - 16-alpine
           - 17-alpine
-        profiles:
-          - ""
-          - postgres_replica
         run_id: [1, 2, 3, 4, 5]
     steps:
       - uses: actions/checkout@v5
@@ -133,7 +136,6 @@ jobs:
         uses: ./.github/actions/setup
         with:
           postgresql_version: ${{ matrix.psql }}
-          profiles: ${{ matrix.profiles }}
       - name: run unittest
         env:
           CI_TEST_SEED: ${{ needs.test-make-seed.outputs.seed }}
@@ -141,17 +143,10 @@ jobs:
           CI_TOTAL_RUNS: "5"
         run: |
           uv run make ci-test
-      - if: ${{ always() }}
-        uses: codecov/codecov-action@v5
+      - uses: ./.github/actions/test-results
+        if: ${{ always() }}
         with:
           flags: unit
-          use_oidc: true
-      - if: ${{ !cancelled() }}
-        uses: codecov/test-results-action@v1
-        with:
-          flags: unit
-          file: unittest.xml
-          use_oidc: true
   test-integration:
     runs-on: ubuntu-latest
     timeout-minutes: 30
@@ -165,17 +160,10 @@ jobs:
         run: |
           uv run coverage run manage.py test tests/integration
           uv run coverage xml
-      - if: ${{ always() }}
-        uses: codecov/codecov-action@v5
+      - uses: ./.github/actions/test-results
+        if: ${{ always() }}
         with:
           flags: integration
-          use_oidc: true
-      - if: ${{ !cancelled() }}
-        uses: codecov/test-results-action@v1
-        with:
-          flags: integration
-          file: unittest.xml
-          use_oidc: true
   test-e2e:
     name: test-e2e (${{ matrix.job.name }})
     runs-on: ubuntu-latest
@@ -224,17 +212,10 @@ jobs:
         run: |
           uv run coverage run manage.py test ${{ matrix.job.glob }}
           uv run coverage xml
-      - if: ${{ always() }}
-        uses: codecov/codecov-action@v5
+      - uses: ./.github/actions/test-results
+        if: ${{ always() }}
         with:
           flags: e2e
-          use_oidc: true
-      - if: ${{ !cancelled() }}
-        uses: codecov/test-results-action@v1
-        with:
-          flags: e2e
-          file: unittest.xml
-          use_oidc: true
   ci-core-mark:
     if: always()
     needs:

.github/workflows/translation-advice.yml

@@ -20,14 +20,14 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Find Comment
-        uses: peter-evans/find-comment@v3
+        uses: peter-evans/find-comment@v4
         id: fc
         with:
           issue-number: ${{ github.event.pull_request.number }}
           comment-author: "github-actions[bot]"
           body-includes: authentik translations instructions
       - name: Create or update comment
-        uses: peter-evans/create-or-update-comment@v4
+        uses: peter-evans/create-or-update-comment@v5
         with:
           comment-id: ${{ steps.fc.outputs.comment-id }}
           issue-number: ${{ github.event.pull_request.number }}

CODEOWNERS

@@ -24,6 +24,7 @@ Makefile                                @goauthentik/infrastructure
 .editorconfig                           @goauthentik/infrastructure
 CODEOWNERS                              @goauthentik/infrastructure
 # Backend packages
+packages/django-postgres-cache          @goauthentik/backend
 packages/django-dramatiq-postgres       @goauthentik/backend
 # Web packages
 packages/docusaurus-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-bookworm AS go-builder
+FROM --platform=${BUILDPLATFORM} docker.io/library/golang:1.25.1-bookworm AS go-builder
 
 ARG TARGETOS
 ARG TARGETARCH
@@ -119,7 +119,11 @@ RUN --mount=type=cache,id=apt-$TARGETARCH$TARGETVARIANT,sharing=locked,target=/v
     libltdl-dev && \
     curl https://sh.rustup.rs -sSf | sh -s -- -y
 
-ENV UV_NO_BINARY_PACKAGE="cryptography lxml python-kadmin-rs xmlsec"
+ENV UV_NO_BINARY_PACKAGE="cryptography lxml python-kadmin-rs xmlsec" \
+    # https://github.com/rust-lang/rustup/issues/2949
+    # Fixes issues where the rust version in the build cache is older than latest
+    # and rustup tries to update it, which fails
+    RUSTUP_PERMIT_COPY_RENAME="true"
 
 RUN --mount=type=bind,target=pyproject.toml,src=pyproject.toml \
     --mount=type=bind,target=uv.lock,src=uv.lock \

authentik/admin/signals.py

@@ -0,0 +1,9 @@
+from django.dispatch import receiver
+
+from authentik.admin.tasks import _set_prom_info
+from authentik.root.signals import post_startup
+
+
+@receiver(post_startup)
+def post_startup_admin_metrics(sender, **_):
+    _set_prom_info()

authentik/admin/tasks.py

@@ -2,7 +2,6 @@
 
 from django.core.cache import cache
 from django.utils.translation import gettext_lazy as _
-from django_dramatiq_postgres.middleware import CurrentTask
 from dramatiq import actor
 from packaging.version import parse
 from requests import RequestException
@@ -13,7 +12,7 @@ from authentik.admin.apps import PROM_INFO
 from authentik.events.models import Event, EventAction
 from authentik.lib.config import CONFIG
 from authentik.lib.utils.http import get_http_session
-from authentik.tasks.models import Task
+from authentik.tasks.middleware import CurrentTask
 
 LOGGER = get_logger()
 VERSION_NULL = "0.0.0"
@@ -35,7 +34,7 @@ def _set_prom_info():
 
 @actor(description=_("Update latest version info."))
 def update_latest_version():
-    self: Task = CurrentTask.get_task()
+    self = CurrentTask.get_task()
     if CONFIG.get_bool("disable_update_check"):
         cache.set(VERSION_CACHE_KEY, VERSION_NULL, VERSION_CACHE_TIMEOUT)
         self.info("Version check disabled.")
@@ -72,6 +71,3 @@ def update_latest_version():
     except (RequestException, IndexError) as exc:
         cache.set(VERSION_CACHE_KEY, VERSION_NULL, VERSION_CACHE_TIMEOUT)
         raise exc
-
-
-_set_prom_info()

authentik/api/schema.py

@@ -1,5 +1,8 @@
 """Error Response schema, from https://github.com/axnsan12/drf-yasg/issues/224"""
 
+from collections.abc import Callable
+from typing import Any
+
 from django.utils.translation import gettext_lazy as _
 from drf_spectacular.generators import SchemaGenerator
 from drf_spectacular.plumbing import (
@@ -8,6 +11,7 @@ from drf_spectacular.plumbing import (
     build_basic_type,
     build_object_type,
 )
+from drf_spectacular.renderers import OpenApiJsonRenderer
 from drf_spectacular.settings import spectacular_settings
 from drf_spectacular.types import OpenApiTypes
 from rest_framework.settings import api_settings
@@ -15,34 +19,28 @@ from rest_framework.settings import api_settings
 from authentik.api.apps import AuthentikAPIConfig
 from authentik.api.pagination import PAGINATION_COMPONENT_NAME, PAGINATION_SCHEMA
 
-
-def build_standard_type(obj, **kwargs):
-    """Build a basic type with optional add owns."""
-    schema = build_basic_type(obj)
-    schema.update(kwargs)
-    return schema
-
-
 GENERIC_ERROR = build_object_type(
     description=_("Generic API Error"),
     properties={
-        "detail": build_standard_type(OpenApiTypes.STR),
-        "code": build_standard_type(OpenApiTypes.STR),
+        "detail": build_basic_type(OpenApiTypes.STR),
+        "code": build_basic_type(OpenApiTypes.STR),
     },
     required=["detail"],
 )
 VALIDATION_ERROR = build_object_type(
     description=_("Validation Error"),
     properties={
-        api_settings.NON_FIELD_ERRORS_KEY: build_array_type(build_standard_type(OpenApiTypes.STR)),
-        "code": build_standard_type(OpenApiTypes.STR),
+        api_settings.NON_FIELD_ERRORS_KEY: build_array_type(build_basic_type(OpenApiTypes.STR)),
+        "code": build_basic_type(OpenApiTypes.STR),
     },
     required=[],
     additionalProperties={},
 )
 
 
-def create_component(generator: SchemaGenerator, name, schema, type_=ResolvedComponent.SCHEMA):
+def create_component(
+    generator: SchemaGenerator, name: str, schema: Any, type_=ResolvedComponent.SCHEMA
+) -> ResolvedComponent:
     """Register a component and return a reference to it."""
     component = ResolvedComponent(
         name=name,
@@ -54,7 +52,18 @@ def create_component(generator: SchemaGenerator, name, schema, type_=ResolvedCom
     return component
 
 
-def postprocess_schema_responses(result, generator: SchemaGenerator, **kwargs):
+def preprocess_schema_exclude_non_api(endpoints: list[tuple[str, Any, Any, Callable]], **kwargs):
+    """Filter out all API Views which are not mounted under /api"""
+    return [
+        (path, path_regex, method, callback)
+        for path, path_regex, method, callback in endpoints
+        if path.startswith("/" + AuthentikAPIConfig.mountpoint)
+    ]
+
+
+def postprocess_schema_responses(
+    result: dict[str, Any], generator: SchemaGenerator, **kwargs
+) -> dict[str, Any]:
     """Workaround to set a default response for endpoints.
     Workaround suggested at
     <https://github.com/tfranzel/drf-spectacular/issues/119#issuecomment-656970357>
@@ -104,7 +113,11 @@ def postprocess_schema_responses(result, generator: SchemaGenerator, **kwargs):
     return result
 
 
-def postprocess_schema_pagination(result, generator: SchemaGenerator, **kwargs):
+def postprocess_schema_pagination(
+    result: dict[str, Any], generator: SchemaGenerator, **kwargs
+) -> dict[str, Any]:
+    """Optimise pagination parameters, instead of redeclaring parameters for each endpoint
+    declare them globally and refer to them"""
     to_replace = {
         "ordering": create_component(
             generator,
@@ -157,19 +170,24 @@ def postprocess_schema_pagination(result, generator: SchemaGenerator, **kwargs):
     }
     for path in result["paths"].values():
         for method in path.values():
-            # print(method["parameters"])
             for idx, param in enumerate(method.get("parameters", [])):
                 for replace_name, replace_ref in to_replace.items():
                     if param["name"] == replace_name:
                         method["parameters"][idx] = replace_ref.ref
-            # print(method["parameters"])
     return result
 
 
-def preprocess_schema_exclude_non_api(endpoints, **kwargs):
-    """Filter out all API Views which are not mounted under /api"""
-    return [
-        (path, path_regex, method, callback)
-        for path, path_regex, method, callback in endpoints
-        if path.startswith("/" + AuthentikAPIConfig.mountpoint)
-    ]
+def postprocess_schema_remove_unused(
+    result: dict[str, Any], generator: SchemaGenerator, **kwargs
+) -> dict[str, Any]:
+    """Remove unused components"""
+    # To check if the schema is used, render it to JSON and then substring check that
+    # less efficient than walking through the tree but a lot simpler and no
+    # possibility that we miss something
+    raw = OpenApiJsonRenderer().render(result, renderer_context={}).decode()
+    for key in result["components"][ResolvedComponent.SCHEMA].keys():
+        if raw.count(key) > 1:
+            continue
+        del generator.registry._components[(key, ResolvedComponent.SCHEMA)]
+    result["components"] = generator.registry.build(spectacular_settings.APPEND_COMPONENTS)
+    return result

authentik/blueprints/v1/tasks.py

@@ -12,7 +12,7 @@ from django.db import DatabaseError, InternalError, ProgrammingError
 from django.utils.text import slugify
 from django.utils.timezone import now
 from django.utils.translation import gettext_lazy as _
-from django_dramatiq_postgres.middleware import CurrentTask, CurrentTaskNotFound
+from django_dramatiq_postgres.middleware import CurrentTaskNotFound
 from dramatiq.actor import actor
 from dramatiq.middleware import Middleware
 from structlog.stdlib import get_logger
@@ -39,6 +39,7 @@ from authentik.events.logs import capture_logs
 from authentik.events.utils import sanitize_dict
 from authentik.lib.config import CONFIG
 from authentik.tasks.apps import PRIORITY_HIGH
+from authentik.tasks.middleware import CurrentTask
 from authentik.tasks.models import Task
 from authentik.tasks.schedules.models import Schedule
 from authentik.tenants.models import Tenant
@@ -155,7 +156,7 @@ def blueprints_find() -> list[BlueprintFile]:
     throws=(DatabaseError, ProgrammingError, InternalError),
 )
 def blueprints_discovery(path: str | None = None):
-    self: Task = CurrentTask.get_task()
+    self = CurrentTask.get_task()
     count = 0
     for blueprint in blueprints_find():
         if path and blueprint.path != path:
@@ -195,7 +196,7 @@ def check_blueprint_v1_file(blueprint: BlueprintFile):
 @actor(description=_("Apply single blueprint."))
 def apply_blueprint(instance_pk: UUID):
     try:
-        self: Task = CurrentTask.get_task()
+        self = CurrentTask.get_task()
     except CurrentTaskNotFound:
         self = Task()
     self.set_uid(str(instance_pk))

authentik/core/api/groups.py

@@ -29,8 +29,8 @@ from authentik.rbac.api.roles import RoleSerializer
 from authentik.rbac.decorators import permission_required
 
 
-class GroupMemberSerializer(ModelSerializer):
-    """Stripped down user serializer to show relevant users for groups"""
+class PartialUserSerializer(ModelSerializer):
+    """Partial User Serializer, does not include child relations."""
 
     attributes = JSONDictField(required=False)
     uid = CharField(read_only=True)
@@ -94,11 +94,11 @@ class GroupSerializer(ModelSerializer):
             return True
         return str(request.query_params.get("include_children", "false")).lower() == "true"
 
-    @extend_schema_field(GroupMemberSerializer(many=True))
-    def get_users_obj(self, instance: Group) -> list[GroupMemberSerializer] | None:
+    @extend_schema_field(PartialUserSerializer(many=True))
+    def get_users_obj(self, instance: Group) -> list[PartialUserSerializer] | None:
         if not self._should_include_users:
             return None
-        return GroupMemberSerializer(instance.users, many=True).data
+        return PartialUserSerializer(instance.users, many=True).data
 
     @extend_schema_field(GroupChildSerializer(many=True))
     def get_children_obj(self, instance: Group) -> list[GroupChildSerializer] | None:

authentik/core/api/users.py

@@ -97,8 +97,8 @@ class ParamUserSerializer(PassiveSerializer):
     user = PrimaryKeyRelatedField(queryset=User.objects.all().exclude_anonymous(), required=False)
 
 
-class UserGroupSerializer(ModelSerializer):
-    """Simplified Group Serializer for user's groups"""
+class PartialGroupSerializer(ModelSerializer):
+    """Partial Group Serializer, does not include child relations."""
 
     attributes = JSONDictField(required=False)
     parent_name = CharField(source="parent.name", read_only=True, allow_null=True)
@@ -143,11 +143,11 @@ class UserSerializer(ModelSerializer):
             return True
         return str(request.query_params.get("include_groups", "true")).lower() == "true"
 
-    @extend_schema_field(UserGroupSerializer(many=True))
-    def get_groups_obj(self, instance: User) -> list[UserGroupSerializer] | None:
+    @extend_schema_field(PartialGroupSerializer(many=True))
+    def get_groups_obj(self, instance: User) -> list[PartialGroupSerializer] | None:
         if not self._should_include_groups:
             return None
-        return UserGroupSerializer(instance.ak_groups, many=True).data
+        return PartialGroupSerializer(instance.ak_groups, many=True).data
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
@@ -334,6 +334,21 @@ class UserPasswordSetSerializer(PassiveSerializer):
     password = CharField(required=True)
 
 
+class UserServiceAccountSerializer(PassiveSerializer):
+    """Payload to create a service account"""
+
+    name = CharField(
+        required=True,
+        validators=[UniqueValidator(queryset=User.objects.all().order_by("username"))],
+    )
+    create_group = BooleanField(default=False)
+    expiring = BooleanField(default=True)
+    expires = DateTimeField(
+        required=False,
+        help_text="If not provided, valid for 360 days",
+    )
+
+
 class UsersFilter(FilterSet):
     """Filter for users"""
 
@@ -494,18 +509,7 @@ class UserViewSet(UsedByMixin, ModelViewSet):
 
     @permission_required(None, ["authentik_core.add_user", "authentik_core.add_token"])
     @extend_schema(
-        request=inline_serializer(
-            "UserServiceAccountSerializer",
-            {
-                "name": CharField(required=True),
-                "create_group": BooleanField(default=False),
-                "expiring": BooleanField(default=True),
-                "expires": DateTimeField(
-                    required=False,
-                    help_text="If not provided, valid for 360 days",
-                ),
-            },
-        ),
+        request=UserServiceAccountSerializer,
         responses={
             200: inline_serializer(
                 "UserServiceAccountResponse",
@@ -527,11 +531,12 @@ class UserViewSet(UsedByMixin, ModelViewSet):
     )
     def service_account(self, request: Request) -> Response:
         """Create a new user account that is marked as a service account"""
-        username = request.data.get("name")
-        create_group = request.data.get("create_group", False)
-        expiring = request.data.get("expiring", True)
-        expires = request.data.get("expires", now() + timedelta(days=360))
+        data = UserServiceAccountSerializer(data=request.data)
+        data.is_valid(raise_exception=True)
+        expires = data.validated_data.get("expires", now() + timedelta(days=360))
 
+        username = data.validated_data["name"]
+        expiring = data.validated_data["expiring"]
         with atomic():
             try:
                 user: User = User.objects.create(
@@ -549,10 +554,10 @@ class UserViewSet(UsedByMixin, ModelViewSet):
                     "user_uid": user.uid,
                     "user_pk": user.pk,
                 }
-                if create_group and self.request.user.has_perm("authentik_core.add_group"):
-                    group = Group.objects.create(
-                        name=username,
-                    )
+                if data.validated_data["create_group"] and self.request.user.has_perm(
+                    "authentik_core.add_group"
+                ):
+                    group = Group.objects.create(name=username)
                     group.users.add(user)
                     response["group_pk"] = str(group.pk)
                 token = Token.objects.create(
@@ -565,7 +570,29 @@ class UserViewSet(UsedByMixin, ModelViewSet):
                 response["token"] = token.key
                 return Response(response)
             except IntegrityError as exc:
-                return Response(data={"non_field_errors": [str(exc)]}, status=400)
+                error_msg = str(exc).lower()
+
+                if "unique" in error_msg:
+                    return Response(
+                        data={
+                            "non_field_errors": [
+                                _("A user/group with these details already exists")
+                            ]
+                        },
+                        status=400,
+                    )
+                else:
+                    LOGGER.warning("Service account creation failed", exc=exc)
+                    return Response(
+                        data={"non_field_errors": [_("Unable to create user")]},
+                        status=400,
+                    )
+            except (ValueError, TypeError) as exc:
+                LOGGER.error("Unexpected error during service account creation", exc=exc)
+                return Response(
+                    data={"non_field_errors": [_("Unknown error occurred")]},
+                    status=500,
+                )
 
     @extend_schema(responses={200: SessionUserSerializer(many=False)})
     @action(
@@ -719,7 +746,7 @@ class UserViewSet(UsedByMixin, ModelViewSet):
         return Response(status=204)
 
     @extend_schema(
-        request=OpenApiTypes.NONE,
+        request=None,
         responses={
             "204": OpenApiResponse(description="Successfully ended impersonation"),
         },

authentik/core/migrations/0018_auto_20210330_1345_squashed_0028_alter_token_intent.py

@@ -13,14 +13,6 @@ import authentik.core.models
 import authentik.lib.models
 
 
-def migrate_sessions(apps: Apps, schema_editor: BaseDatabaseSchemaEditor):
-    from django.contrib.sessions.backends.cache import KEY_PREFIX
-    from django.core.cache import cache
-
-    session_keys = cache.keys(KEY_PREFIX + "*")
-    cache.delete_many(session_keys)
-
-
 def fix_duplicates(apps: Apps, schema_editor: BaseDatabaseSchemaEditor):
     db_alias = schema_editor.connection.alias
     Token = apps.get_model("authentik_core", "token")
@@ -151,9 +143,6 @@ class Migration(migrations.Migration):
                 "abstract": False,
             },
         ),
-        migrations.RunPython(
-            code=migrate_sessions,
-        ),
         migrations.AlterField(
             model_name="application",
             name="meta_launch_url",

authentik/core/migrations/0046_session_and_more.py

@@ -7,15 +7,10 @@ from django.contrib.auth import BACKEND_SESSION_KEY, HASH_SESSION_KEY, SESSION_K
 from django.db import migrations, models
 import django.db.models.deletion
 from django.conf import settings
-from django.contrib.sessions.backends.cache import KEY_PREFIX
-from django.utils.timezone import now, timedelta
 from authentik.lib.migrations import progress_bar
 from authentik.root.middleware import ClientIPMiddleware
 
 
-SESSION_CACHE_ALIAS = "default"
-
-
 class PickleSerializer:
     """
     Simple wrapper around pickle to be used in signing.dumps()/loads() and
@@ -83,27 +78,6 @@ def _migrate_session(
         )
 
 
-def migrate_redis_sessions(apps, schema_editor):
-    from django.core.cache import caches
-
-    db_alias = schema_editor.connection.alias
-    cache = caches[SESSION_CACHE_ALIAS]
-
-    # Not a redis cache, skipping
-    if not hasattr(cache, "keys"):
-        return
-
-    print("\nMigrating Redis sessions to database, this might take a couple of minutes...")
-    for key, session_data in progress_bar(cache.get_many(cache.keys(f"{KEY_PREFIX}*")).items()):
-        _migrate_session(
-            apps=apps,
-            db_alias=db_alias,
-            session_key=key.removeprefix(KEY_PREFIX),
-            session_data=session_data,
-            expires=now() + timedelta(seconds=cache.ttl(key)),
-        )
-
-
 def migrate_database_sessions(apps, schema_editor):
     DjangoSession = apps.get_model("sessions", "Session")
     db_alias = schema_editor.connection.alias
@@ -231,10 +205,6 @@ class Migration(migrations.Migration):
                 "verbose_name_plural": "Authenticated Sessions",
             },
         ),
-        migrations.RunPython(
-            code=migrate_redis_sessions,
-            reverse_code=migrations.RunPython.noop,
-        ),
         migrations.RunPython(
             code=migrate_database_sessions,
             reverse_code=migrations.RunPython.noop,

authentik/core/migrations/0051_group_authentik_c_is_supe_1e5a97_idx.py

@@ -0,0 +1,18 @@
+# Generated by Django 5.1.12 on 2025-09-25 13:39
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("authentik_core", "0050_user_last_updated_and_more"),
+        ("authentik_rbac", "0006_alter_role_options"),
+    ]
+
+    operations = [
+        migrations.AddIndex(
+            model_name="group",
+            index=models.Index(fields=["is_superuser"], name="authentik_c_is_supe_1e5a97_idx"),
+        ),
+    ]

authentik/core/models.py

@@ -206,7 +206,10 @@ class Group(SerializerModel, AttributesMixin):
                 "parent",
             ),
         )
-        indexes = [models.Index(fields=["name"])]
+        indexes = (
+            models.Index(fields=["name"]),
+            models.Index(fields=["is_superuser"]),
+        )
         verbose_name = _("Group")
         verbose_name_plural = _("Groups")
         permissions = [
@@ -403,6 +406,8 @@ class User(SerializerModel, GuardianUserMixin, AttributesMixin, AbstractUser):
 
     def locale(self, request: HttpRequest | None = None) -> str:
         """Get the locale the user has configured"""
+        if request and hasattr(request, "LANGUAGE_CODE"):
+            return request.LANGUAGE_CODE
         try:
             return self.attributes.get("settings", {}).get("locale", "")
 

authentik/core/tasks.py

@@ -4,7 +4,7 @@ from datetime import datetime, timedelta
 
 from django.utils.timezone import now
 from django.utils.translation import gettext_lazy as _
-from django_dramatiq_postgres.middleware import CurrentTask
+from django_postgres_cache.tasks import clear_expired_cache
 from dramatiq.actor import actor
 from structlog.stdlib import get_logger
 
@@ -15,14 +15,14 @@ from authentik.core.models import (
     User,
 )
 from authentik.lib.utils.db import chunked_queryset
-from authentik.tasks.models import Task
+from authentik.tasks.middleware import CurrentTask
 
 LOGGER = get_logger()
 
 
 @actor(description=_("Remove expired objects."))
 def clean_expired_models():
-    self: Task = CurrentTask.get_task()
+    self = CurrentTask.get_task()
     for cls in ExpiringModel.__subclasses__():
         cls: ExpiringModel
         objects = (
@@ -33,11 +33,12 @@ def clean_expired_models():
             obj.expire_action()
         LOGGER.debug("Expired models", model=cls, amount=amount)
         self.info(f"Expired {amount} {cls._meta.verbose_name_plural}")
+    clear_expired_cache()
 
 
 @actor(description=_("Remove temporary users created by SAML Sources."))
 def clean_temporary_users():
-    self: Task = CurrentTask.get_task()
+    self = CurrentTask.get_task()
     _now = datetime.now()
     deleted_users = 0
     for user in User.objects.filter(**{f"attributes__{USER_ATTRIBUTE_GENERATED}": True}):

authentik/core/templates/if/admin.html

@@ -8,6 +8,7 @@
 {% endblock %}
 
 {% block body %}
+<ak-skip-to-content></ak-skip-to-content>
 <ak-message-container alignment="bottom"></ak-message-container>
 <ak-interface-admin>
     <ak-loading></ak-loading>

authentik/core/templates/if/user.html

@@ -8,6 +8,7 @@
 {% endblock %}
 
 {% block body %}
+<ak-skip-to-content></ak-skip-to-content>
 <ak-message-container></ak-message-container>
 <ak-interface-user>
     <ak-loading></ak-loading>

authentik/core/templates/login/base_full.html

@@ -45,6 +45,7 @@
 {% block body %}
 <div class="pf-c-background-image">
 </div>
+<ak-skip-to-content></ak-skip-to-content>
 <ak-message-container></ak-message-container>
 <div class="pf-c-login stacked">
     <div class="ak-login-container">

authentik/core/tests/test_users_api.py

@@ -469,3 +469,274 @@ class TestUsersAPI(APITestCase):
         body = loads(response.content)
         self.assertEqual(len(body["results"]), 2)
         self.assertEqual(body["results"][0]["pk"], user.pk)
+
+    def test_service_account_validation_empty_username(self):
+        """Test service account creation with empty/blank username validation"""
+        self.client.force_login(self.admin)
+
+        # Test with empty string
+        response = self.client.post(
+            reverse("authentik_api:user-service-account"),
+            data={
+                "name": "",
+                "create_group": True,
+            },
+        )
+        self.assertEqual(response.status_code, 400)
+        self.assertJSONEqual(
+            response.content,
+            {"name": ["This field may not be blank."]},
+        )
+
+        # Test with only whitespace
+        response = self.client.post(
+            reverse("authentik_api:user-service-account"),
+            data={
+                "name": "   ",
+                "create_group": True,
+            },
+        )
+        self.assertEqual(response.status_code, 400)
+        self.assertJSONEqual(
+            response.content,
+            {"name": ["This field may not be blank."]},
+        )
+
+        # Test with tab and newline characters
+        response = self.client.post(
+            reverse("authentik_api:user-service-account"),
+            data={
+                "name": "\t\n",
+                "create_group": True,
+            },
+        )
+        self.assertEqual(response.status_code, 400)
+        self.assertJSONEqual(
+            response.content,
+            {"name": ["This field may not be blank."]},
+        )
+
+    def test_service_account_validation_valid_username(self):
+        """Test service account creation with valid username"""
+        self.client.force_login(self.admin)
+
+        # Test with valid username
+        response = self.client.post(
+            reverse("authentik_api:user-service-account"),
+            data={
+                "name": "valid-service-account",
+                "create_group": True,
+            },
+        )
+        self.assertEqual(response.status_code, 200)
+
+        # Verify response structure
+        body = loads(response.content)
+        self.assertIn("username", body)
+        self.assertIn("user_uid", body)
+        self.assertIn("user_pk", body)
+        self.assertIn("group_pk", body)  # Should exist since create_group=True
+        self.assertIn("token", body)
+
+        # Verify field types
+        self.assertEqual(body["username"], "valid-service-account")
+        self.assertIsInstance(body["user_pk"], int)
+        self.assertIsInstance(body["user_uid"], str)
+        self.assertIsInstance(body["token"], str)
+        self.assertIsInstance(body["group_pk"], str)
+
+    def test_service_account_validation_without_group(self):
+        """Test service account creation without creating a group"""
+        self.client.force_login(self.admin)
+
+        response = self.client.post(
+            reverse("authentik_api:user-service-account"),
+            data={
+                "name": "no-group-service-account",
+                "create_group": False,
+            },
+        )
+        self.assertEqual(response.status_code, 200)
+
+        body = loads(response.content)
+        self.assertIn("username", body)
+        self.assertIn("user_uid", body)
+        self.assertIn("user_pk", body)
+        self.assertIn("token", body)
+        # Should NOT have group_pk when create_group=False
+        self.assertNotIn("group_pk", body)
+
+    def test_service_account_validation_duplicate_username(self):
+        """Test service account creation with duplicate username"""
+        self.client.force_login(self.admin)
+
+        # Create first service account
+        response = self.client.post(
+            reverse("authentik_api:user-service-account"),
+            data={
+                "name": "duplicate-test",
+                "create_group": True,
+            },
+        )
+        self.assertEqual(response.status_code, 200)
+
+        # Attempt to create second with same username
+        response = self.client.post(
+            reverse("authentik_api:user-service-account"),
+            data={
+                "name": "duplicate-test",
+                "create_group": True,
+            },
+        )
+        self.assertEqual(response.status_code, 400)
+        self.assertJSONEqual(
+            response.content,
+            {"name": ["This field must be unique."]},
+        )
+
+    def test_service_account_validation_invalid_create_group(self):
+        """Test service account creation with invalid create_group field"""
+        self.client.force_login(self.admin)
+
+        # Test with string instead of boolean
+        response = self.client.post(
+            reverse("authentik_api:user-service-account"),
+            data={
+                "name": "test-sa",
+                "create_group": "invalid",
+            },
+        )
+        self.assertEqual(response.status_code, 400)
+        self.assertJSONEqual(
+            response.content,
+            {"create_group": ["Must be a valid boolean."]},
+        )
+
+        # Test with number instead of boolean
+        response = self.client.post(
+            reverse("authentik_api:user-service-account"),
+            data={
+                "name": "test-sa",
+                "create_group": 123,
+            },
+        )
+        self.assertEqual(response.status_code, 400)
+        self.assertJSONEqual(
+            response.content,
+            {"create_group": ["Must be a valid boolean."]},
+        )
+
+    def test_service_account_validation_invalid_expiring(self):
+        """Test service account creation with invalid expiring field"""
+        self.client.force_login(self.admin)
+
+        # Test with string instead of boolean
+        response = self.client.post(
+            reverse("authentik_api:user-service-account"),
+            data={
+                "name": "test-sa",
+                "expiring": "invalid",
+            },
+        )
+        self.assertEqual(response.status_code, 400)
+        self.assertJSONEqual(
+            response.content,
+            {"expiring": ["Must be a valid boolean."]},
+        )
+
+    def test_service_account_validation_invalid_expires(self):
+        """Test service account creation with invalid expires field"""
+        self.client.force_login(self.admin)
+
+        # Test with invalid datetime string
+        response = self.client.post(
+            reverse("authentik_api:user-service-account"),
+            data={
+                "name": "test-sa",
+                "expires": "invalid-datetime",
+            },
+        )
+        self.assertEqual(response.status_code, 400)
+        self.assertJSONEqual(
+            response.content,
+            {
+                "expires": [
+                    "Datetime has wrong format. Use one of these formats instead: "
+                    "YYYY-MM-DDThh:mm[:ss[.uuuuuu]][+HH:MM|-HH:MM|Z]."
+                ]
+            },
+        )
+
+        # Test with invalid format
+        response = self.client.post(
+            reverse("authentik_api:user-service-account"),
+            data={
+                "name": "test-sa",
+                "expires": "2024-13-45",  # Invalid month/day
+            },
+        )
+        self.assertEqual(response.status_code, 400)
+        self.assertJSONEqual(
+            response.content,
+            {
+                "expires": [
+                    "Datetime has wrong format. Use one of these formats instead: "
+                    "YYYY-MM-DDThh:mm[:ss[.uuuuuu]][+HH:MM|-HH:MM|Z]."
+                ]
+            },
+        )
+
+    def test_service_account_validation_multiple_errors(self):
+        """Test service account creation with multiple validation errors"""
+        self.client.force_login(self.admin)
+
+        response = self.client.post(
+            reverse("authentik_api:user-service-account"),
+            data={
+                "name": "",  # Empty username
+                "create_group": "invalid",  # Invalid boolean
+                "expiring": 123,  # Invalid boolean
+                "expires": "not-a-date",  # Invalid datetime
+            },
+        )
+        self.assertEqual(response.status_code, 400)
+        self.assertJSONEqual(
+            response.content,
+            {
+                "name": ["This field may not be blank."],
+                "create_group": ["Must be a valid boolean."],
+                "expiring": ["Must be a valid boolean."],
+                "expires": [
+                    "Datetime has wrong format. Use one of these formats instead: "
+                    "YYYY-MM-DDThh:mm[:ss[.uuuuuu]][+HH:MM|-HH:MM|Z]."
+                ],
+            },
+        )
+
+    def test_service_account_validation_user_friendly_duplicate_error(self):
+        """Test that duplicate username returns user-friendly error, not database error"""
+        self.client.force_login(self.admin)
+
+        # Create first service account
+        response = self.client.post(
+            reverse("authentik_api:user-service-account"),
+            data={
+                "name": "duplicate-username-test",
+                "create_group": True,
+            },
+        )
+        self.assertEqual(response.status_code, 200)
+
+        # Attempt to create second with same username
+        response = self.client.post(
+            reverse("authentik_api:user-service-account"),
+            data={
+                "name": "duplicate-username-test",
+                "create_group": True,
+            },
+        )
+        self.assertEqual(response.status_code, 400)
+        self.assertJSONEqual(
+            response.content,
+            {"name": ["This field must be unique."]},
+        )

authentik/core/urls.py

@@ -30,6 +30,7 @@ from authentik.flows.views.interface import FlowInterfaceView
 from authentik.root.asgi_middleware import AuthMiddlewareStack
 from authentik.root.messages.consumer import MessageConsumer
 from authentik.root.middleware import ChannelsLoggingMiddleware
+from authentik.tenants.channels import TenantsAwareMiddleware
 
 urlpatterns = [
     path(
@@ -97,7 +98,9 @@ api_urlpatterns = [
 websocket_urlpatterns = [
     path(
         "ws/client/",
-        ChannelsLoggingMiddleware(AuthMiddlewareStack(MessageConsumer.as_asgi())),
+        ChannelsLoggingMiddleware(
+            TenantsAwareMiddleware(AuthMiddlewareStack(MessageConsumer.as_asgi()))
+        ),
     ),
 ]
 

authentik/crypto/models.py

@@ -20,6 +20,11 @@ from authentik.lib.models import CreatedUpdatedModel, SerializerModel
 LOGGER = get_logger()
 
 
+def fingerprint_sha256(cert: Certificate) -> str:
+    """Get SHA256 Fingerprint of certificate"""
+    return hexlify(cert.fingerprint(hashes.SHA256()), ":").decode("utf-8")
+
+
 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."""
@@ -82,7 +87,7 @@ class CertificateKeyPair(SerializerModel, ManagedModel, CreatedUpdatedModel):
     @property
     def fingerprint_sha256(self) -> str:
         """Get SHA256 Fingerprint of certificate_data"""
-        return hexlify(self.certificate.fingerprint(hashes.SHA256()), ":").decode("utf-8")
+        return fingerprint_sha256(self.certificate)
 
     @property
     def fingerprint_sha1(self) -> str:

authentik/crypto/tasks.py

@@ -7,13 +7,12 @@ from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives.serialization import load_pem_private_key
 from cryptography.x509.base import load_pem_x509_certificate
 from django.utils.translation import gettext_lazy as _
-from django_dramatiq_postgres.middleware import CurrentTask
 from dramatiq.actor import actor
 from structlog.stdlib import get_logger
 
 from authentik.crypto.models import CertificateKeyPair
 from authentik.lib.config import CONFIG
-from authentik.tasks.models import Task
+from authentik.tasks.middleware import CurrentTask
 
 LOGGER = get_logger()
 
@@ -38,7 +37,7 @@ def ensure_certificate_valid(body: str):
 
 @actor(description=_("Discover, import and update certificates from the filesystem."))
 def certificate_discovery():
-    self: Task = CurrentTask.get_task()
+    self = CurrentTask.get_task()
     certs = {}
     private_keys = {}
     discovered = 0

authentik/crypto/tests.py

@@ -27,7 +27,7 @@ class TestCrypto(APITestCase):
     def test_model_private(self):
         """Test model private key"""
         cert = CertificateKeyPair.objects.create(
-            name="test",
+            name=generate_id(),
             certificate_data="foo",
             key_data="foo",
         )
@@ -271,7 +271,7 @@ class TestCrypto(APITestCase):
         keypair = create_test_cert()
         provider = OAuth2Provider.objects.create(
             name=generate_id(),
-            client_id="test",
+            client_id=generate_id(),
             client_secret=generate_key(),
             authorization_flow=create_test_flow(),
             redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://localhost")],
@@ -303,7 +303,7 @@ class TestCrypto(APITestCase):
         keypair = create_test_cert()
         OAuth2Provider.objects.create(
             name=generate_id(),
-            client_id="test",
+            client_id=generate_id(),
             client_secret=generate_key(),
             authorization_flow=create_test_flow(),
             redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://localhost")],

authentik/enterprise/policies/unique_password/tasks.py

@@ -1,6 +1,5 @@
 from django.db.models.aggregates import Count
 from django.utils.translation import gettext_lazy as _
-from django_dramatiq_postgres.middleware import CurrentTask
 from dramatiq.actor import actor
 from structlog import get_logger
 
@@ -8,7 +7,7 @@ from authentik.enterprise.policies.unique_password.models import (
     UniquePasswordPolicy,
     UserPasswordHistory,
 )
-from authentik.tasks.models import Task
+from authentik.tasks.middleware import CurrentTask
 
 LOGGER = get_logger()
 
@@ -19,7 +18,7 @@ LOGGER = get_logger()
     )
 )
 def check_and_purge_password_history():
-    self: Task = CurrentTask.get_task()
+    self = CurrentTask.get_task()
 
     if not UniquePasswordPolicy.objects.exists():
         UserPasswordHistory.objects.all().delete()
@@ -39,7 +38,7 @@ def trim_password_histories():
     UniquePasswordPolicy policies.
     """
 
-    self: Task = CurrentTask.get_task()
+    self = CurrentTask.get_task()
 
     # No policy, we'll let the cleanup above do its thing
     if not UniquePasswordPolicy.objects.exists():

authentik/enterprise/providers/google_workspace/api/groups.py

@@ -4,7 +4,7 @@ from rest_framework import mixins
 from rest_framework.viewsets import GenericViewSet
 
 from authentik.core.api.used_by import UsedByMixin
-from authentik.core.api.users import UserGroupSerializer
+from authentik.core.api.users import PartialGroupSerializer
 from authentik.core.api.utils import ModelSerializer
 from authentik.enterprise.providers.google_workspace.models import GoogleWorkspaceProviderGroup
 from authentik.lib.sync.outgoing.api import OutgoingSyncConnectionCreateMixin
@@ -13,7 +13,7 @@ from authentik.lib.sync.outgoing.api import OutgoingSyncConnectionCreateMixin
 class GoogleWorkspaceProviderGroupSerializer(ModelSerializer):
     """GoogleWorkspaceProviderGroup Serializer"""
 
-    group_obj = UserGroupSerializer(source="group", read_only=True)
+    group_obj = PartialGroupSerializer(source="group", read_only=True)
 
     class Meta:
 

authentik/enterprise/providers/google_workspace/api/users.py

@@ -3,7 +3,7 @@
 from rest_framework import mixins
 from rest_framework.viewsets import GenericViewSet
 
-from authentik.core.api.groups import GroupMemberSerializer
+from authentik.core.api.groups import PartialUserSerializer
 from authentik.core.api.used_by import UsedByMixin
 from authentik.core.api.utils import ModelSerializer
 from authentik.enterprise.providers.google_workspace.models import GoogleWorkspaceProviderUser
@@ -13,7 +13,7 @@ from authentik.lib.sync.outgoing.api import OutgoingSyncConnectionCreateMixin
 class GoogleWorkspaceProviderUserSerializer(ModelSerializer):
     """GoogleWorkspaceProviderUser Serializer"""
 
-    user_obj = GroupMemberSerializer(source="user", read_only=True)
+    user_obj = PartialUserSerializer(source="user", read_only=True)
 
     class Meta:
 

authentik/enterprise/providers/microsoft_entra/api/groups.py

@@ -4,7 +4,7 @@ from rest_framework import mixins
 from rest_framework.viewsets import GenericViewSet
 
 from authentik.core.api.used_by import UsedByMixin
-from authentik.core.api.users import UserGroupSerializer
+from authentik.core.api.users import PartialGroupSerializer
 from authentik.core.api.utils import ModelSerializer
 from authentik.enterprise.providers.microsoft_entra.models import MicrosoftEntraProviderGroup
 from authentik.lib.sync.outgoing.api import OutgoingSyncConnectionCreateMixin
@@ -13,7 +13,7 @@ from authentik.lib.sync.outgoing.api import OutgoingSyncConnectionCreateMixin
 class MicrosoftEntraProviderGroupSerializer(ModelSerializer):
     """MicrosoftEntraProviderGroup Serializer"""
 
-    group_obj = UserGroupSerializer(source="group", read_only=True)
+    group_obj = PartialGroupSerializer(source="group", read_only=True)
 
     class Meta:
 

authentik/enterprise/providers/microsoft_entra/api/users.py

@@ -3,7 +3,7 @@
 from rest_framework import mixins
 from rest_framework.viewsets import GenericViewSet
 
-from authentik.core.api.groups import GroupMemberSerializer
+from authentik.core.api.groups import PartialUserSerializer
 from authentik.core.api.used_by import UsedByMixin
 from authentik.core.api.utils import ModelSerializer
 from authentik.enterprise.providers.microsoft_entra.models import MicrosoftEntraProviderUser
@@ -13,7 +13,7 @@ from authentik.lib.sync.outgoing.api import OutgoingSyncConnectionCreateMixin
 class MicrosoftEntraProviderUserSerializer(ModelSerializer):
     """MicrosoftEntraProviderUser Serializer"""
 
-    user_obj = GroupMemberSerializer(source="user", read_only=True)
+    user_obj = PartialUserSerializer(source="user", read_only=True)
 
     class Meta:
 

authentik/enterprise/providers/ssf/tasks.py

@@ -4,7 +4,6 @@ from uuid import UUID
 from django.http import HttpRequest
 from django.utils.timezone import now
 from django.utils.translation import gettext_lazy as _
-from django_dramatiq_postgres.middleware import CurrentTask
 from dramatiq.actor import actor
 from requests.exceptions import RequestException
 from structlog.stdlib import get_logger
@@ -20,7 +19,7 @@ from authentik.enterprise.providers.ssf.models import (
 from authentik.lib.utils.http import get_http_session
 from authentik.lib.utils.time import timedelta_from_string
 from authentik.policies.engine import PolicyEngine
-from authentik.tasks.models import Task
+from authentik.tasks.middleware import CurrentTask
 
 session = get_http_session()
 LOGGER = get_logger()
@@ -74,7 +73,7 @@ def _check_app_access(stream: Stream, event_data: dict) -> bool:
 
 @actor(description=_("Send an SSF event."))
 def send_ssf_event(stream_uuid: UUID, event_data: dict[str, Any]):
-    self: Task = CurrentTask.get_task()
+    self = CurrentTask.get_task()
 
     stream = Stream.objects.filter(pk=stream_uuid).first()
     if not stream:

authentik/enterprise/search/settings.py

@@ -2,6 +2,7 @@ SPECTACULAR_SETTINGS = {
     "POSTPROCESSING_HOOKS": [
         "authentik.api.schema.postprocess_schema_responses",
         "authentik.api.schema.postprocess_schema_pagination",
+        "authentik.api.schema.postprocess_schema_remove_unused",
         "authentik.enterprise.search.schema.postprocess_schema_search_autocomplete",
         "drf_spectacular.hooks.postprocess_schema_enums",
     ],

authentik/enterprise/stages/mtls/stage.py

@@ -7,6 +7,8 @@ from cryptography.x509 import (
     Certificate,
     NameOID,
     ObjectIdentifier,
+    RFC822Name,
+    SubjectAlternativeName,
     UnsupportedGeneralNameType,
     load_pem_x509_certificate,
 )
@@ -15,7 +17,7 @@ from django.utils.translation import gettext_lazy as _
 
 from authentik.brands.models import Brand
 from authentik.core.models import User
-from authentik.crypto.models import CertificateKeyPair
+from authentik.crypto.models import CertificateKeyPair, fingerprint_sha256
 from authentik.enterprise.stages.mtls.models import (
     CertAttributes,
     MutualTLSStage,
@@ -137,7 +139,7 @@ class MTLSStageView(ChallengeStageView):
             case CertAttributes.COMMON_NAME:
                 cert_attr = self.get_cert_attribute(cert, NameOID.COMMON_NAME)
             case CertAttributes.EMAIL:
-                cert_attr = self.get_cert_attribute(cert, NameOID.EMAIL_ADDRESS)
+                cert_attr = self.get_cert_email(cert)
         match stage.user_attribute:
             case UserAttributes.USERNAME:
                 user_attr = "username"
@@ -171,7 +173,7 @@ class MTLSStageView(ChallengeStageView):
         self.executor.plan.context.setdefault(PLAN_CONTEXT_PROMPT, {})
         self.executor.plan.context[PLAN_CONTEXT_PROMPT].update(
             {
-                "email": self.get_cert_attribute(cert, NameOID.EMAIL_ADDRESS),
+                "email": self.get_cert_email(cert),
                 "name": self.get_cert_attribute(cert, NameOID.COMMON_NAME),
             }
         )
@@ -183,6 +185,13 @@ class MTLSStageView(ChallengeStageView):
             return None
         return str(attr[0].value)
 
+    def get_cert_email(self, cert: Certificate) -> str | None:
+        ext = cert.extensions.get_extension_for_class(SubjectAlternativeName)
+        _cert_attr = ext.value.get_values_for_type(RFC822Name)
+        if len(_cert_attr) < 1:
+            return None
+        return str(_cert_attr[0])
+
     def dispatch(self, request, *args, **kwargs):
         stage: MutualTLSStage = self.executor.current_stage
         certs = [
@@ -210,6 +219,7 @@ class MTLSStageView(ChallengeStageView):
         if not cert and stage.mode == TLSMode.OPTIONAL:
             self.logger.info("No certificate given, continuing")
             return self.executor.stage_ok()
+        self.logger.debug("Received certificate", cert=fingerprint_sha256(cert))
         existing_user = self.check_if_user(cert)
         if self.executor.flow.designation == FlowDesignation.ENROLLMENT:
             self.enroll_prepare_user(cert)

authentik/events/tasks.py

@@ -4,7 +4,6 @@ from uuid import UUID
 
 from django.db.models.query_utils import Q
 from django.utils.translation import gettext_lazy as _
-from django_dramatiq_postgres.middleware import CurrentTask
 from dramatiq.actor import actor
 from guardian.shortcuts import get_anonymous_user
 from structlog.stdlib import get_logger
@@ -19,7 +18,7 @@ from authentik.events.models import (
 from authentik.lib.utils.db import chunked_queryset
 from authentik.policies.engine import PolicyEngine
 from authentik.policies.models import PolicyBinding, PolicyEngineMode
-from authentik.tasks.models import Task
+from authentik.tasks.middleware import CurrentTask
 
 LOGGER = get_logger()
 
@@ -38,7 +37,7 @@ def event_trigger_dispatch(event_uuid: UUID):
 )
 def event_trigger_handler(event_uuid: UUID, trigger_name: str):
     """Check if policies attached to NotificationRule match event"""
-    self: Task = CurrentTask.get_task()
+    self = CurrentTask.get_task()
 
     event: Event = Event.objects.filter(event_uuid=event_uuid).first()
     if not event:
@@ -131,7 +130,7 @@ def gdpr_cleanup(user_pk: int):
 @actor(description=_("Cleanup seen notifications and notifications whose event expired."))
 def notification_cleanup():
     """Cleanup seen notifications and notifications whose event expired."""
-    self: Task = CurrentTask.get_task()
+    self = CurrentTask.get_task()
     notifications = Notification.objects.filter(Q(event=None) | Q(seen=True))
     amount = notifications.count()
     notifications.delete()

authentik/flows/api/bindings.py

@@ -46,5 +46,5 @@ class FlowStageBindingViewSet(UsedByMixin, ModelViewSet):
     serializer_class = FlowStageBindingSerializer
     filterset_fields = "__all__"
     search_fields = ["stage__name"]
-    ordering = ["order"]
-    ordering_fields = ["order", "stage__name"]
+    ordering = ["order", "pk"]
+    ordering_fields = ["order", "stage__name", "target__uuid", "pk"]

authentik/flows/challenge.py

@@ -54,6 +54,7 @@ class Challenge(PassiveSerializer):
 
     flow_info = ContextualFlowInfo(required=False)
     component = CharField(default="")
+    xid = CharField(required=False)
 
     response_errors = DictField(
         child=ErrorDetailSerializer(many=True), allow_empty=True, required=False

authentik/flows/planner.py

@@ -143,10 +143,12 @@ class FlowPlan:
         request: HttpRequest,
         flow: Flow,
         allowed_silent_types: list["StageView"] | None = None,
+        **get_params,
     ) -> HttpResponse:
         """Redirect to the flow executor for this flow plan"""
         from authentik.flows.views.executor import (
             SESSION_KEY_PLAN,
+            FlowContainer,
             FlowExecutorView,
         )
 
@@ -157,6 +159,7 @@ class FlowPlan:
             # No unskippable stages found, so we can directly return the response of the last stage
             final_stage: type[StageView] = self.bindings[-1].stage.view
             temp_exec = FlowExecutorView(flow=flow, request=request, plan=self)
+            temp_exec.container = FlowContainer(request)
             temp_exec.current_stage = self.bindings[-1].stage
             temp_exec.current_stage_view = final_stage
             temp_exec.setup(request, flow.slug)
@@ -174,6 +177,9 @@ class FlowPlan:
         ):
             get_qs["inspector"] = "available"
 
+        for key, value in get_params:
+            get_qs[key] = value
+
         return redirect_with_qs(
             "authentik_core:if-flow",
             get_qs,

authentik/flows/stage.py

@@ -192,6 +192,7 @@ class ChallengeStageView(StageView):
                 )
                 flow_info.is_valid()
                 challenge.initial_data["flow_info"] = flow_info.data
+            challenge.initial_data["xid"] = self.executor.container.exec_id
             if isinstance(challenge, WithUserInfoChallenge):
                 # If there's a pending user, update the `username` field
                 # this field is only used by password managers.
@@ -286,6 +287,12 @@ class SessionEndStage(ChallengeStageView):
     that the user is likely to take after signing out of a provider."""
 
     def get_challenge(self, *args, **kwargs) -> Challenge:
+        if not self.request.user.is_authenticated:
+            return RedirectChallenge(
+                data={
+                    "to": reverse("authentik_core:root-redirect"),
+                },
+            )
         application: Application | None = self.executor.plan.context.get(PLAN_CONTEXT_APPLICATION)
         data = {
             "component": "ak-stage-session-end",

authentik/flows/templates/if/flow.html

@@ -27,8 +27,9 @@ window.authentik.flow = {
 {% endblock %}
 
 {% block body %}
+<ak-skip-to-content></ak-skip-to-content>
 <ak-message-container></ak-message-container>
-<ak-flow-executor flowSlug="{{ flow.slug }}">
+<ak-flow-executor flowSlug="{{ flow.slug }}" xid="{{ xid }}">
     <ak-loading></ak-loading>
 </ak-flow-executor>
 {% endblock %}

authentik/flows/views/executor.py

@@ -1,6 +1,7 @@
 """authentik multi-stage authentication engine"""
 
 from copy import deepcopy
+from uuid import uuid4
 
 from django.conf import settings
 from django.contrib.auth.mixins import LoginRequiredMixin
@@ -63,6 +64,7 @@ from authentik.policies.engine import PolicyEngine
 LOGGER = get_logger()
 # Argument used to redirect user after login
 NEXT_ARG_NAME = "next"
+SESSION_KEY_PLAN_CONTAINER = "authentik/flows/plan_container/%s"
 SESSION_KEY_PLAN = "authentik/flows/plan"
 SESSION_KEY_APPLICATION_PRE = "authentik/flows/application_pre"
 SESSION_KEY_GET = "authentik/flows/get"
@@ -70,6 +72,7 @@ SESSION_KEY_POST = "authentik/flows/post"
 SESSION_KEY_HISTORY = "authentik/flows/history"
 QS_KEY_TOKEN = "flow_token"  # nosec
 QS_QUERY = "query"
+QS_EXEC_ID = "xid"
 
 
 def challenge_types():
@@ -96,6 +99,88 @@ class InvalidStageError(SentryIgnoredException):
     """Error raised when a challenge from a stage is not valid"""
 
 
+class FlowContainer:
+    """Allow for multiple concurrent flow executions in the same session"""
+
+    def __init__(self, request: HttpRequest, exec_id: str | None = None) -> None:
+        self.request = request
+        self.exec_id = exec_id
+
+    @staticmethod
+    def new(request: HttpRequest):
+        exec_id = str(uuid4())
+        request.session[SESSION_KEY_PLAN_CONTAINER % exec_id] = {}
+        return FlowContainer(request, exec_id)
+
+    def exists(self) -> bool:
+        """Check if flow exists in container/session"""
+        return SESSION_KEY_PLAN in self.session
+
+    def save(self):
+        self.request.session.modified = True
+
+    @property
+    def session(self):
+        # Backwards compatibility: store session plan/etc directly in session
+        if not self.exec_id:
+            return self.request.session
+        self.request.session.setdefault(SESSION_KEY_PLAN_CONTAINER % self.exec_id, {})
+        return self.request.session.get(SESSION_KEY_PLAN_CONTAINER % self.exec_id, {})
+
+    @property
+    def plan(self) -> FlowPlan:
+        return self.session.get(SESSION_KEY_PLAN)
+
+    def to_redirect(
+        self,
+        request: HttpRequest,
+        flow: Flow,
+        allowed_silent_types: list[StageView] | None = None,
+        **get_params,
+    ) -> HttpResponse:
+        get_params[QS_EXEC_ID] = self.exec_id
+        return self.plan.to_redirect(
+            request, flow, allowed_silent_types=allowed_silent_types, **get_params
+        )
+
+    @plan.setter
+    def plan(self, value: FlowPlan):
+        self.session[SESSION_KEY_PLAN] = value
+        self.request.session.modified = True
+        self.save()
+
+    @property
+    def application_pre(self):
+        return self.session.get(SESSION_KEY_APPLICATION_PRE)
+
+    @property
+    def get(self) -> QueryDict:
+        return self.session.get(SESSION_KEY_GET)
+
+    @get.setter
+    def get(self, value: QueryDict):
+        self.session[SESSION_KEY_GET] = value
+        self.save()
+
+    @property
+    def post(self) -> QueryDict:
+        return self.session.get(SESSION_KEY_POST)
+
+    @post.setter
+    def post(self, value: QueryDict):
+        self.session[SESSION_KEY_POST] = value
+        self.save()
+
+    @property
+    def history(self) -> list[FlowPlan]:
+        return self.session.get(SESSION_KEY_HISTORY)
+
+    @history.setter
+    def history(self, value: list[FlowPlan]):
+        self.session[SESSION_KEY_HISTORY] = value
+        self.save()
+
+
 @method_decorator(xframe_options_sameorigin, name="dispatch")
 class FlowExecutorView(APIView):
     """Flow executor, passing requests to Stage Views"""
@@ -103,8 +188,9 @@ class FlowExecutorView(APIView):
     permission_classes = [AllowAny]
 
     flow: Flow = None
-
     plan: FlowPlan | None = None
+    container: FlowContainer
+
     current_binding: FlowStageBinding | None = None
     current_stage: Stage
     current_stage_view: View
@@ -160,10 +246,12 @@ class FlowExecutorView(APIView):
             if QS_KEY_TOKEN in get_params:
                 plan = self._check_flow_token(get_params[QS_KEY_TOKEN])
                 if plan:
-                    self.request.session[SESSION_KEY_PLAN] = plan
+                    container = FlowContainer.new(request)
+                    container.plan = plan
             # Early check if there's an active Plan for the current session
-            if SESSION_KEY_PLAN in self.request.session:
-                self.plan: FlowPlan = self.request.session[SESSION_KEY_PLAN]
+            self.container = FlowContainer(request, request.GET.get(QS_EXEC_ID))
+            if self.container.exists():
+                self.plan: FlowPlan = self.container.plan
                 if self.plan.flow_pk != self.flow.pk.hex:
                     self._logger.warning(
                         "f(exec): Found existing plan for other flow, deleting plan",
@@ -176,13 +264,14 @@ class FlowExecutorView(APIView):
                     self._logger.debug("f(exec): Continuing existing plan")
 
             # Initial flow request, check if we have an upstream query string passed in
-            request.session[SESSION_KEY_GET] = get_params
+            self.container.get = get_params
             # Don't check session again as we've either already loaded the plan or we need to plan
             if not self.plan:
-                request.session[SESSION_KEY_HISTORY] = []
+                self.container.history = []
                 self._logger.debug("f(exec): No active Plan found, initiating planner")
                 try:
                     self.plan = self._initiate_plan()
+                    self.container.plan = self.plan
                 except FlowNonApplicableException as exc:
                     self._logger.warning("f(exec): Flow not applicable to current user", exc=exc)
                     return self.handle_invalid_flow(exc)
@@ -255,12 +344,19 @@ class FlowExecutorView(APIView):
         request=OpenApiTypes.NONE,
         parameters=[
             OpenApiParameter(
-                name="query",
+                name=QS_QUERY,
                 location=OpenApiParameter.QUERY,
                 required=True,
                 description="Querystring as received",
                 type=OpenApiTypes.STR,
-            )
+            ),
+            OpenApiParameter(
+                name=QS_EXEC_ID,
+                location=OpenApiParameter.QUERY,
+                required=False,
+                description="Flow execution ID",
+                type=OpenApiTypes.STR,
+            ),
         ],
         operation_id="flows_executor_get",
     )
@@ -287,8 +383,8 @@ class FlowExecutorView(APIView):
                 span.set_data("authentik Stage", self.current_stage_view)
                 span.set_data("authentik Flow", self.flow.slug)
                 stage_response = self.current_stage_view.dispatch(request)
-                return to_stage_response(request, stage_response)
-        except Exception as exc:  # noqa
+                return to_stage_response(request, stage_response, self.container.exec_id)
+        except Exception as exc:
             return self.handle_exception(exc)
 
     @extend_schema(
@@ -306,12 +402,19 @@ class FlowExecutorView(APIView):
         ),
         parameters=[
             OpenApiParameter(
-                name="query",
+                name=QS_QUERY,
                 location=OpenApiParameter.QUERY,
                 required=True,
                 description="Querystring as received",
                 type=OpenApiTypes.STR,
-            )
+            ),
+            OpenApiParameter(
+                name=QS_EXEC_ID,
+                location=OpenApiParameter.QUERY,
+                required=True,
+                description="Flow execution ID",
+                type=OpenApiTypes.STR,
+            ),
         ],
         operation_id="flows_executor_solve",
     )
@@ -338,14 +441,15 @@ class FlowExecutorView(APIView):
                 span.set_data("authentik Stage", self.current_stage_view)
                 span.set_data("authentik Flow", self.flow.slug)
                 stage_response = self.current_stage_view.dispatch(request)
-                return to_stage_response(request, stage_response)
+                return to_stage_response(request, stage_response, self.container.exec_id)
         except Exception as exc:  # noqa
             return self.handle_exception(exc)
 
     def _initiate_plan(self) -> FlowPlan:
         planner = FlowPlanner(self.flow)
         plan = planner.plan(self.request)
-        self.request.session[SESSION_KEY_PLAN] = plan
+        container = FlowContainer.new(self.request)
+        container.plan = plan
         try:
             # Call the has_stages getter to check that
             # there are no issues with the class we might've gotten
@@ -369,7 +473,7 @@ class FlowExecutorView(APIView):
         except FlowNonApplicableException as exc:
             self._logger.warning("f(exec): Flow restart not applicable to current user", exc=exc)
             return self.handle_invalid_flow(exc)
-        self.request.session[SESSION_KEY_PLAN] = plan
+        self.container.plan = plan
         kwargs = self.kwargs
         kwargs.update({"flow_slug": self.flow.slug})
         return redirect_with_qs("authentik_api:flow-executor", self.request.GET, **kwargs)
@@ -391,9 +495,13 @@ class FlowExecutorView(APIView):
         )
         self.cancel()
         if next_param and not is_url_absolute(next_param):
-            return to_stage_response(self.request, redirect_with_qs(next_param))
+            return to_stage_response(
+                self.request, redirect_with_qs(next_param), self.container.exec_id
+            )
         return to_stage_response(
-            self.request, self.stage_invalid(error_message=_("Invalid next URL"))
+            self.request,
+            self.stage_invalid(error_message=_("Invalid next URL")),
+            self.container.exec_id,
         )
 
     def stage_ok(self) -> HttpResponse:
@@ -407,7 +515,7 @@ class FlowExecutorView(APIView):
             self.current_stage_view.cleanup()
         self.request.session.get(SESSION_KEY_HISTORY, []).append(deepcopy(self.plan))
         self.plan.pop()
-        self.request.session[SESSION_KEY_PLAN] = self.plan
+        self.container.plan = self.plan
         if self.plan.bindings:
             self._logger.debug(
                 "f(exec): Continuing with next stage",
@@ -450,6 +558,7 @@ class FlowExecutorView(APIView):
 
     def cancel(self):
         """Cancel current flow execution"""
+        # TODO: Clean up container
         keys_to_delete = [
             SESSION_KEY_APPLICATION_PRE,
             SESSION_KEY_PLAN,
@@ -472,8 +581,8 @@ class CancelView(View):
 
     def get(self, request: HttpRequest) -> HttpResponse:
         """View which canels the currently active plan"""
-        if SESSION_KEY_PLAN in request.session:
-            del request.session[SESSION_KEY_PLAN]
+        if FlowContainer(request, request.GET.get(QS_EXEC_ID)).exists():
+            del request.session[SESSION_KEY_PLAN_CONTAINER % request.GET.get(QS_EXEC_ID)]
             LOGGER.debug("Canceled current plan")
         return redirect("authentik_flows:default-invalidation")
 
@@ -521,19 +630,12 @@ class ToDefaultFlow(View):
 
     def dispatch(self, request: HttpRequest) -> HttpResponse:
         flow = self.get_flow()
-        # If user already has a pending plan, clear it so we don't have to later.
-        if SESSION_KEY_PLAN in self.request.session:
-            plan: FlowPlan = self.request.session[SESSION_KEY_PLAN]
-            if plan.flow_pk != flow.pk.hex:
-                LOGGER.warning(
-                    "f(def): Found existing plan for other flow, deleting plan",
-                    flow_slug=flow.slug,
-                )
-                del self.request.session[SESSION_KEY_PLAN]
-        return redirect_with_qs("authentik_core:if-flow", request.GET, flow_slug=flow.slug)
+        get_qs = request.GET.copy()
+        get_qs[QS_EXEC_ID] = str(uuid4())
+        return redirect_with_qs("authentik_core:if-flow", get_qs, flow_slug=flow.slug)
 
 
-def to_stage_response(request: HttpRequest, source: HttpResponse) -> HttpResponse:
+def to_stage_response(request: HttpRequest, source: HttpResponse, xid: str) -> HttpResponse:
     """Convert normal HttpResponse into JSON Response"""
     if (
         isinstance(source, HttpResponseRedirect)
@@ -552,6 +654,7 @@ def to_stage_response(request: HttpRequest, source: HttpResponse) -> HttpRespons
             RedirectChallenge(
                 {
                     "to": str(redirect_url),
+                    "xid": xid,
                 }
             )
         )
@@ -560,6 +663,7 @@ def to_stage_response(request: HttpRequest, source: HttpResponse) -> HttpRespons
             ShellChallenge(
                 {
                     "body": source.render().content.decode("utf-8"),
+                    "xid": xid,
                 }
             )
         )
@@ -569,6 +673,7 @@ def to_stage_response(request: HttpRequest, source: HttpResponse) -> HttpRespons
             ShellChallenge(
                 {
                     "body": source.content.decode("utf-8"),
+                    "xid": xid,
                 }
             )
         )
@@ -600,4 +705,6 @@ class ConfigureFlowInitView(LoginRequiredMixin, View):
         except FlowNonApplicableException:
             LOGGER.warning("Flow not applicable to user")
             raise Http404 from None
-        return plan.to_redirect(request, stage.configure_flow)
+        container = FlowContainer.new(request)
+        container.plan = plan
+        return container.to_redirect(request, stage.configure_flow)

authentik/flows/views/interface.py

@@ -7,6 +7,7 @@ from ua_parser.user_agent_parser import Parse
 
 from authentik.core.views.interface import InterfaceView
 from authentik.flows.models import Flow
+from authentik.flows.views.executor import QS_EXEC_ID
 
 
 class FlowInterfaceView(InterfaceView):
@@ -17,6 +18,7 @@ class FlowInterfaceView(InterfaceView):
         kwargs["flow"] = flow
         kwargs["flow_background_url"] = flow.background_url(self.request)
         kwargs["inspector"] = "inspector" in self.request.GET
+        kwargs["xid"] = self.request.GET.get(QS_EXEC_ID)
         return super().get_context_data(**kwargs)
 
     def compat_needs_sfe(self) -> bool:

authentik/lib/config.py

@@ -444,10 +444,6 @@ def django_db_config(config: ConfigLoader | None = None) -> dict:
             f"postgresql.read_replicas.{replica}.conn_options", default={}
         )
         _database["OPTIONS"].update(replica_conn_options)
-        _database["TEST"] = {
-            "MIRROR": "default",
-            "NAME": config.get("postgresql.test.name"),
-        }
 
         db[f"replica_{replica}"] = _database
     return db

authentik/lib/debug.py

@@ -19,7 +19,7 @@ def start_debug_server(**kwargs) -> bool:
         )
         return False
 
-    listen: str = CONFIG.get("listen.listen_debug_py", "127.0.0.1:9901")
+    listen: str = CONFIG.get("listen.debug_py", "127.0.0.1:9901")
     host, _, port = listen.rpartition(":")
     try:
         debugpy.listen((host, int(port)), **kwargs)  # nosec

authentik/lib/default.yml

@@ -31,14 +31,14 @@ postgresql:
   #   host: replica1.example.com
 
 listen:
-  listen_http: 0.0.0.0:9000
-  listen_https: 0.0.0.0:9443
-  listen_ldap: 0.0.0.0:3389
-  listen_ldaps: 0.0.0.0:6636
-  listen_radius: 0.0.0.0:1812
-  listen_metrics: 0.0.0.0:9300
-  listen_debug: 0.0.0.0:9900
-  listen_debug_py: 0.0.0.0:9901
+  http: 0.0.0.0:9000
+  https: 0.0.0.0:9443
+  ldap: 0.0.0.0:3389
+  ldaps: 0.0.0.0:6636
+  radius: 0.0.0.0:1812
+  metrics: 0.0.0.0:9300
+  debug: 0.0.0.0:9900
+  debug_py: 0.0.0.0:9901
   trusted_proxy_cidrs:
     - 127.0.0.0/8
     - 10.0.0.0/8
@@ -152,7 +152,7 @@ worker:
   processes: 1
   threads: 2
   consumer_listen_timeout: "seconds=30"
-  task_max_retries: 20
+  task_max_retries: 5
   task_default_time_limit: "minutes=10"
   lock_purge_interval: "minutes=1"
   task_purge_interval: "days=1"

authentik/lib/expression/evaluator.py

@@ -3,9 +3,10 @@
 import re
 import socket
 from ipaddress import ip_address, ip_network
+from smtplib import SMTPException
 from textwrap import indent
 from types import CodeType
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from cachetools import TLRUCache, cached
 from django.core.exceptions import FieldError
@@ -29,6 +30,10 @@ from authentik.policies.types import PolicyRequest, PolicyResult
 from authentik.providers.oauth2.id_token import IDToken
 from authentik.providers.oauth2.models import AccessToken, OAuth2Provider
 from authentik.stages.authenticator import devices_for_user
+from authentik.stages.email.utils import TemplateEmailMessage
+
+if TYPE_CHECKING:
+    from authentik.stages.email.models import EmailStage
 
 LOGGER = get_logger()
 
@@ -57,11 +62,12 @@ class BaseEvaluator:
         self._globals = {
             "ak_call_policy": self.expr_func_call_policy,
             "ak_create_event": self.expr_event_create,
+            "ak_create_jwt": self.expr_create_jwt,
             "ak_is_group_member": BaseEvaluator.expr_is_group_member,
             "ak_logger": get_logger(self._filename).bind(),
+            "ak_send_email": self.expr_send_email,
             "ak_user_by": BaseEvaluator.expr_user_by,
             "ak_user_has_authenticator": BaseEvaluator.expr_func_user_has_authenticator,
-            "ak_create_jwt": self.expr_create_jwt,
             "ip_address": ip_address,
             "ip_network": ip_network,
             "list_flatten": BaseEvaluator.expr_flatten,
@@ -216,6 +222,81 @@ class BaseEvaluator:
         access_token.save()
         return access_token.token
 
+    def expr_send_email(
+        self,
+        address: str | list[str],
+        subject: str,
+        body: str | None = None,
+        stage: "EmailStage | None" = None,
+        template: str | None = None,
+        context: dict | None = None,
+    ) -> bool:
+        """Send an email using authentik's email system
+
+        Args:
+            address: Email address(es) to send to. Can be:
+                - Single email: "user@example.com"
+                - List of emails: ["user1@example.com", "user2@example.com"]
+            subject: Email subject
+            body: Email body (plain text/HTML). Mutually exclusive with template.
+            stage: EmailStage instance to use for settings. If None, uses global settings.
+            template: Template name to render. Mutually exclusive with body.
+            context: Additional context variables for template rendering.
+
+        Returns:
+            bool: True if email was queued successfully, False otherwise
+        """
+        # Deferred imports to avoid circular import issues
+        from authentik.stages.email.tasks import send_mails
+
+        if body and template:
+            raise ValueError("body and template parameters are mutually exclusive")
+
+        if not body and not template:
+            raise ValueError("Either body or template parameter must be provided")
+
+        # Normalize address parameter to list of (name, email) tuples
+        if isinstance(address, str):
+            # Single email address
+            to_addresses = [("", address)]
+        elif isinstance(address, list):
+            if not address:
+                raise ValueError("Address list cannot be empty")
+            # List of email strings
+            to_addresses = [("", email) for email in address]
+        else:
+            raise ValueError("Address must be a string or list of strings")
+
+        try:
+            if template is not None:
+                # Use all available context from the evaluator for template rendering
+                template_context = self._context.copy()
+                # Add any custom context passed to the function
+                if context:
+                    template_context.update(context)
+
+                # Use template rendering
+                message = TemplateEmailMessage(
+                    subject=subject,
+                    to=to_addresses,
+                    template_name=template,
+                    template_context=template_context,
+                )
+            else:
+                # Use plain body
+                message = TemplateEmailMessage(
+                    subject=subject,
+                    to=to_addresses,
+                    body=body,
+                )
+
+            send_mails(stage, message)
+            return True
+
+        except (SMTPException, ConnectionError, ValidationError, ValueError) as exc:
+            LOGGER.warning("Failed to send email", exc=exc, addresses=to_addresses, subject=subject)
+            return False
+
     def wrap_expression(self, expression: str) -> str:
         """Wrap expression in a function, call it, and save the result as `result`"""
         handler_signature = ",".join(sanitize_arg(x) for x in self._context.keys())

authentik/lib/logging.py

@@ -112,6 +112,7 @@ def get_logger_config():
         "hpack": "WARNING",
         "httpx": "WARNING",
         "azure": "WARNING",
+        "channels_postgres": "WARNING",
     }
     for handler_name, level in handler_level_map.items():
         base_config["loggers"][handler_name] = {

authentik/lib/sentry.py

@@ -3,19 +3,15 @@
 from asyncio.exceptions import CancelledError
 from typing import Any
 
-from channels_redis.core import ChannelFull
 from django.conf import settings
 from django.core.exceptions import ImproperlyConfigured, SuspiciousOperation, ValidationError
 from django.db import DatabaseError, InternalError, OperationalError, ProgrammingError
 from django.http.response import Http404
-from django_redis.exceptions import ConnectionInterrupted
 from docker.errors import DockerException
 from dramatiq.errors import Retry
 from h11 import LocalProtocolError
 from ldap3.core.exceptions import LDAPException
 from psycopg.errors import Error
-from redis.exceptions import ConnectionError as RedisConnectionError
-from redis.exceptions import RedisError, ResponseError
 from rest_framework.exceptions import APIException
 from sentry_sdk import HttpTransport, get_current_scope
 from sentry_sdk import init as sentry_sdk_init
@@ -23,7 +19,6 @@ from sentry_sdk.api import set_tag
 from sentry_sdk.integrations.argv import ArgvIntegration
 from sentry_sdk.integrations.django import DjangoIntegration
 from sentry_sdk.integrations.dramatiq import DramatiqIntegration
-from sentry_sdk.integrations.redis import RedisIntegration
 from sentry_sdk.integrations.socket import SocketIntegration
 from sentry_sdk.integrations.stdlib import StdlibIntegration
 from sentry_sdk.integrations.threading import ThreadingIntegration
@@ -59,13 +54,7 @@ ignored_classes = (
     ProgrammingError,
     SuspiciousOperation,
     ValidationError,
-    # Redis errors
-    RedisConnectionError,
-    ConnectionInterrupted,
-    RedisError,
-    ResponseError,
     # websocket errors
-    ChannelFull,
     WebSocketException,
     LocalProtocolError,
     # rest_framework error
@@ -112,7 +101,6 @@ def sentry_init(**sentry_init_kwargs):
             ArgvIntegration(),
             DjangoIntegration(transaction_style="function_name", cache_spans=True),
             DramatiqIntegration(),
-            RedisIntegration(),
             SocketIntegration(),
             StdlibIntegration(),
             ThreadingIntegration(propagate_hub=True),
@@ -159,9 +147,7 @@ def before_send(event: dict, hint: dict) -> dict | None:
         if event["logger"] in [
             "asyncio",
             "multiprocessing",
-            "django_redis",
             "django.security.DisallowedHost",
-            "django_redis.cache",
             "paramiko.transport",
         ]:
             return None

authentik/lib/sync/outgoing/tasks.py

@@ -2,7 +2,6 @@ from django.core.paginator import Paginator
 from django.db.models import Model, QuerySet
 from django.db.models.query import Q
 from django.utils.text import slugify
-from django_dramatiq_postgres.middleware import CurrentTask
 from dramatiq.actor import Actor
 from dramatiq.composition import group
 from dramatiq.errors import Retry
@@ -22,6 +21,7 @@ from authentik.lib.sync.outgoing.exceptions import (
 from authentik.lib.sync.outgoing.models import OutgoingSyncProvider
 from authentik.lib.utils.errors import exception_to_dict
 from authentik.lib.utils.reflection import class_to_path, path_to_class
+from authentik.tasks.middleware import CurrentTask
 from authentik.tasks.models import Task
 
 
@@ -61,7 +61,7 @@ class SyncTasks:
         provider_pk: int,
         sync_objects: Actor[[str, int, int, bool], None],
     ):
-        task: Task = CurrentTask.get_task()
+        task = CurrentTask.get_task()
         self.logger = get_logger().bind(
             provider_type=class_to_path(self._provider_model),
             provider_pk=provider_pk,
@@ -118,7 +118,7 @@ class SyncTasks:
         override_dry_run=False,
         **filter,
     ):
-        task: Task = CurrentTask.get_task()
+        task = CurrentTask.get_task()
         _object_type: type[Model] = path_to_class(object_type)
         self.logger = get_logger().bind(
             provider_type=class_to_path(self._provider_model),
@@ -173,7 +173,7 @@ class SyncTasks:
             except TransientSyncException as exc:
                 self.logger.warning("failed to sync object", exc=exc, user=obj)
                 task.warning(
-                    f"Failed to sync {str(obj)} due to " f"transient error: {str(exc)}",
+                    f"Failed to sync {str(obj)} due to transient error: {str(exc)}",
                     obj=sanitize_item(obj),
                     exception=exception_to_dict(exc),
                 )
@@ -207,7 +207,7 @@ class SyncTasks:
         provider_pk: int,
         raw_op: str,
     ):
-        task: Task = CurrentTask.get_task()
+        task = CurrentTask.get_task()
         self.logger = get_logger().bind(
             provider_type=class_to_path(self._provider_model),
         )
@@ -281,7 +281,7 @@ class SyncTasks:
         action: str,
         pk_set: list[int],
     ):
-        task: Task = CurrentTask.get_task()
+        task = CurrentTask.get_task()
         self.logger = get_logger().bind(
             provider_type=class_to_path(self._provider_model),
         )

authentik/lib/tests/test_evaluator.py

@@ -1,5 +1,7 @@
 """Test Evaluator base functions"""
 
+from unittest.mock import patch
+
 from django.test import RequestFactory, TestCase
 from django.urls import reverse
 from jwt import decode
@@ -77,3 +79,163 @@ class TestEvaluator(TestCase):
             jwt, provider.client_secret, algorithms=["HS256"], audience=provider.client_id
         )
         self.assertEqual(decoded["preferred_username"], user.username)
+
+    @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"""
+        user = create_test_user()
+        evaluator = BaseEvaluator(generate_id())
+        evaluator._context = {"user": user}
+
+        # Test sending email with body
+        result = evaluator.evaluate(
+            "return ak_send_email('test@example.com', 'Test Subject', body='Test Body')"
+        )
+
+        self.assertTrue(result)
+        mock_send_mails.assert_called_once()
+
+        # Verify the call arguments - send_mails is called with (stage, message)
+        args, kwargs = mock_send_mails.call_args
+        stage, message = args
+
+        # Check that global settings are used (stage is None)
+        self.assertIsNone(stage)
+
+        # Check message properties
+        self.assertEqual(message.subject, "Test Subject")
+        self.assertEqual(message.to, ["test@example.com"])
+        self.assertEqual(message.body, "Test Body")
+
+    @patch("authentik.stages.email.tasks.send_mails")
+    def test_expr_send_email_with_template(self, mock_send_mails):
+        """Test ak_send_email with template parameter"""
+        user = create_test_user()
+        evaluator = BaseEvaluator(generate_id())
+        evaluator._context = {"user": user}
+
+        # Test sending email with template
+        result = evaluator.evaluate(
+            "return ak_send_email('test@example.com', 'Test Subject', "
+            "template='email/password_reset.html')"
+        )
+
+        self.assertTrue(result)
+        mock_send_mails.assert_called_once()
+
+    def test_expr_send_email_validation_errors(self):
+        """Test ak_send_email validation errors"""
+        evaluator = BaseEvaluator(generate_id())
+
+        # Test error when both body and template are provided
+        with self.assertRaises(ValueError) as cm:
+            evaluator.evaluate(
+                "return ak_send_email('test@example.com', 'Test', "
+                "body='Body', template='template.html')"
+            )
+        self.assertIn("mutually exclusive", str(cm.exception))
+
+        # Test error when neither body nor template are provided
+        with self.assertRaises(ValueError) as cm:
+            evaluator.evaluate("return ak_send_email('test@example.com', 'Test')")
+        self.assertIn("Either body or template parameter must be provided", str(cm.exception))
+
+    @patch("authentik.stages.email.tasks.send_mails")
+    def test_expr_send_email_with_custom_stage(self, mock_send_mails):
+        """Test ak_send_email with custom EmailStage"""
+        from authentik.stages.email.models import EmailStage
+
+        user = create_test_user()
+        custom_stage = EmailStage(
+            name="custom-stage", use_global_settings=False, from_address="custom@example.com"
+        )
+
+        evaluator = BaseEvaluator(generate_id())
+        evaluator._context = {"user": user, "custom_stage": custom_stage}
+
+        # Test sending email with custom stage
+        result = evaluator.evaluate(
+            "return ak_send_email('test@example.com', 'Test Subject', "
+            "body='Test Body', stage=custom_stage)"
+        )
+
+        self.assertTrue(result)
+        mock_send_mails.assert_called_once()
+
+        # Verify the custom stage was used
+        args, kwargs = mock_send_mails.call_args
+        stage, message = args
+
+        self.assertEqual(stage, custom_stage)
+        self.assertFalse(stage.use_global_settings)
+
+    @patch("authentik.stages.email.tasks.send_mails")
+    def test_expr_send_email_with_context(self, mock_send_mails):
+        """Test ak_send_email with custom context parameter"""
+        user = create_test_user()
+        evaluator = BaseEvaluator(generate_id())
+        evaluator._context = {"user": user, "request_id": "123"}
+
+        # Test sending email with template and custom context
+        result = evaluator.evaluate(
+            "return ak_send_email('test@example.com', 'Test Subject', "
+            "template='email/password_reset.html', "
+            "context={'url': 'http://localhost', 'expires': '2026-01-01'})"
+        )
+
+        self.assertTrue(result)
+        mock_send_mails.assert_called_once()
+
+        # Verify the call arguments - send_mails is called with (stage, message)
+        args, kwargs = mock_send_mails.call_args
+        stage, message = args
+
+        # Check that global settings are used (stage is None)
+        self.assertIsNone(stage)
+
+        self.assertEqual(message.subject, "Test Subject")
+        self.assertEqual(message.to, ["test@example.com"])
+        self.assertIn("2026-01-01", message.body)
+        self.assertIn("http://localhost", message.body)
+
+    @patch("authentik.stages.email.tasks.send_mails")
+    def test_expr_send_email_multiple_addresses(self, mock_send_mails):
+        """Test ak_send_email with multiple email addresses"""
+        user = create_test_user()
+        evaluator = BaseEvaluator(generate_id())
+        evaluator._context = {"user": user}
+
+        # Test sending email to multiple addresses
+        result = evaluator.evaluate(
+            "return ak_send_email(['user1@example.com', 'user2@example.com'], "
+            "'Test Subject', body='Test Body')"
+        )
+
+        self.assertTrue(result)
+        mock_send_mails.assert_called_once()
+
+        # Verify the call arguments - send_mails is called with (stage, message)
+        args, kwargs = mock_send_mails.call_args
+        stage, message = args
+
+        # Check that global settings are used (stage is None)
+        self.assertIsNone(stage)
+
+        # Check message properties - should have multiple recipients
+        self.assertEqual(message.subject, "Test Subject")
+        self.assertEqual(message.to, ["user1@example.com", "user2@example.com"])
+        self.assertEqual(message.body, "Test Body")
+
+    def test_expr_send_email_multiple_addresses_validation(self):
+        """Test ak_send_email validation with multiple addresses"""
+        evaluator = BaseEvaluator(generate_id())
+
+        # Test error when empty list is provided
+        with self.assertRaises(ValueError) as cm:
+            evaluator.evaluate("return ak_send_email([], 'Test', body='Body')")
+        self.assertIn("Address list cannot be empty", str(cm.exception))
+
+        # Test error when invalid type is provided
+        with self.assertRaises(ValueError) as cm:
+            evaluator.evaluate("return ak_send_email(123, 'Test', body='Body')")
+        self.assertIn("Address must be a string or list of strings", str(cm.exception))

authentik/lib/utils/errors.py

@@ -4,9 +4,11 @@ from traceback import extract_tb
 
 from structlog.tracebacks import ExceptionDictTransformer
 
+from authentik.lib.config import CONFIG
 from authentik.lib.utils.reflection import class_to_path
 
 TRACEBACK_HEADER = "Traceback (most recent call last):"
+_exception_transformer = ExceptionDictTransformer(show_locals=CONFIG.get_bool("debug"))
 
 
 def exception_to_string(exc: Exception) -> str:
@@ -23,4 +25,4 @@ def exception_to_string(exc: Exception) -> str:
 
 def exception_to_dict(exc: Exception) -> dict:
     """Format exception as a dictionary"""
-    return ExceptionDictTransformer()((type(exc), exc, exc.__traceback__))
+    return _exception_transformer((type(exc), exc, exc.__traceback__))

authentik/outposts/consumer.py

@@ -3,7 +3,9 @@
 from dataclasses import asdict, dataclass, field
 from datetime import datetime
 from enum import IntEnum
+from hashlib import sha256
 from typing import Any
+from uuid import UUID
 
 from asgiref.sync import async_to_sync
 from channels.exceptions import DenyConnection
@@ -18,8 +20,15 @@ from structlog.stdlib import BoundLogger, get_logger
 from authentik.outposts.apps import GAUGE_OUTPOSTS_CONNECTED, GAUGE_OUTPOSTS_LAST_UPDATE
 from authentik.outposts.models import OUTPOST_HELLO_INTERVAL, Outpost, OutpostState
 
-OUTPOST_GROUP = "group_outpost_%(outpost_pk)s"
-OUTPOST_GROUP_INSTANCE = "group_outpost_%(outpost_pk)s_%(instance)s"
+
+def build_outpost_group(outpost_pk: str | UUID) -> str:
+    return sha256(f"{connection.schema_name}/group_outpost_{str(outpost_pk)}".encode()).hexdigest()
+
+
+def build_outpost_group_instance(outpost_pk: str | UUID, instance: str) -> str:
+    return sha256(
+        f"{connection.schema_name}/group_outpost_{str(outpost_pk)}_{instance}".encode()
+    ).hexdigest()
 
 
 class WebsocketMessageInstruction(IntEnum):
@@ -64,26 +73,24 @@ class OutpostConsumer(JsonWebsocketConsumer):
     def connect(self):
         uuid = self.scope["url_route"]["kwargs"]["pk"]
         user = self.scope["user"]
-        outpost = (
+        self.outpost: Outpost | None = (
             get_objects_for_user(user, "authentik_outposts.view_outpost").filter(pk=uuid).first()
         )
-        if not outpost:
+        if self.outpost is None:
             raise DenyConnection()
-        self.logger = self.logger.bind(outpost=outpost)
+        self.logger = self.logger.bind(outpost=self.outpost)
         try:
             self.accept()
         except RuntimeError as exc:
             self.logger.warning("runtime error during accept", exc=exc)
             raise DenyConnection() from None
-        self.outpost = outpost
         query = QueryDict(self.scope["query_string"].decode())
         self.instance_uid = query.get("instance_uuid", self.channel_name)
         async_to_sync(self.channel_layer.group_add)(
-            OUTPOST_GROUP % {"outpost_pk": str(self.outpost.pk)}, self.channel_name
+            build_outpost_group(self.outpost.pk), self.channel_name
         )
         async_to_sync(self.channel_layer.group_add)(
-            OUTPOST_GROUP_INSTANCE
-            % {"outpost_pk": str(self.outpost.pk), "instance": self.instance_uid},
+            build_outpost_group_instance(self.outpost.pk, self.instance_uid),
             self.channel_name,
         )
         GAUGE_OUTPOSTS_CONNECTED.labels(
@@ -96,12 +103,11 @@ class OutpostConsumer(JsonWebsocketConsumer):
     def disconnect(self, code):
         if self.outpost:
             async_to_sync(self.channel_layer.group_discard)(
-                OUTPOST_GROUP % {"outpost_pk": str(self.outpost.pk)}, self.channel_name
+                build_outpost_group(self.outpost.pk), self.channel_name
             )
             if self.instance_uid:
                 async_to_sync(self.channel_layer.group_discard)(
-                    OUTPOST_GROUP_INSTANCE
-                    % {"outpost_pk": str(self.outpost.pk), "instance": self.instance_uid},
+                    build_outpost_group_instance(self.outpost.pk, self.instance_uid),
                     self.channel_name,
                 )
         if self.outpost and self.instance_uid:

authentik/outposts/controllers/kubernetes.py

@@ -13,6 +13,7 @@ from urllib3.exceptions import HTTPError
 from yaml import dump_all
 
 from authentik.events.logs import LogEvent, capture_logs
+from authentik.lib.utils.reflection import class_to_path
 from authentik.outposts.controllers.base import BaseClient, BaseController, ControllerException
 from authentik.outposts.controllers.k8s.base import KubernetesObjectReconciler
 from authentik.outposts.controllers.k8s.deployment import DeploymentReconciler
@@ -105,7 +106,7 @@ class KubernetesController(BaseController):
                         LogEvent(
                             log_level="info",
                             event=f"{reconcile_key.title()}: Disabled",
-                            logger=str(type(self)),
+                            logger=class_to_path(self.__class__),
                         )
                     )
                     continue
@@ -144,7 +145,7 @@ class KubernetesController(BaseController):
                         LogEvent(
                             log_level="info",
                             event=f"{reconcile_key.title()}: Disabled",
-                            logger=str(type(self)),
+                            logger=class_to_path(self.__class__),
                         )
                     )
                     continue

authentik/outposts/models.py

@@ -357,7 +357,7 @@ class Outpost(ScheduledModel, SerializerModel, ManagedModel):
                             message=(
                                 "While setting the permissions for the service-account, a "
                                 "permission was not found: Check "
-                                "https://goauthentik.io/docs/troubleshooting/missing_permission"
+                                "https://docs.goauthentik.io/troubleshooting/missing_permission"
                             ),
                         ).with_exception(exc).set_user(user).save()
                 else:

authentik/outposts/tasks.py

@@ -12,7 +12,6 @@ from channels.layers import get_channel_layer
 from django.core.cache import cache
 from django.utils.text import slugify
 from django.utils.translation import gettext_lazy as _
-from django_dramatiq_postgres.middleware import CurrentTask
 from docker.constants import DEFAULT_UNIX_SOCKET
 from dramatiq.actor import actor
 from kubernetes.config.incluster_config import SERVICE_TOKEN_FILENAME
@@ -21,7 +20,7 @@ from structlog.stdlib import get_logger
 from yaml import safe_load
 
 from authentik.lib.config import CONFIG
-from authentik.outposts.consumer import OUTPOST_GROUP
+from authentik.outposts.consumer import build_outpost_group
 from authentik.outposts.controllers.base import BaseController, ControllerException
 from authentik.outposts.controllers.docker import DockerClient
 from authentik.outposts.controllers.kubernetes import KubernetesClient
@@ -41,7 +40,7 @@ from authentik.providers.rac.controllers.docker import RACDockerController
 from authentik.providers.rac.controllers.kubernetes import RACKubernetesController
 from authentik.providers.radius.controllers.docker import RadiusDockerController
 from authentik.providers.radius.controllers.kubernetes import RadiusKubernetesController
-from authentik.tasks.models import Task
+from authentik.tasks.middleware import CurrentTask
 
 LOGGER = get_logger()
 CACHE_KEY_OUTPOST_DOWN = "goauthentik.io/outposts/teardown/%s"
@@ -108,7 +107,7 @@ def outpost_service_connection_monitor(connection_pk: Any):
 @actor(description=_("Create/update/monitor/delete the deployment of an Outpost."))
 def outpost_controller(outpost_pk: str, action: str = "up", from_cache: bool = False):
     """Create/update/monitor/delete the deployment of an Outpost"""
-    self: Task = CurrentTask.get_task()
+    self = CurrentTask.get_task()
     self.set_uid(outpost_pk)
     logs = []
     if from_cache:
@@ -142,7 +141,7 @@ def outpost_token_ensurer():
     """
     Periodically ensure that all Outposts have valid Service Accounts and Tokens
     """
-    self: Task = CurrentTask.get_task()
+    self = CurrentTask.get_task()
     all_outposts = Outpost.objects.all()
     for outpost in all_outposts:
         _ = outpost.token
@@ -161,7 +160,7 @@ def outpost_send_update(pk: Any):
     _ = outpost.token
     outpost.build_user_permissions(outpost.user)
     layer = get_channel_layer()
-    group = OUTPOST_GROUP % {"outpost_pk": str(outpost.pk)}
+    group = build_outpost_group(outpost.pk)
     LOGGER.debug("sending update", channel=group, outpost=outpost)
     async_to_sync(layer.group_send)(group, {"type": "event.update"})
 
@@ -169,7 +168,7 @@ def outpost_send_update(pk: Any):
 @actor(description=_("Checks the local environment and create Service connections."))
 def outpost_connection_discovery():
     """Checks the local environment and create Service connections."""
-    self: Task = CurrentTask.get_task()
+    self = CurrentTask.get_task()
     if not CONFIG.get_bool("outposts.discover"):
         self.info("Outpost integration discovery is disabled")
         return
@@ -213,7 +212,7 @@ def outpost_session_end(session_id: str):
     hashed_session_id = hash_session_key(session_id)
     for outpost in Outpost.objects.all():
         LOGGER.info("Sending session end signal to outpost", outpost=outpost)
-        group = OUTPOST_GROUP % {"outpost_pk": str(outpost.pk)}
+        group = build_outpost_group(outpost.pk)
         async_to_sync(layer.group_send)(
             group,
             {

authentik/outposts/tests/test_ws.py

@@ -45,7 +45,7 @@ class TestOutpostWS(TransactionTestCase):
         communicator = WebsocketCommunicator(
             URLRouter(websocket.websocket_urlpatterns),
             f"/ws/outpost/{self.outpost.pk}/",
-            {b"authorization": f"Bearer {self.token}".encode()},
+            [(b"authorization", f"Bearer {self.token}".encode())],
         )
         connected, _ = await communicator.connect()
         self.assertTrue(connected)
@@ -56,7 +56,7 @@ class TestOutpostWS(TransactionTestCase):
         communicator = WebsocketCommunicator(
             URLRouter(websocket.websocket_urlpatterns),
             f"/ws/outpost/{self.outpost.pk}/",
-            {b"authorization": f"Bearer {self.token}".encode()},
+            [(b"authorization", f"Bearer {self.token}".encode())],
         )
         connected, _ = await communicator.connect()
         self.assertTrue(connected)
@@ -83,7 +83,7 @@ class TestOutpostWS(TransactionTestCase):
         communicator = WebsocketCommunicator(
             URLRouter(websocket.websocket_urlpatterns),
             f"/ws/outpost/{self.outpost.pk}/",
-            {b"authorization": f"Bearer {self.token}".encode()},
+            [(b"authorization", f"Bearer {self.token}".encode())],
         )
         connected, _ = await communicator.connect()
         self.assertTrue(connected)

authentik/outposts/urls.py

@@ -11,11 +11,14 @@ from authentik.outposts.api.service_connections import (
 from authentik.outposts.channels import TokenOutpostMiddleware
 from authentik.outposts.consumer import OutpostConsumer
 from authentik.root.middleware import ChannelsLoggingMiddleware
+from authentik.tenants.channels import TenantsAwareMiddleware
 
 websocket_urlpatterns = [
     path(
         "ws/outpost/<uuid:pk>/",
-        ChannelsLoggingMiddleware(TokenOutpostMiddleware(OutpostConsumer.as_asgi())),
+        ChannelsLoggingMiddleware(
+            TenantsAwareMiddleware(TokenOutpostMiddleware(OutpostConsumer.as_asgi()))
+        ),
     ),
 ]
 

authentik/policies/api/bindings.py

@@ -10,9 +10,9 @@ from rest_framework.serializers import PrimaryKeyRelatedField
 from rest_framework.viewsets import ModelViewSet
 from structlog.stdlib import get_logger
 
-from authentik.core.api.groups import GroupSerializer
+from authentik.core.api.groups import PartialUserSerializer
 from authentik.core.api.used_by import UsedByMixin
-from authentik.core.api.users import UserSerializer
+from authentik.core.api.users import PartialGroupSerializer
 from authentik.core.api.utils import ModelSerializer
 from authentik.policies.api.policies import PolicySerializer
 from authentik.policies.models import PolicyBinding, PolicyBindingModel
@@ -61,8 +61,8 @@ class PolicyBindingSerializer(ModelSerializer):
     )
 
     policy_obj = PolicySerializer(required=False, read_only=True, source="policy")
-    group_obj = GroupSerializer(required=False, read_only=True, source="group")
-    user_obj = UserSerializer(required=False, read_only=True, source="user")
+    group_obj = PartialGroupSerializer(required=False, read_only=True, source="group")
+    user_obj = PartialUserSerializer(required=False, read_only=True, source="user")
 
     class Meta:
         model = PolicyBinding
@@ -124,4 +124,5 @@ class PolicyBindingViewSet(UsedByMixin, ModelViewSet):
     serializer_class = PolicyBindingSerializer
     search_fields = ["policy__name"]
     filterset_class = PolicyBindingFilter
-    ordering = ["target", "order"]
+    ordering = ["order", "pk"]
+    ordering_fields = ["order", "target__uuid", "pk"]

authentik/providers/oauth2/api/providers.py

@@ -66,6 +66,7 @@ class OAuth2ProviderSerializer(ProviderSerializer):
             "access_code_validity",
             "access_token_validity",
             "refresh_token_validity",
+            "refresh_token_threshold",
             "include_claims_in_id_token",
             "signing_key",
             "encryption_key",

authentik/providers/oauth2/constants.py

@@ -23,6 +23,8 @@ SCOPE_OPENID_PROFILE = "profile"
 SCOPE_OPENID_EMAIL = "email"
 SCOPE_OFFLINE_ACCESS = "offline_access"
 
+UI_LOCALES = "ui_locales"
+
 # https://www.iana.org/assignments/oauth-parameters/auth-parameters.xhtml#pkce-code-challenge-method
 PKCE_METHOD_PLAIN = "plain"
 PKCE_METHOD_S256 = "S256"

authentik/providers/oauth2/migrations/0030_oauth2provider_refresh_token_threshold.py

@@ -0,0 +1,23 @@
+# Generated by Django 5.1.12 on 2025-09-25 15:26
+
+import authentik.lib.utils.time
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("authentik_providers_oauth2", "0029_oauth2provider__backchannel_logout_uris"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="oauth2provider",
+            name="refresh_token_threshold",
+            field=models.TextField(
+                default="seconds=0",
+                help_text="When refreshing a token, if the refresh token is valid for less than this duration, it will be renewed. When set to seconds=0, token will always be renewed. (Format: hours=1;minutes=2;seconds=3).",
+                validators=[authentik.lib.utils.time.timedelta_string_validator],
+            ),
+        ),
+    ]

authentik/providers/oauth2/models.py

@@ -238,6 +238,16 @@ class OAuth2Provider(WebfingerProvider, Provider):
             "(Format: hours=1;minutes=2;seconds=3)."
         ),
     )
+    refresh_token_threshold = models.TextField(
+        default="seconds=0",
+        validators=[timedelta_string_validator],
+        help_text=_(
+            "When refreshing a token, if the refresh token is valid for less than "
+            "this duration, it will be renewed. "
+            "When set to seconds=0, token will always be renewed. "
+            "(Format: hours=1;minutes=2;seconds=3)."
+        ),
+    )
 
     sub_mode = models.TextField(
         choices=SubModes.choices,

authentik/providers/oauth2/tasks.py

@@ -1,14 +1,13 @@
 """OAuth2 Provider Tasks"""
 
 from django.utils.translation import gettext_lazy as _
-from django_dramatiq_postgres.middleware import CurrentTask
 from dramatiq.actor import actor
 from structlog.stdlib import get_logger
 
 from authentik.lib.utils.http import get_http_session
 from authentik.providers.oauth2.models import OAuth2Provider
 from authentik.providers.oauth2.utils import create_logout_token
-from authentik.tasks.models import Task
+from authentik.tasks.middleware import CurrentTask
 
 LOGGER = get_logger()
 
@@ -31,7 +30,7 @@ def send_backchannel_logout_request(
     Returns:
         bool: True if the request was sent successfully, False otherwise
     """
-    self: Task = CurrentTask.get_task()
+    self = CurrentTask.get_task()
     LOGGER.debug("Sending back-channel logout request", provider_pk=provider_pk, sub=sub)
 
     provider = OAuth2Provider.objects.filter(pk=provider_pk).first()

authentik/providers/oauth2/tests/test_authorize.py

@@ -1,6 +1,7 @@
 """Test authorize view"""
 
 from unittest.mock import MagicMock, patch
+from urllib.parse import parse_qs, urlparse
 
 from django.test import RequestFactory
 from django.urls import reverse
@@ -670,3 +671,55 @@ class TestAuthorize(OAuthTestCase):
         )
         parsed = OAuthAuthorizationParams.from_request(request)
         self.assertNotIn(SCOPE_OFFLINE_ACCESS, parsed.scope)
+
+    def test_ui_locales(self):
+        """Test OIDC ui_locales authorization"""
+        flow = create_test_flow()
+        provider = OAuth2Provider.objects.create(
+            name=generate_id(),
+            client_id="test",
+            authorization_flow=flow,
+            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "foo://localhost")],
+            access_code_validity="seconds=100",
+        )
+        Application.objects.create(name="app", slug="app", provider=provider)
+        state = generate_id()
+        self.client.logout()
+        response = self.client.get(
+            reverse("authentik_providers_oauth2:authorize"),
+            data={
+                "response_type": "code",
+                "client_id": "test",
+                "state": state,
+                "redirect_uri": "foo://localhost",
+                "ui_locales": "invalid fr",
+            },
+        )
+        parsed = parse_qs(urlparse(response.url).query)
+        self.assertEqual(parsed["locale"], ["fr"])
+
+    def test_ui_locales_invalid(self):
+        """Test OIDC ui_locales authorization"""
+        flow = create_test_flow()
+        provider = OAuth2Provider.objects.create(
+            name=generate_id(),
+            client_id="test",
+            authorization_flow=flow,
+            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "foo://localhost")],
+            access_code_validity="seconds=100",
+        )
+        Application.objects.create(name="app", slug="app", provider=provider)
+        state = generate_id()
+        self.client.logout()
+        response = self.client.get(
+            reverse("authentik_providers_oauth2:authorize"),
+            data={
+                "response_type": "code",
+                "client_id": "test",
+                "state": state,
+                "redirect_uri": "foo://localhost",
+                "ui_locales": "invalid",
+            },
+        )
+        parsed = parse_qs(urlparse(response.url).query)
+        self.assertNotIn("locale", parsed)

authentik/providers/oauth2/tests/test_token.py

@@ -376,3 +376,63 @@ class TestToken(OAuthTestCase):
         )
         self.assertEqual(response.status_code, 400)
         self.assertTrue(Event.objects.filter(action=EventAction.SUSPICIOUS_REQUEST).exists())
+
+    @apply_blueprint("system/providers-oauth2.yaml")
+    def test_refresh_token_view_threshold(self):
+        """test request param"""
+        provider = OAuth2Provider.objects.create(
+            name=generate_id(),
+            authorization_flow=create_test_flow(),
+            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://local.invalid")],
+            signing_key=self.keypair,
+            refresh_token_threshold="hours=1",  # nosec
+        )
+        provider.property_mappings.set(
+            ScopeMapping.objects.filter(
+                managed__in=[
+                    "goauthentik.io/providers/oauth2/scope-openid",
+                    "goauthentik.io/providers/oauth2/scope-email",
+                    "goauthentik.io/providers/oauth2/scope-profile",
+                    "goauthentik.io/providers/oauth2/scope-offline_access",
+                ]
+            )
+        )
+        # Needs to be assigned to an application for iss to be set
+        self.app.provider = provider
+        self.app.save()
+        header = b64encode(f"{provider.client_id}:{provider.client_secret}".encode()).decode()
+        user = create_test_admin_user()
+        token: RefreshToken = RefreshToken.objects.create(
+            provider=provider,
+            user=user,
+            token=generate_id(),
+            _id_token=dumps({}),
+            auth_time=timezone.now(),
+            _scope="offline_access",
+        )
+        response = self.client.post(
+            reverse("authentik_providers_oauth2:token"),
+            data={
+                "grant_type": GRANT_TYPE_REFRESH_TOKEN,
+                "refresh_token": token.token,
+                "redirect_uri": "http://local.invalid",
+            },
+            HTTP_AUTHORIZATION=f"Basic {header}",
+            HTTP_ORIGIN="http://local.invalid",
+        )
+        self.assertEqual(response["Access-Control-Allow-Credentials"], "true")
+        self.assertEqual(response["Access-Control-Allow-Origin"], "http://local.invalid")
+        access: AccessToken = AccessToken.objects.filter(user=user, provider=provider).first()
+        self.assertJSONEqual(
+            response.content.decode(),
+            {
+                "access_token": access.token,
+                "token_type": TOKEN_TYPE,
+                "expires_in": 3600,
+                "id_token": provider.encode(
+                    access.id_token.to_dict(),
+                ),
+                "scope": "offline_access",
+            },
+        )
+        self.validate_jwt(access, provider)

authentik/providers/oauth2/tests/test_token_cc_user_pw.py

@@ -8,7 +8,12 @@ from jwt import decode
 
 from authentik.blueprints.tests import apply_blueprint
 from authentik.core.models import Application, Group, Token, TokenIntents, UserTypes
-from authentik.core.tests.utils import create_test_admin_user, create_test_cert, create_test_flow
+from authentik.core.tests.utils import (
+    create_test_admin_user,
+    create_test_cert,
+    create_test_flow,
+    create_test_user,
+)
 from authentik.policies.models import PolicyBinding
 from authentik.providers.oauth2.constants import (
     GRANT_TYPE_CLIENT_CREDENTIALS,
@@ -182,6 +187,47 @@ class TestTokenClientCredentialsUserNamePassword(OAuthTestCase):
         self.assertEqual(jwt["given_name"], self.user.name)
         self.assertEqual(jwt["preferred_username"], self.user.username)
 
+    def test_successful_two_tokens(self):
+        """test successful when two app passwords with the same key exist"""
+        Token.objects.create(
+            identifier="sa-token-two",
+            user=create_test_user(),
+            intent=TokenIntents.INTENT_APP_PASSWORD,
+            expiring=False,
+            key=self.token.key,
+        )
+
+        response = self.client.post(
+            reverse("authentik_providers_oauth2:token"),
+            {
+                "grant_type": GRANT_TYPE_CLIENT_CREDENTIALS,
+                "scope": f"{SCOPE_OPENID} {SCOPE_OPENID_EMAIL} {SCOPE_OPENID_PROFILE}",
+                "client_id": self.provider.client_id,
+                "username": "sa",
+                "password": self.token.key,
+            },
+        )
+        self.assertEqual(response.status_code, 200)
+        body = loads(response.content.decode())
+        self.assertEqual(body["token_type"], TOKEN_TYPE)
+        _, alg = self.provider.jwt_key
+        jwt = decode(
+            body["access_token"],
+            key=self.provider.signing_key.public_key,
+            algorithms=[alg],
+            audience=self.provider.client_id,
+        )
+        self.assertEqual(jwt["given_name"], self.user.name)
+        self.assertEqual(jwt["preferred_username"], self.user.username)
+        jwt = decode(
+            body["id_token"],
+            key=self.provider.signing_key.public_key,
+            algorithms=[alg],
+            audience=self.provider.client_id,
+        )
+        self.assertEqual(jwt["given_name"], self.user.name)
+        self.assertEqual(jwt["preferred_username"], self.user.username)
+
     def test_successful_password(self):
         """test successful (password grant)"""
         response = self.client.post(

authentik/providers/oauth2/views/authorize.py

@@ -5,13 +5,14 @@ from datetime import timedelta
 from json import dumps
 from re import error as RegexError
 from re import fullmatch
-from urllib.parse import parse_qs, urlencode, urlparse, urlsplit, urlunsplit
+from urllib.parse import parse_qs, quote, urlencode, urlparse, urlsplit, urlunparse, urlunsplit
 from uuid import uuid4
 
-from django.http import HttpRequest, HttpResponse
+from django.conf import settings
+from django.http import HttpRequest, HttpResponse, HttpResponseRedirect
 from django.http.response import Http404, HttpResponseBadRequest
 from django.shortcuts import get_object_or_404
-from django.utils import timezone
+from django.utils import timezone, translation
 from django.utils.translation import gettext as _
 from structlog.stdlib import get_logger
 
@@ -41,6 +42,7 @@ from authentik.providers.oauth2.constants import (
     SCOPE_OFFLINE_ACCESS,
     SCOPE_OPENID,
     TOKEN_TYPE,
+    UI_LOCALES,
 )
 from authentik.providers.oauth2.errors import (
     AuthorizeError,
@@ -387,6 +389,45 @@ class AuthorizationFlowInitView(BufferedPolicyAccessView):
         request.context["oauth_response_type"] = self.params.response_type
         return request
 
+    def dispatch_with_language(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
+        """Activate language from OIDC specific ui_locales parameter, picking the earliest one
+        available"""
+        selected_language = None
+        if UI_LOCALES in self.request.GET:
+            languages = str(self.request.GET[UI_LOCALES]).split(" ")
+            for language in languages:
+                if translation.check_for_language(language):
+                    selected_language = translation.get_supported_language_variant(language)
+                    LOGGER.debug(
+                        "Activating language from oidc ui_locales", locale=selected_language
+                    )
+                    break
+            translation.activate(selected_language)
+        response = super().dispatch(request, *args, **kwargs)
+        if selected_language:
+            response.set_cookie(
+                settings.LANGUAGE_COOKIE_NAME,
+                selected_language,
+                max_age=settings.LANGUAGE_COOKIE_AGE,
+                path=settings.LANGUAGE_COOKIE_PATH,
+                domain=settings.LANGUAGE_COOKIE_DOMAIN,
+                secure=settings.LANGUAGE_COOKIE_SECURE,
+                httponly=settings.LANGUAGE_COOKIE_HTTPONLY,
+                samesite=settings.LANGUAGE_COOKIE_SAMESITE,
+            )
+            if isinstance(response, HttpResponseRedirect):
+                parsed_url = urlparse(response.url)
+                args = parse_qs(parsed_url.query)
+                args["locale"] = selected_language
+                response["Location"] = urlunparse(
+                    parsed_url._replace(query=urlencode(args, quote_via=quote, doseq=True))
+                )
+        return response
+
+    def dispatch(self, request: HttpRequest, *args, **kwargs):
+        # Activate language before parsing params (error messages should be localised)
+        return self.dispatch_with_language(request, *args, **kwargs)
+
     def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
         """Start FlowPLanner, return to flow executor shell"""
         # Require a login event to be set, otherwise make the user re-login

authentik/providers/oauth2/views/token.py

@@ -340,7 +340,7 @@ class TokenParams:
         if not user:
             raise TokenError("invalid_grant")
         token: Token = Token.filter_not_expired(
-            key=password, intent=TokenIntents.INTENT_APP_PASSWORD
+            key=password, intent=TokenIntents.INTENT_APP_PASSWORD, user=user
         ).first()
         if not token or token.user.uid != user.uid:
             raise TokenError("invalid_grant")
@@ -684,32 +684,8 @@ class TokenView(View):
         )
         access_token.save()
 
-        refresh_token_expiry = now + timedelta_from_string(self.provider.refresh_token_validity)
-        refresh_token = RefreshToken(
-            user=self.params.refresh_token.user,
-            scope=self.params.refresh_token.scope,
-            expires=refresh_token_expiry,
-            provider=self.provider,
-            auth_time=self.params.refresh_token.auth_time,
-            session=self.params.refresh_token.session,
-        )
-        id_token = IDToken.new(
-            self.provider,
-            refresh_token,
-            self.request,
-        )
-        id_token.nonce = self.params.refresh_token.id_token.nonce
-        id_token.at_hash = access_token.at_hash
-        refresh_token.id_token = id_token
-        refresh_token.save()
-
-        # Mark old token as revoked
-        self.params.refresh_token.revoked = True
-        self.params.refresh_token.save()
-
-        return {
+        res = {
             "access_token": access_token.token,
-            "refresh_token": refresh_token.token,
             "token_type": TOKEN_TYPE,
             "scope": " ".join(access_token.scope),
             "expires_in": int(
@@ -718,6 +694,37 @@ class TokenView(View):
             "id_token": access_token.id_token.to_jwt(self.provider),
         }
 
+        refresh_token_threshold = timedelta_from_string(self.provider.refresh_token_threshold)
+        if (
+            refresh_token_threshold.total_seconds() == 0
+            or (now - self.params.refresh_token.expires) > refresh_token_threshold
+        ):
+            refresh_token_expiry = now + timedelta_from_string(self.provider.refresh_token_validity)
+            refresh_token = RefreshToken(
+                user=self.params.refresh_token.user,
+                scope=self.params.refresh_token.scope,
+                expires=refresh_token_expiry,
+                provider=self.provider,
+                auth_time=self.params.refresh_token.auth_time,
+                session=self.params.refresh_token.session,
+            )
+            id_token = IDToken.new(
+                self.provider,
+                refresh_token,
+                self.request,
+            )
+            id_token.nonce = self.params.refresh_token.id_token.nonce
+            id_token.at_hash = access_token.at_hash
+            refresh_token.id_token = id_token
+            refresh_token.save()
+
+            # Mark old token as revoked
+            self.params.refresh_token.revoked = True
+            self.params.refresh_token.save()
+            res["refresh_token"] = refresh_token.token
+
+        return res
+
     def create_client_credentials_response(self) -> dict[str, Any]:
         """See https://datatracker.ietf.org/doc/html/rfc6749#section-4.4"""
         now = timezone.now()

authentik/providers/proxy/tasks.py

@@ -5,7 +5,7 @@ from channels.layers import get_channel_layer
 from django.utils.translation import gettext_lazy as _
 from dramatiq.actor import actor
 
-from authentik.outposts.consumer import OUTPOST_GROUP
+from authentik.outposts.consumer import build_outpost_group
 from authentik.outposts.models import Outpost, OutpostType
 from authentik.providers.oauth2.id_token import hash_session_key
 
@@ -15,7 +15,7 @@ def proxy_on_logout(session_id: str):
     layer = get_channel_layer()
     hashed_session_id = hash_session_key(session_id)
     for outpost in Outpost.objects.filter(type=OutpostType.PROXY):
-        group = OUTPOST_GROUP % {"outpost_pk": str(outpost.pk)}
+        group = build_outpost_group(outpost.pk)
         async_to_sync(layer.group_send)(
             group,
             {

authentik/providers/rac/api/connection_tokens.py

@@ -3,7 +3,7 @@
 from rest_framework import mixins
 from rest_framework.viewsets import GenericViewSet
 
-from authentik.core.api.groups import GroupMemberSerializer
+from authentik.core.api.groups import PartialUserSerializer
 from authentik.core.api.used_by import UsedByMixin
 from authentik.core.api.utils import ModelSerializer
 from authentik.providers.rac.api.endpoints import EndpointSerializer
@@ -16,7 +16,7 @@ class ConnectionTokenSerializer(ModelSerializer):
 
     provider_obj = RACProviderSerializer(source="provider", read_only=True)
     endpoint_obj = EndpointSerializer(source="endpoint", read_only=True)
-    user = GroupMemberSerializer(source="session.user", read_only=True)
+    user = PartialUserSerializer(source="session.user", read_only=True)
 
     class Meta:
         model = ConnectionToken

authentik/providers/rac/consumer_client.py

@@ -1,28 +1,46 @@
 """RAC Client consumer"""
 
+from hashlib import sha256
+
 from asgiref.sync import async_to_sync
 from channels.db import database_sync_to_async
 from channels.exceptions import ChannelFull, DenyConnection
 from channels.generic.websocket import AsyncWebsocketConsumer
+from django.db import connection
 from django.http.request import QueryDict
 from structlog.stdlib import BoundLogger, get_logger
 
-from authentik.outposts.consumer import OUTPOST_GROUP_INSTANCE
+from authentik.outposts.consumer import build_outpost_group_instance
 from authentik.outposts.models import Outpost, OutpostState, OutpostType
 from authentik.providers.rac.models import ConnectionToken, RACProvider
 
-# Global broadcast group, which messages are sent to when the outpost connects back
-# to authentik for a specific connection
-# The `RACClientConsumer` consumer adds itself to this group on connection,
-# and removes itself once it has been assigned a specific outpost channel
-RAC_CLIENT_GROUP = "group_rac_client"
-# A group for all connections in a given authentik session ID
-# A disconnect message is sent to this group when the session expires/is deleted
-RAC_CLIENT_GROUP_SESSION = "group_rac_client_%(session)s"
-# A group for all connections with a specific token, which in almost all cases
-# is just one connection, however this is used to disconnect the connection
-# when the token is deleted
-RAC_CLIENT_GROUP_TOKEN = "group_rac_token_%(token)s"  # nosec
+
+def build_rac_client_group() -> str:
+    """
+    Global broadcast group, which messages are sent to when the outpost connects back
+    to authentik for a specific connection
+    The `RACClientConsumer` consumer adds itself to this group on connection,
+    and removes itself once it has been assigned a specific outpost channel
+    """
+    return sha256(f"{connection.schema_name}/group_rac_client".encode()).hexdigest()
+
+
+def build_rac_client_group_session(session_key: str) -> str:
+    """
+    A group for all connections in a given authentik session ID
+    A disconnect message is sent to this group when the session expires/is deleted
+    """
+    return sha256(f"{connection.schema_name}/group_rac_client_{session_key}".encode()).hexdigest()
+
+
+def build_rac_client_group_token(token: str) -> str:
+    """
+    A group for all connections with a specific token, which in almost all cases
+    is just one connection, however this is used to disconnect the connection
+    when the token is deleted
+    """
+    return sha256(f"{connection.schema_name}/group_rac_token_{token}".encode()).hexdigest()
+
 
 # Step 1: Client connects to this websocket endpoint
 # Step 2: We prepare all the connection args for Guac
@@ -45,22 +63,23 @@ class RACClientConsumer(AsyncWebsocketConsumer):
     async def connect(self):
         self.logger = get_logger()
         await self.accept("guacamole")
-        await self.channel_layer.group_add(RAC_CLIENT_GROUP, self.channel_name)
+        await self.channel_layer.group_add(build_rac_client_group(), self.channel_name)
         await self.channel_layer.group_add(
-            RAC_CLIENT_GROUP_SESSION % {"session": self.scope["session"].session_key},
+            build_rac_client_group_session(self.scope["session"].session_key),
             self.channel_name,
         )
         await self.init_outpost_connection()
 
     async def disconnect(self, code):
         self.logger.debug("Disconnecting")
-        # Tell the outpost we're disconnecting
-        await self.channel_layer.send(
-            self.dest_channel_id,
-            {
-                "type": "event.disconnect",
-            },
-        )
+        if self.dest_channel_id:
+            # Tell the outpost we're disconnecting
+            await self.channel_layer.send(
+                self.dest_channel_id,
+                {
+                    "type": "event.disconnect",
+                },
+            )
 
     @database_sync_to_async
     def init_outpost_connection(self):
@@ -109,10 +128,8 @@ class RACClientConsumer(AsyncWebsocketConsumer):
             if len(states) < 1:
                 continue
             self.logger.debug("Sending out connection broadcast")
-            async_to_sync(self.channel_layer.group_send)(
-                OUTPOST_GROUP_INSTANCE % {"outpost_pk": str(outpost.pk), "instance": states[0].uid},
-                msg,
-            )
+            group = build_outpost_group_instance(outpost.pk, states[0].uid)
+            async_to_sync(self.channel_layer.group_send)(group, msg)
         if self.provider and self.provider.delete_token_on_disconnect:
             self.logger.info("Deleting connection token to prevent reconnect", token=self.token)
             self.token.delete()
@@ -157,7 +174,7 @@ class RACClientConsumer(AsyncWebsocketConsumer):
         self.dest_channel_id = outpost_channel
         # Since we have a specific outpost channel now, we can remove
         # ourselves from the global broadcast group
-        await self.channel_layer.group_discard(RAC_CLIENT_GROUP, self.channel_name)
+        await self.channel_layer.group_discard(build_rac_client_group(), self.channel_name)
 
     async def event_send(self, event: dict):
         """Handler called by outpost websocket that sends data to this specific

authentik/providers/rac/consumer_outpost.py

@@ -3,7 +3,7 @@
 from channels.exceptions import ChannelFull
 from channels.generic.websocket import AsyncWebsocketConsumer
 
-from authentik.providers.rac.consumer_client import RAC_CLIENT_GROUP
+from authentik.providers.rac.consumer_client import build_rac_client_group
 
 
 class RACOutpostConsumer(AsyncWebsocketConsumer):
@@ -15,7 +15,7 @@ class RACOutpostConsumer(AsyncWebsocketConsumer):
         self.dest_channel_id = self.scope["url_route"]["kwargs"]["channel"]
         await self.accept()
         await self.channel_layer.group_send(
-            RAC_CLIENT_GROUP,
+            build_rac_client_group(),
             {
                 "type": "event.outpost.connected",
                 "outpost_channel": self.channel_name,

authentik/providers/rac/signals.py

@@ -9,8 +9,8 @@ from django.dispatch import receiver
 from authentik.core.models import AuthenticatedSession
 from authentik.providers.rac.api.endpoints import user_endpoint_cache_key
 from authentik.providers.rac.consumer_client import (
-    RAC_CLIENT_GROUP_SESSION,
-    RAC_CLIENT_GROUP_TOKEN,
+    build_rac_client_group_session,
+    build_rac_client_group_token,
 )
 from authentik.providers.rac.models import ConnectionToken, Endpoint
 
@@ -19,10 +19,7 @@ from authentik.providers.rac.models import ConnectionToken, Endpoint
 def user_session_deleted(sender, instance: AuthenticatedSession, **_):
     layer = get_channel_layer()
     async_to_sync(layer.group_send)(
-        RAC_CLIENT_GROUP_SESSION
-        % {
-            "session": instance.session.session_key,
-        },
+        build_rac_client_group_session(instance.session.session_key),
         {"type": "event.disconnect", "reason": "session_logout"},
     )
 
@@ -32,10 +29,7 @@ def pre_delete_connection_token_disconnect(sender, instance: ConnectionToken, **
     """Disconnect session when connection token is deleted"""
     layer = get_channel_layer()
     async_to_sync(layer.group_send)(
-        RAC_CLIENT_GROUP_TOKEN
-        % {
-            "token": instance.token,
-        },
+        build_rac_client_group_token(instance.token),
         {"type": "event.disconnect", "reason": "token_delete"},
     )
 

authentik/providers/rac/urls.py

@@ -12,6 +12,7 @@ from authentik.providers.rac.consumer_outpost import RACOutpostConsumer
 from authentik.providers.rac.views import RACInterface, RACStartView
 from authentik.root.asgi_middleware import AuthMiddlewareStack
 from authentik.root.middleware import ChannelsLoggingMiddleware
+from authentik.tenants.channels import TenantsAwareMiddleware
 
 urlpatterns = [
     path(
@@ -29,11 +30,15 @@ urlpatterns = [
 websocket_urlpatterns = [
     path(
         "ws/rac/<str:token>/",
-        ChannelsLoggingMiddleware(AuthMiddlewareStack(RACClientConsumer.as_asgi())),
+        ChannelsLoggingMiddleware(
+            TenantsAwareMiddleware(AuthMiddlewareStack(RACClientConsumer.as_asgi()))
+        ),
     ),
     path(
         "ws/outpost_rac/<str:channel>/",
-        ChannelsLoggingMiddleware(TokenOutpostMiddleware(RACOutpostConsumer.as_asgi())),
+        ChannelsLoggingMiddleware(
+            TenantsAwareMiddleware(TokenOutpostMiddleware(RACOutpostConsumer.as_asgi()))
+        ),
     ),
 ]
 

authentik/providers/saml/tests/test_auth_n_request.py

@@ -97,6 +97,7 @@ class TestAuthNRequest(TestCase):
             pre_authentication_flow=create_test_flow(),
             signing_kp=self.cert,
             verification_kp=self.cert,
+            signed_assertion=True,
         )
 
     def test_signed_valid(self):
@@ -171,6 +172,7 @@ class TestAuthNRequest(TestCase):
         self.provider.sign_assertion = True
         self.provider.sign_response = True
         self.provider.save()
+        self.source.signed_response = True
         http_request = get_request("/")
 
         # First create an AuthNRequest

authentik/providers/scim/api/groups.py

@@ -4,7 +4,7 @@ from rest_framework import mixins
 from rest_framework.viewsets import GenericViewSet
 
 from authentik.core.api.used_by import UsedByMixin
-from authentik.core.api.users import UserGroupSerializer
+from authentik.core.api.users import PartialGroupSerializer
 from authentik.core.api.utils import ModelSerializer
 from authentik.lib.sync.outgoing.api import OutgoingSyncConnectionCreateMixin
 from authentik.providers.scim.models import SCIMProviderGroup
@@ -13,7 +13,7 @@ from authentik.providers.scim.models import SCIMProviderGroup
 class SCIMProviderGroupSerializer(ModelSerializer):
     """SCIMProviderGroup Serializer"""
 
-    group_obj = UserGroupSerializer(source="group", read_only=True)
+    group_obj = PartialGroupSerializer(source="group", read_only=True)
 
     class Meta:
 

authentik/providers/scim/api/users.py

@@ -3,7 +3,7 @@
 from rest_framework import mixins
 from rest_framework.viewsets import GenericViewSet
 
-from authentik.core.api.groups import GroupMemberSerializer
+from authentik.core.api.groups import PartialUserSerializer
 from authentik.core.api.used_by import UsedByMixin
 from authentik.core.api.utils import ModelSerializer
 from authentik.lib.sync.outgoing.api import OutgoingSyncConnectionCreateMixin
@@ -13,7 +13,7 @@ from authentik.providers.scim.models import SCIMProviderUser
 class SCIMProviderUserSerializer(ModelSerializer):
     """SCIMProviderUser Serializer"""
 
-    user_obj = GroupMemberSerializer(source="user", read_only=True)
+    user_obj = PartialUserSerializer(source="user", read_only=True)
 
     class Meta:
 

authentik/providers/scim/clients/base.py

@@ -88,10 +88,11 @@ class SCIMClient[TModel: "Model", TConnection: "Model", TSchema: "BaseModel"](
     def get_service_provider_config(self):
         """Get Service provider config"""
         default_config = ServiceProviderConfiguration.default()
+        path = "/ServiceProviderConfig"
+        if self.provider.compatibility_mode == SCIMCompatibilityMode.SALESFORCE:
+            path = "/ServiceProviderConfigs"
         try:
-            config = ServiceProviderConfiguration.model_validate(
-                self._request("GET", "/ServiceProviderConfig")
-            )
+            config = ServiceProviderConfiguration.model_validate(self._request("GET", path))
             if self.provider.compatibility_mode == SCIMCompatibilityMode.AWS:
                 config.patch.supported = False
             if self.provider.compatibility_mode == SCIMCompatibilityMode.SLACK:

authentik/providers/scim/migrations/0015_alter_scimprovider_compatibility_mode.py

@@ -0,0 +1,32 @@
+# Generated by Django 5.1.12 on 2025-09-24 12:10
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        (
+            "authentik_providers_scim",
+            "0014_scimprovider_auth_mode_scimprovider_auth_oauth_and_more",
+        ),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="scimprovider",
+            name="compatibility_mode",
+            field=models.CharField(
+                choices=[
+                    ("default", "Default"),
+                    ("aws", "AWS"),
+                    ("slack", "Slack"),
+                    ("sfdc", "Salesforce"),
+                ],
+                default="default",
+                help_text="Alter authentik behavior for vendor-specific SCIM implementations.",
+                max_length=30,
+                verbose_name="SCIM Compatibility Mode",
+            ),
+        ),
+    ]

authentik/providers/scim/models.py

@@ -78,6 +78,7 @@ class SCIMCompatibilityMode(models.TextChoices):
     DEFAULT = "default", _("Default")
     AWS = "aws", _("AWS")
     SLACK = "slack", _("Slack")
+    SALESFORCE = "sfdc", _("Salesforce")
 
 
 class SCIMProvider(OutgoingSyncProvider, BackchannelProvider):

authentik/rbac/api/rbac_assigned_by_users.py

@@ -1,5 +1,6 @@
 """common RBAC serializers"""
 
+from django.contrib.auth.models import Permission
 from django.db.models import Q, QuerySet
 from django.db.transaction import atomic
 from django_filters.filters import CharFilter, ChoiceFilter
@@ -15,9 +16,9 @@ from rest_framework.request import Request
 from rest_framework.response import Response
 from rest_framework.viewsets import GenericViewSet
 
-from authentik.core.api.groups import GroupMemberSerializer
+from authentik.core.api.groups import PartialUserSerializer
 from authentik.core.api.utils import ModelSerializer
-from authentik.core.models import User, UserTypes
+from authentik.core.models import Group, User, UserTypes
 from authentik.policies.event_matcher.models import model_choices
 from authentik.rbac.api.rbac import PermissionAssignResultSerializer, PermissionAssignSerializer
 from authentik.rbac.decorators import permission_required
@@ -37,15 +38,15 @@ class UserObjectPermissionSerializer(ModelSerializer):
         fields = ["id", "codename", "model", "app_label", "object_pk", "name"]
 
 
-class UserAssignedObjectPermissionSerializer(GroupMemberSerializer):
+class UserAssignedObjectPermissionSerializer(PartialUserSerializer):
     """Users assigned object permission serializer"""
 
     permissions = UserObjectPermissionSerializer(many=True, source="userobjectpermission_set")
     is_superuser = BooleanField()
 
     class Meta:
-        model = GroupMemberSerializer.Meta.model
-        fields = GroupMemberSerializer.Meta.fields + ["permissions", "is_superuser"]
+        model = PartialUserSerializer.Meta.model
+        fields = PartialUserSerializer.Meta.fields + ["permissions", "is_superuser"]
 
 
 class UserAssignedPermissionFilter(FilterSet):
@@ -54,26 +55,56 @@ class UserAssignedPermissionFilter(FilterSet):
     model = ChoiceFilter(choices=model_choices(), method="filter_model", required=True)
     object_pk = CharFilter(method="filter_object_pk")
 
+    def filter_queryset(self, queryset):
+        queryset = super().filter_queryset(queryset)
+        data = self.form.cleaned_data
+        model: str = data["model"]
+        object_pk: str | None = data.get("object_pk", None)
+        app, _, model = model.partition(".")
+
+        superuser_pks = (
+            Group.objects.filter(is_superuser=True).values_list("users", flat=True).distinct()
+        )
+
+        permissions = Permission.objects.filter(
+            content_type__app_label=app,
+            content_type__model=model,
+        )
+
+        user_pks_with_model_permission = (
+            permissions.order_by().values_list("user", flat=True).distinct()
+        )
+        user_pks_with_object_permission = []
+        if object_pk:
+            user_pks_with_object_permission = (
+                UserObjectPermission.objects.filter(
+                    permission__in=permissions,
+                    object_pk=object_pk,
+                )
+                .order_by()
+                .values_list("user", flat=True)
+                .distinct()
+            )
+
+        return queryset.filter(
+            Q(pk__in=superuser_pks)
+            | Q(pk__in=user_pks_with_model_permission)
+            | Q(pk__in=user_pks_with_object_permission)
+        )
+
     def filter_model(self, queryset: QuerySet, name, value: str) -> QuerySet:
         """Filter by object type"""
-        app, _, model = value.partition(".")
-        return queryset.filter(
-            Q(
-                user_permissions__content_type__app_label=app,
-                user_permissions__content_type__model=model,
-            )
-            | Q(
-                userobjectpermission__permission__content_type__app_label=app,
-                userobjectpermission__permission__content_type__model=model,
-            )
-            | Q(ak_groups__is_superuser=True)
-        ).distinct()
+        # Actual filtering is handled by the above method where both `model` and `object_pk` are
+        # available. Don't do anything here, this method is only left here to avoid overriding too
+        # much of filter_queryset.
+        return queryset
 
     def filter_object_pk(self, queryset: QuerySet, name, value: str) -> QuerySet:
         """Filter by object primary key"""
-        return queryset.filter(
-            Q(userobjectpermission__object_pk=value) | Q(ak_groups__is_superuser=True),
-        ).distinct()
+        # Actual filtering is handled by the above method where both `model` and `object_pk` are
+        # available. Don't do anything here, this method is only left here to avoid overriding too
+        # much of filter_queryset.
+        return queryset
 
 
 class UserAssignedPermissionViewSet(ListModelMixin, GenericViewSet):
@@ -83,7 +114,7 @@ class UserAssignedPermissionViewSet(ListModelMixin, GenericViewSet):
     ordering = ["username"]
     # The filtering is done in the filterset,
     # which has a required filter that does the heavy lifting
-    queryset = User.objects.all()
+    queryset = User.objects.all().prefetch_related("userobjectpermission_set")
     filterset_class = UserAssignedPermissionFilter
 
     @permission_required("authentik_core.assign_user_permissions")

authentik/root/channels.py

@@ -0,0 +1,35 @@
+from typing import Any
+
+from channels_postgres.core import PostgresChannelLayer as BasePostgresChannelLayer
+from channels_postgres.db import DatabaseLayer as BaseDatabaseLayer
+from django.conf import settings
+from psycopg_pool import AsyncConnectionPool
+
+from authentik.root.db.base import DatabaseWrapper
+
+
+class DatabaseLayer(BaseDatabaseLayer):
+    async def get_db_pool(self, db_params: dict[str, Any]) -> AsyncConnectionPool:
+        db_wrapper = DatabaseWrapper(settings.CHANNEL_LAYERS["default"]["CONFIG"])
+        db_params = db_wrapper.get_connection_params()
+        db_params.pop("cursor_factory")
+        db_params.pop("context")
+        return await super().get_db_pool(db_params)
+
+
+class PostgresChannelLayer(BasePostgresChannelLayer):
+    def __init__(self, *args, **kwargs) -> None:
+        super().__init__(*args, **kwargs)
+        self.django_db = DatabaseLayer(self.django_db.psycopg_options, self.db_params)
+
+    @property
+    def db_params(self):
+        db_wrapper = DatabaseWrapper(settings.CHANNEL_LAYERS["default"]["CONFIG"])
+        db_params = db_wrapper.get_connection_params()
+        db_params.pop("cursor_factory")
+        db_params.pop("context")
+        return db_params
+
+    @db_params.setter
+    def db_params(self, value):
+        pass

authentik/root/middleware.py

@@ -292,7 +292,7 @@ class ChannelsLoggingMiddleware:
         except DenyConnection:
             return await send({"type": "websocket.close"})
         except Exception as exc:
-            if settings.DEBUG:
+            if settings.DEBUG or settings.TEST:
                 raise exc
             LOGGER.warning("Exception in ASGI application", exc=exc)
             return await send({"type": "websocket.close"})

authentik/root/monitoring.py

@@ -11,8 +11,6 @@ from django.dispatch import Signal
 from django.http import HttpRequest, HttpResponse
 from django.views import View
 from django_prometheus.exports import ExportToDjangoView
-from django_redis import get_redis_connection
-from redis.exceptions import RedisError
 
 monitoring_set = Signal()
 
@@ -44,19 +42,17 @@ class LiveView(View):
 
 
 class ReadyView(View):
-    """View for readiness probe, always returns Http 200, unless sql or redis is down"""
+    """View for readiness probe, always returns Http 200, unless sql is down"""
+
+    def check_db(self):
+        for db_conn in connections.all():
+            # Force connection reload
+            db_conn.connect()
+            _ = db_conn.cursor()
 
     def dispatch(self, request: HttpRequest) -> HttpResponse:
         try:
-            for db_conn in connections.all():
-                # Force connection reload
-                db_conn.connect()
-                _ = db_conn.cursor()
+            self.check_db()
         except OperationalError:  # pragma: no cover
             return HttpResponse(status=503)
-        try:
-            redis_conn = get_redis_connection()
-            redis_conn.ping()
-        except RedisError:  # pragma: no cover
-            return HttpResponse(status=503)
         return HttpResponse(status=200)

authentik/root/settings.py

@@ -10,7 +10,7 @@ from sentry_sdk import set_tag
 from xmlsec import enable_debug_trace
 
 from authentik import authentik_version
-from authentik.lib.config import CONFIG, django_db_config, redis_url
+from authentik.lib.config import CONFIG, django_db_config
 from authentik.lib.logging import get_logger_config, structlog_configure
 from authentik.lib.sentry import sentry_init
 from authentik.lib.utils.reflection import get_env
@@ -64,6 +64,7 @@ SHARED_APPS = [
     "pgactivity",
     "pglock",
     "channels",
+    "channels_postgres",
     "django_dramatiq_postgres",
     "authentik.tasks",
 ]
@@ -72,6 +73,7 @@ TENANT_APPS = [
     "django.contrib.contenttypes",
     "django.contrib.sessions",
     "pgtrigger",
+    "django_postgres_cache",
     "authentik.admin",
     "authentik.api",
     "authentik.core",
@@ -103,6 +105,7 @@ TENANT_APPS = [
     "authentik.sources.plex",
     "authentik.sources.saml",
     "authentik.sources.scim",
+    "authentik.sources.telegram",
     "authentik.stages.authenticator",
     "authentik.stages.authenticator_duo",
     "authentik.stages.authenticator_email",
@@ -185,6 +188,7 @@ SPECTACULAR_SETTINGS = {
     "POSTPROCESSING_HOOKS": [
         "authentik.api.schema.postprocess_schema_responses",
         "authentik.api.schema.postprocess_schema_pagination",
+        "authentik.api.schema.postprocess_schema_remove_unused",
         "drf_spectacular.hooks.postprocess_schema_enums",
     ],
 }
@@ -224,20 +228,11 @@ REST_FRAMEWORK = {
 
 CACHES = {
     "default": {
-        "BACKEND": "django_redis.cache.RedisCache",
-        "LOCATION": CONFIG.get("cache.url") or redis_url(CONFIG.get("redis.db")),
-        "TIMEOUT": CONFIG.get_int("cache.timeout", 300),
-        "OPTIONS": {
-            "CLIENT_CLASS": "django_redis.client.DefaultClient",
-        },
-        "KEY_PREFIX": "authentik_cache",
+        "BACKEND": "django_postgres_cache.backend.DatabaseCache",
         "KEY_FUNCTION": "django_tenants.cache.make_key",
         "REVERSE_KEY_FUNCTION": "django_tenants.cache.reverse_key",
     }
 }
-DJANGO_REDIS_SCAN_ITERSIZE = 1000
-DJANGO_REDIS_IGNORE_EXCEPTIONS = True
-DJANGO_REDIS_LOG_IGNORED_EXCEPTIONS = True
 SESSION_ENGINE = "authentik.core.sessions"
 # Configured via custom SessionMiddleware
 # SESSION_COOKIE_SAMESITE = "None"
@@ -295,16 +290,6 @@ TEMPLATES = [
 
 ASGI_APPLICATION = "authentik.root.asgi.application"
 
-CHANNEL_LAYERS = {
-    "default": {
-        "BACKEND": "channels_redis.pubsub.RedisPubSubChannelLayer",
-        "CONFIG": {
-            "hosts": [CONFIG.get("channel.url") or redis_url(CONFIG.get("redis.db"))],
-            "prefix": "authentik_channels_",
-        },
-    },
-}
-
 
 # Database
 # https://docs.djangoproject.com/en/2.1/ref/settings/#databases
@@ -317,6 +302,16 @@ DATABASE_ROUTERS = (
     "django_tenants.routers.TenantSyncRouter",
 )
 
+CHANNEL_LAYERS = {
+    "default": {
+        "BACKEND": "authentik.root.channels.PostgresChannelLayer",
+        "CONFIG": {
+            **DATABASES["default"],
+            "TIME_ZONE": None,
+        },
+    },
+}
+
 # Email
 # These values should never actually be used, emails are only sent from email stages, which
 # loads the config directly from CONFIG
@@ -398,8 +393,6 @@ DRAMATIQ = {
     ).total_seconds(),
     "middlewares": (
         ("django_dramatiq_postgres.middleware.FullyQualifiedActorName", {}),
-        # TODO: fixme
-        # ("dramatiq.middleware.prometheus.Prometheus", {}),
         ("django_dramatiq_postgres.middleware.DbConnectionMiddleware", {}),
         ("dramatiq.middleware.age_limit.AgeLimit", {}),
         (
@@ -416,10 +409,13 @@ DRAMATIQ = {
         ("dramatiq.middleware.pipelines.Pipelines", {}),
         (
             "dramatiq.middleware.retries.Retries",
-            {"max_retries": CONFIG.get_int("worker.task_max_retries") if not TEST else 0},
+            {
+                "max_retries": CONFIG.get_int("worker.task_max_retries") if not TEST else 0,
+                "max_backoff": 60 * 60 * 1000,  # 1 hour
+            },
         ),
         ("dramatiq.results.middleware.Results", {"store_results": True}),
-        ("django_dramatiq_postgres.middleware.CurrentTask", {}),
+        ("authentik.tasks.middleware.CurrentTask", {}),
         ("authentik.tasks.middleware.TenantMiddleware", {}),
         ("authentik.tasks.middleware.RelObjMiddleware", {}),
         ("authentik.tasks.middleware.MessagesMiddleware", {}),

authentik/root/test_runner.py

@@ -8,7 +8,6 @@ from unittest.mock import patch
 import pytest
 from django.conf import settings
 from django.contrib.contenttypes.models import ContentType
-from django.test import TestCase as DjangoTestCase
 from django.test.runner import DiscoverRunner
 from structlog.stdlib import get_logger
 
@@ -21,8 +20,6 @@ from authentik.tasks.test import use_test_broker
 
 # globally set maxDiff to none to show full assert error
 TestCase.maxDiff = None
-# allow testing with read-replicas
-DjangoTestCase.databases = "__all__"
 
 
 def get_docker_tag() -> str:
@@ -65,15 +62,11 @@ class PytestTestRunner(DiscoverRunner):  # pragma: no cover
         """Configure test environment settings"""
         settings.TEST = True
         settings.DRAMATIQ["test"] = True
-
-        # Set any other test databases's name to their test name early
-        # django does this itself, however only _after_ migrating the default alias
-        # which triggers some reads that might go to the read replica, which
-        # would be routed to the wrong database
-        for alias, db in settings.DATABASES.items():
-            if alias == "default":
-                continue
-            db["NAME"] = db["TEST"]["NAME"]
+        settings.CHANNEL_LAYERS["default"]["CONFIG"] = {
+            **settings.DATABASES["default"],
+            **settings.DATABASES["default"]["TEST"],
+            "TIME_ZONE": None,
+        }
 
         # Test-specific configuration
         test_config = {

authentik/sources/kerberos/tasks.py

@@ -2,7 +2,6 @@
 
 from django.core.cache import cache
 from django.utils.translation import gettext_lazy as _
-from django_dramatiq_postgres.middleware import CurrentTask
 from dramatiq.actor import actor
 from structlog.stdlib import get_logger
 
@@ -10,7 +9,7 @@ from authentik.lib.config import CONFIG
 from authentik.lib.sync.outgoing.exceptions import StopSync
 from authentik.sources.kerberos.models import KerberosSource
 from authentik.sources.kerberos.sync import KerberosSync
-from authentik.tasks.models import Task
+from authentik.tasks.middleware import CurrentTask
 
 LOGGER = get_logger()
 CACHE_KEY_STATUS = "goauthentik.io/sources/kerberos/status/"
@@ -33,7 +32,7 @@ def kerberos_connectivity_check(pk: str):
     description=_("Sync Kerberos source."),
 )
 def kerberos_sync(pk: str):
-    self: Task = CurrentTask.get_task()
+    self = CurrentTask.get_task()
     source: KerberosSource = KerberosSource.objects.filter(enabled=True, pk=pk).first()
     if not source:
         return

authentik/sources/ldap/tasks.py

@@ -4,7 +4,6 @@ from uuid import uuid4
 
 from django.core.cache import cache
 from django.utils.translation import gettext_lazy as _
-from django_dramatiq_postgres.middleware import CurrentTask
 from dramatiq.actor import actor
 from dramatiq.composition import group
 from dramatiq.message import Message
@@ -21,6 +20,7 @@ from authentik.sources.ldap.sync.forward_delete_users import UserLDAPForwardDele
 from authentik.sources.ldap.sync.groups import GroupLDAPSynchronizer
 from authentik.sources.ldap.sync.membership import MembershipLDAPSynchronizer
 from authentik.sources.ldap.sync.users import UserLDAPSynchronizer
+from authentik.tasks.middleware import CurrentTask
 from authentik.tasks.models import Task
 
 LOGGER = get_logger()
@@ -53,7 +53,7 @@ def ldap_connectivity_check(pk: str | None = None):
 )
 def ldap_sync(source_pk: str):
     """Sync a single source"""
-    task: Task = CurrentTask.get_task()
+    task = CurrentTask.get_task()
     source: LDAPSource = LDAPSource.objects.filter(pk=source_pk, enabled=True).first()
     if not source:
         return
@@ -127,7 +127,7 @@ def ldap_sync_paginator(
 )
 def ldap_sync_page(source_pk: str, sync_class: str, page_cache_key: str):
     """Synchronization of an LDAP Source"""
-    self: Task = CurrentTask.get_task()
+    self = CurrentTask.get_task()
     source: LDAPSource = LDAPSource.objects.filter(pk=source_pk).first()
     if not source:
         # Because the source couldn't be found, we don't have a UID

authentik/sources/oauth/tasks.py

@@ -3,14 +3,13 @@
 from json import dumps
 
 from django.utils.translation import gettext_lazy as _
-from django_dramatiq_postgres.middleware import CurrentTask
 from dramatiq.actor import actor
 from requests import RequestException
 from structlog.stdlib import get_logger
 
 from authentik.lib.utils.http import get_http_session
 from authentik.sources.oauth.models import OAuthSource
-from authentik.tasks.models import Task
+from authentik.tasks.middleware import CurrentTask
 
 LOGGER = get_logger()
 
@@ -21,7 +20,7 @@ LOGGER = get_logger()
     )
 )
 def update_well_known_jwks():
-    self: Task = CurrentTask.get_task()
+    self = CurrentTask.get_task()
     session = get_http_session()
     for source in OAuthSource.objects.all().exclude(oidc_well_known_url=""):
         try:

authentik/sources/oauth/tests/test_views.py

@@ -1,10 +1,17 @@
 """OAuth Source tests"""
 
+from urllib.parse import parse_qs
+
 from django.urls import reverse
 from requests_mock import Mocker
 from rest_framework.test import APITestCase
 
+from authentik.core.models import User
 from authentik.core.tests.utils import create_test_admin_user
+from authentik.flows.planner import PLAN_CONTEXT_PENDING_USER, FlowPlan
+from authentik.flows.stage import PLAN_CONTEXT_PENDING_USER_IDENTIFIER
+from authentik.flows.views.executor import SESSION_KEY_PLAN
+from authentik.lib.generators import generate_id
 from authentik.sources.oauth.api.source import OAuthSourceSerializer
 from authentik.sources.oauth.models import OAuthSource
 
@@ -124,20 +131,68 @@ class TestOAuthSource(APITestCase):
             )
             self.assertFalse(serializer.is_valid())
 
-    def test_source_redirect(self):
-        """test redirect view"""
-        self.client.get(
+    def test_source_redirect_login_hint_user(self):
+        """test redirect view with login hint"""
+        user = User(email="foo@authentik.company")
+        session = self.client.session
+        plan = FlowPlan(generate_id())
+        plan.context[PLAN_CONTEXT_PENDING_USER] = user
+        session[SESSION_KEY_PLAN] = plan
+        session.save()
+
+        res = self.client.get(
             reverse(
                 "authentik_sources_oauth:oauth-client-login",
                 kwargs={"source_slug": self.source.slug},
             )
         )
+        self.assertEqual(res.status_code, 302)
+        qs = parse_qs(res.url)
+        self.assertEqual(qs["login_hint"], ["foo@authentik.company"])
+
+    def test_source_redirect_login_hint_user_identifier(self):
+        """test redirect view with login hint"""
+        session = self.client.session
+        plan = FlowPlan(generate_id())
+        plan.context[PLAN_CONTEXT_PENDING_USER_IDENTIFIER] = "foo@authentik.company"
+        session[SESSION_KEY_PLAN] = plan
+        session.save()
+
+        res = self.client.get(
+            reverse(
+                "authentik_sources_oauth:oauth-client-login",
+                kwargs={"source_slug": self.source.slug},
+            )
+        )
+        self.assertEqual(res.status_code, 302)
+        qs = parse_qs(res.url)
+        self.assertEqual(qs["login_hint"], ["foo@authentik.company"])
+
+    def test_source_redirect(self):
+        """test redirect view"""
+        res = self.client.get(
+            reverse(
+                "authentik_sources_oauth:oauth-client-login",
+                kwargs={"source_slug": self.source.slug},
+            )
+        )
+        self.assertEqual(res.status_code, 302)
+        qs = parse_qs(res.url)
+
+        session = self.client.session
+        state = session[f"oauth-client-{self.source.name}-request-state"]
+
+        self.assertEqual(qs["redirect_uri"], ["http://testserver/source/oauth/callback/test/"])
+        self.assertEqual(qs["response_type"], ["code"])
+        self.assertEqual(qs["state"], [state])
+        self.assertEqual(qs["scope"], ["email openid profile"])
 
     def test_source_callback(self):
         """test callback view"""
-        self.client.get(
+        res = self.client.get(
             reverse(
                 "authentik_sources_oauth:oauth-client-callback",
                 kwargs={"source_slug": self.source.slug},
             )
         )
+        self.assertEqual(res.status_code, 302)

authentik/sources/oauth/views/redirect.py

@@ -7,6 +7,9 @@ from django.urls import reverse
 from django.views.generic import RedirectView
 from structlog.stdlib import get_logger
 
+from authentik.flows.planner import PLAN_CONTEXT_PENDING_USER, FlowPlan
+from authentik.flows.stage import PLAN_CONTEXT_PENDING_USER_IDENTIFIER
+from authentik.flows.views.executor import SESSION_KEY_PLAN
 from authentik.sources.oauth.models import OAuthSource
 from authentik.sources.oauth.views.base import OAuthClientMixin
 
@@ -30,6 +33,19 @@ class OAuthRedirect(OAuthClientMixin, RedirectView):
             kwargs={"source_slug": source.slug},
         )
 
+    def _try_login_hint_extract(self) -> dict[str, str]:
+        """Check if we're running in a flow and if we have a pending user, use that
+        as login_hint"""
+        params = {}
+        plan: FlowPlan = self.request.session.get(SESSION_KEY_PLAN, None)
+        if not plan:
+            return params
+        if user := plan.context.get(PLAN_CONTEXT_PENDING_USER):
+            params["login_hint"] = user.email
+        if identifier := plan.context.get(PLAN_CONTEXT_PENDING_USER_IDENTIFIER):
+            params["login_hint"] = identifier
+        return params
+
     def get_redirect_url(self, **kwargs) -> str:
         "Build redirect url for a given source."
         slug = kwargs.get("source_slug", "")
@@ -47,4 +63,5 @@ class OAuthRedirect(OAuthClientMixin, RedirectView):
                 params["scope"] = source.additional_scopes[1:].split(" ")
             else:
                 params["scope"] += source.additional_scopes.split(" ")
+        params.update(self._try_login_hint_extract())
         return client.get_redirect_url(params)

authentik/sources/plex/tasks.py

@@ -1,7 +1,6 @@
 """Plex tasks"""
 
 from django.utils.translation import gettext_lazy as _
-from django_dramatiq_postgres.middleware import CurrentTask
 from dramatiq.actor import actor
 from requests import RequestException
 
@@ -9,13 +8,13 @@ from authentik.events.models import Event, EventAction
 from authentik.lib.utils.errors import exception_to_string
 from authentik.sources.plex.models import PlexSource
 from authentik.sources.plex.plex import PlexAuth
-from authentik.tasks.models import Task
+from authentik.tasks.middleware import CurrentTask
 
 
 @actor(description=_("Check the validity of a Plex source."))
 def check_plex_token(source_pk: str):
     """Check the validity of a Plex source."""
-    self: Task = CurrentTask.get_task()
+    self = CurrentTask.get_task()
     sources = PlexSource.objects.filter(pk=source_pk)
     if not sources.exists():
         return

authentik/sources/saml/api/source.py

@@ -1,10 +1,12 @@
 """SAMLSource API Views"""
 
 from django.urls import reverse
+from django.utils.translation import gettext_lazy as _
 from drf_spectacular.utils import extend_schema
 from rest_framework.decorators import action
 from rest_framework.request import Request
 from rest_framework.response import Response
+from rest_framework.serializers import ValidationError
 from rest_framework.viewsets import ModelViewSet
 
 from authentik.core.api.sources import SourceSerializer
@@ -17,6 +19,18 @@ from authentik.sources.saml.processors.metadata import MetadataProcessor
 class SAMLSourceSerializer(SourceSerializer):
     """SAMLSource Serializer"""
 
+    def validate(self, attrs: dict):
+        if attrs.get("verification_kp"):
+            if not attrs.get("signed_assertion") and not attrs.get("signed_response"):
+                raise ValidationError(
+                    _(
+                        "With a Verification Certificate selected, at least one of"
+                        " 'Verify Assertion Signature' or 'Verify Response Signature' "
+                        "must be selected."
+                    )
+                )
+        return super().validate(attrs)
+
     class Meta:
         model = SAMLSource
         fields = SourceSerializer.Meta.fields + [
@@ -34,6 +48,8 @@ class SAMLSourceSerializer(SourceSerializer):
             "signature_algorithm",
             "temporary_user_delete_after",
             "encryption_kp",
+            "signed_assertion",
+            "signed_response",
         ]
 
 
@@ -65,6 +81,8 @@ class SAMLSourceViewSet(UsedByMixin, ModelViewSet):
         "digest_algorithm",
         "signature_algorithm",
         "temporary_user_delete_after",
+        "signed_assertion",
+        "signed_response",
     ]
     search_fields = ["name", "slug"]
     ordering = ["name"]

authentik/sources/saml/migrations/0021_samlsource_signed_assertion_and_more.py

@@ -0,0 +1,23 @@
+# Generated by Django 5.1.12 on 2025-09-26 04:32
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("authentik_sources_saml", "0020_alter_samlsource_name_id_policy"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="samlsource",
+            name="signed_assertion",
+            field=models.BooleanField(default=True),
+        ),
+        migrations.AddField(
+            model_name="samlsource",
+            name="signed_response",
+            field=models.BooleanField(default=False),
+        ),
+    ]