web/test: e2e ak-mdx pipeline via OAuth2 provider docs

The OAuth2 provider docs page exercises the full pipeline in one place: frontmatter title, multiple H2 slugs, `:::caution`/`:::info` admonitions (with and without titles), relative-doc links, external links, and an embedded mermaid sequence diagram. Tests boot the admin UI, provision a fresh OAuth2 provider, navigate to its view page, and assert against the compiled DOM inside `<ak-mdx>`'s shadow root. Co-Authored-By: Agent <279763771+playpen-agent@users.noreply.github.com>
web/bundler/mdx-plugin: pre-render markdown to HTML at build time
2026-05-06 07:02:51 +02:00 · 2026-04-30 15:02:46 +02:00 · 2026-04-30 15:02:46 +02:00 · 2026-04-30 15:02:46 +02:00 · 2026-04-30 15:02:46 +02:00 · 2026-04-30 14:17:48 +02:00
443 changed files with 34527 additions and 10417 deletions
--- a/.cargo/config.toml
+++ b/.cargo/config.toml
@@ -1,5 +1,5 @@
 [alias]
-t = ["nextest", "run"]
+t = ["nextest", "run", "--workspace"]

 [build]
 rustflags = ["--cfg", "tokio_unstable"]
--- a/.cargo/deny.toml
+++ b/.cargo/deny.toml
@@ -1,5 +1,6 @@
 [licenses]
 allow = [
+  "Apache-2.0 WITH LLVM-exception",
  "Apache-2.0",
  "BSD-3-Clause",
  "CC0-1.0",
--- a/.github/actions/setup/action.yml
+++ b/.github/actions/setup/action.yml
@@ -37,7 +37,7 @@ runs:
        sudo rsync -a --delete /tmp/empty/ /usr/local/lib/android/
    - name: Install uv
      if: ${{ contains(inputs.dependencies, 'python') }}
-      uses: astral-sh/setup-uv@cec208311dfd045dd5311c1add060b2062131d57 # v5
+      uses: astral-sh/setup-uv@08807647e7069bb48b6ef5acd8ec9567f424441b # v5
      with:
        enable-cache: true
    - name: Setup python
@@ -64,12 +64,12 @@ runs:
        rustflags: ""
    - name: Setup rust dependencies
      if: ${{ contains(inputs.dependencies, 'rust') }}
-      uses: taiki-e/install-action@5939f3337e40968c39aa70f5ecb1417a92fb25a0 # v2
+      uses: taiki-e/install-action@481c34c1cf3a84c68b5e46f4eccfc82af798415a # v2
      with:
        tool: cargo-deny cargo-machete cargo-llvm-cov nextest
    - name: Setup node (web)
      if: ${{ contains(inputs.dependencies, 'node') }}
-      uses: actions/setup-node@53b83947a5a98c8d113130e565377fae1a50d02f # v4
+      uses: actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e # v4
      with:
        node-version-file: "${{ inputs.working-directory }}web/package.json"
        cache: "npm"
@@ -77,7 +77,7 @@ runs:
        registry-url: "https://registry.npmjs.org"
    - name: Setup node (root)
      if: ${{ contains(inputs.dependencies, 'node') }}
-      uses: actions/setup-node@53b83947a5a98c8d113130e565377fae1a50d02f # v4
+      uses: actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e # v4
      with:
        node-version-file: "${{ inputs.working-directory }}package.json"
        cache: "npm"
--- a/.github/actions/setup/compose.yml
+++ b/.github/actions/setup/compose.yml
@@ -2,7 +2,7 @@ services:
  postgresql:
    image: docker.io/library/postgres:${PSQL_TAG:-16}
    volumes:
-      - db-data:/var/lib/postgresql/data
+      - db-data:/var/lib/postgresql
    command: "-c log_statement=all"
    environment:
      POSTGRES_USER: authentik
--- a/.github/dependabot.yml
+++ b/.github/dependabot.yml
@@ -66,6 +66,14 @@ updates:
      default-days: 7
      semver-major-days: 14
      semver-patch-days: 3
+      exclude:
+        - aws-lc-fips-sys
+        - aws-lc-rs
+        - aws-lc-sys
+        - rustls
+        - rustls-pki-types
+        - rustls-platform-verifier
+        - rustls-webpki

  - package-ecosystem: rust-toolchain
    directory: "/"
--- a/.github/workflows/_reusable-docker-build.yml
+++ b/.github/workflows/_reusable-docker-build.yml
@@ -90,7 +90,7 @@ jobs:
          registry: ghcr.io
          username: ${{ github.repository_owner }}
          password: ${{ secrets.GITHUB_TOKEN }}
-      - uses: int128/docker-manifest-create-action@44422a4b046d55dc036df622039ed3aec43c613c # v2
+      - uses: int128/docker-manifest-create-action@7df7f9e221d927eaadf87db231ddf728047308a4 # v2
        id: build
        with:
          tags: ${{ matrix.tag }}
--- a/.github/workflows/ci-api-docs.yml
+++ b/.github/workflows/ci-api-docs.yml
@@ -33,7 +33,7 @@ jobs:

    steps:
      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v5
-      - uses: actions/setup-node@53b83947a5a98c8d113130e565377fae1a50d02f # v5
+      - uses: actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e # v5
        with:
          node-version-file: website/package.json
          cache: "npm"
@@ -71,7 +71,7 @@ jobs:
        with:
          name: api-docs
          path: website/api/build
-      - uses: actions/setup-node@53b83947a5a98c8d113130e565377fae1a50d02f # v5
+      - uses: actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e # v5
        with:
          node-version-file: website/package.json
          cache: "npm"
--- a/.github/workflows/ci-aws-cfn.yml
+++ b/.github/workflows/ci-aws-cfn.yml
@@ -24,7 +24,7 @@ jobs:
      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v5
      - name: Setup authentik env
        uses: ./.github/actions/setup
-      - uses: actions/setup-node@53b83947a5a98c8d113130e565377fae1a50d02f # v5
+      - uses: actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e # v5
        with:
          node-version-file: lifecycle/aws/package.json
          cache: "npm"
--- a/.github/workflows/ci-docs.yml
+++ b/.github/workflows/ci-docs.yml
@@ -36,7 +36,7 @@ jobs:
      NODE_ENV: production
    steps:
      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v5
-      - uses: actions/setup-node@53b83947a5a98c8d113130e565377fae1a50d02f # v5
+      - uses: actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e # v5
        with:
          node-version-file: website/package.json
          cache: "npm"
@@ -53,7 +53,7 @@ jobs:
      NODE_ENV: production
    steps:
      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v5
-      - uses: actions/setup-node@53b83947a5a98c8d113130e565377fae1a50d02f # v5
+      - uses: actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e # v5
        with:
          node-version-file: website/package.json
          cache: "npm"
--- a/.github/workflows/ci-main.yml
+++ b/.github/workflows/ci-main.yml
@@ -127,7 +127,10 @@ jobs:
        with:
          postgresql_version: ${{ matrix.psql }}
      - name: run migrations to stable
-        run: uv run python -m lifecycle.migrate
+        run: |
+          docker ps
+          docker logs setup-postgresql-1
+          uv run python -m lifecycle.migrate
      - name: checkout current code
        run: |
          set -x
--- a/.github/workflows/ci-outpost.yml
+++ b/.github/workflows/ci-outpost.yml
@@ -145,7 +145,7 @@ jobs:
      - uses: actions/setup-go@4a3601121dd01d1626a1e23e37211e3254c1c06c # v6
        with:
          go-version-file: "go.mod"
-      - uses: actions/setup-node@53b83947a5a98c8d113130e565377fae1a50d02f # v5
+      - uses: actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e # v5
        with:
          node-version-file: web/package.json
          cache: "npm"
--- a/.github/workflows/ci-web.yml
+++ b/.github/workflows/ci-web.yml
@@ -32,7 +32,7 @@ jobs:
            project: web
    steps:
      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v5
-      - uses: actions/setup-node@53b83947a5a98c8d113130e565377fae1a50d02f # v5
+      - uses: actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e # v5
        with:
          node-version-file: ${{ matrix.project }}/package.json
          cache: "npm"
@@ -47,7 +47,7 @@ jobs:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v5
-      - uses: actions/setup-node@53b83947a5a98c8d113130e565377fae1a50d02f # v5
+      - uses: actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e # v5
        with:
          node-version-file: web/package.json
          cache: "npm"
@@ -73,7 +73,7 @@ jobs:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v5
-      - uses: actions/setup-node@53b83947a5a98c8d113130e565377fae1a50d02f # v5
+      - uses: actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e # v5
        with:
          node-version-file: web/package.json
          cache: "npm"
--- a/.github/workflows/gen-image-compress.yml
+++ b/.github/workflows/gen-image-compress.yml
@@ -38,7 +38,7 @@ jobs:
          token: ${{ steps.generate_token.outputs.token }}
      - name: Compress images
        id: compress
-        uses: calibreapp/image-actions@4f7260f5dbd809ec86d03721c1ad71b8a841d3e0 # main
+        uses: calibreapp/image-actions@e2cc8db5d49c849e00844dfebf01438318e96fa2 # main
        with:
          GITHUB_TOKEN: ${{ steps.generate_token.outputs.token }}
          compressOnly: ${{ github.event_name != 'pull_request' }}
--- a/.github/workflows/packages-npm-publish.yml
+++ b/.github/workflows/packages-npm-publish.yml
@@ -35,13 +35,13 @@ jobs:
      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v5
        with:
          fetch-depth: 2
-      - uses: actions/setup-node@53b83947a5a98c8d113130e565377fae1a50d02f # v5
+      - uses: actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e # v5
        with:
          node-version-file: ${{ matrix.package }}/package.json
          registry-url: "https://registry.npmjs.org"
      - name: Get changed files
        id: changed-files
-        uses: tj-actions/changed-files@22103cc46bda19c2b464ffe86db46df6922fd323 # 24d32ffd492484c1d75e0c0b894501ddb9d30d62
+        uses: tj-actions/changed-files@9426d40962ed5378910ee2e21d5f8c6fcbf2dd96 # 24d32ffd492484c1d75e0c0b894501ddb9d30d62
        with:
          files: |
            ${{ matrix.package }}/package.json
--- a/.github/workflows/release-publish.yml
+++ b/.github/workflows/release-publish.yml
@@ -87,7 +87,7 @@ jobs:
      - uses: actions/setup-go@4a3601121dd01d1626a1e23e37211e3254c1c06c # v6
        with:
          go-version-file: "go.mod"
-      - uses: actions/setup-node@53b83947a5a98c8d113130e565377fae1a50d02f # v5
+      - uses: actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e # v5
        with:
          node-version-file: web/package.json
          cache: "npm"
@@ -151,7 +151,7 @@ jobs:
      - uses: actions/setup-go@4a3601121dd01d1626a1e23e37211e3254c1c06c # v6
        with:
          go-version-file: "go.mod"
-      - uses: actions/setup-node@53b83947a5a98c8d113130e565377fae1a50d02f # v5
+      - uses: actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e # v5
        with:
          node-version-file: web/package.json
          cache: "npm"
--- a/1
+++ b/1
@@ -14,6 +14,7 @@ pyproject.toml                          @goauthentik/backend
 uv.lock                                 @goauthentik/backend
 Cargo.toml                              @goauthentik/backend
 Cargo.lock                              @goauthentik/backend
+build.rs                                @goauthentik/backend
 go.mod                                  @goauthentik/backend
 go.sum                                  @goauthentik/backend
 .cargo/                                 @goauthentik/backend
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -17,6 +17,18 @@ version = "2.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa"

+[[package]]
+name = "ahash"
+version = "0.8.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5a15f179cd60c4584b8a8c596927aadc462e27f2ca70c04e0071964a73ba7a75"
+dependencies = [
+ "cfg-if",
+ "once_cell",
+ "version_check",
+ "zerocopy",
+]
+
 [[package]]
 name = "aho-corasick"
 version = "1.1.4"
@@ -106,6 +118,37 @@ dependencies = [
 "rustversion",
 ]

+[[package]]
+name = "argh"
+version = "0.1.19"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "211818e820cda9ca6f167a64a5c808837366a6dfd807157c64c1304c486cd033"
+dependencies = [
+ "argh_derive",
+ "argh_shared",
+]
+
+[[package]]
+name = "argh_derive"
+version = "0.1.19"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c442a9d18cef5dde467405d27d461d080d68972d6d0dfd0408265b6749ec427d"
+dependencies = [
+ "argh_shared",
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
+name = "argh_shared"
+version = "0.1.19"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e5ade012bac4db278517a0132c8c10c6427025868dca16c801087c28d5a411f1"
+dependencies = [
+ "serde",
+]
+
 [[package]]
 name = "arraydeque"
 version = "0.5.1"
@@ -138,6 +181,30 @@ version = "1.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0"

+[[package]]
+name = "authentik"
+version = "2026.5.0-rc1"
+dependencies = [
+ "arc-swap",
+ "argh",
+ "authentik-axum",
+ "authentik-common",
+ "axum",
+ "color-eyre",
+ "eyre",
+ "hyper-unix-socket",
+ "hyper-util",
+ "metrics",
+ "metrics-exporter-prometheus",
+ "nix 0.31.2",
+ "pyo3",
+ "pyo3-build-config",
+ "sqlx",
+ "tokio",
+ "tracing",
+ "uuid",
+]
+
 [[package]]
 name = "authentik-axum"
 version = "2026.5.0-rc1"
@@ -150,6 +217,7 @@ dependencies = [
 "eyre",
 "forwarded-header-value",
 "futures",
+ "pin-project-lite",
 "tokio",
 "tokio-rustls",
 "tower",
@@ -231,9 +299,9 @@ dependencies = [

 [[package]]
 name = "aws-lc-rs"
-version = "1.16.2"
+version = "1.16.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a054912289d18629dc78375ba2c3726a3afe3ff71b4edba9dedfca0e3446d1fc"
+checksum = "0ec6fb3fe69024a75fa7e1bfb48aa6cf59706a101658ea01bfd33b2b248a038f"
 dependencies = [
 "aws-lc-fips-sys",
 "aws-lc-sys",
@@ -243,9 +311,9 @@ dependencies = [

 [[package]]
 name = "aws-lc-sys"
-version = "0.39.0"
+version = "0.40.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1fa7e52a4c5c547c741610a2c6f123f3881e409b714cd27e6798ef020c514f0a"
+checksum = "f50037ee5e1e41e7b8f9d161680a725bd1626cb6f8c7e901f91f942850852fe7"
 dependencies = [
 "cc",
 "cmake",
@@ -255,9 +323,9 @@ dependencies = [

 [[package]]
 name = "axum"
-version = "0.8.8"
+version = "0.8.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8b52af3cb4058c895d37317bb27508dccc8e5f2d39454016b297bf4a400597b8"
+checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90"
 dependencies = [
 "axum-core",
 "axum-macros",
@@ -311,9 +379,9 @@ dependencies = [

 [[package]]
 name = "axum-macros"
-version = "0.5.0"
+version = "0.5.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "604fde5e028fea851ce1d8570bbdc034bec850d157f7569d10f347d06808c05c"
+checksum = "7aa268c23bfbbd2c4363b9cd302a4f504fb2a9dfe7e3451d66f35dd392e20aca"
 dependencies = [
 "proc-macro2",
 "quote",
@@ -510,9 +578,9 @@ dependencies = [

 [[package]]
 name = "clap"
-version = "4.6.0"
+version = "4.6.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b193af5b67834b676abd72466a96c1024e6a6ad978a1f484bd90b85c94041351"
+checksum = "1ddb117e43bbf7dacf0a4190fef4d345b9bad68dfc649cb349e7d17d28428e51"
 dependencies = [
 "clap_builder",
 "clap_derive",
@@ -532,9 +600,9 @@ dependencies = [

 [[package]]
 name = "clap_derive"
-version = "4.6.0"
+version = "4.6.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1110bd8a634a1ab8cb04345d8d878267d57c3cf1b38d91b71af6686408bbca6a"
+checksum = "f2ce8604710f6733aa641a2b3731eaa1e8b3d9973d5e3565da11800813f997a9"
 dependencies = [
 "heck",
 "proc-macro2",
@@ -567,6 +635,33 @@ dependencies = [
 "cc",
 ]

+[[package]]
+name = "color-eyre"
+version = "0.6.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e5920befb47832a6d61ee3a3a846565cfa39b331331e68a3b1d1116630f2f26d"
+dependencies = [
+ "backtrace",
+ "color-spantrace",
+ "eyre",
+ "indenter",
+ "once_cell",
+ "owo-colors",
+ "tracing-error",
+]
+
+[[package]]
+name = "color-spantrace"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b8b88ea9df13354b55bc7234ebcce36e6ef896aca2e42a15de9e10edce01b427"
+dependencies = [
+ "once_cell",
+ "owo-colors",
+ "tracing-core",
+ "tracing-error",
+]
+
 [[package]]
 name = "colorchoice"
 version = "1.0.5"
@@ -728,6 +823,15 @@ dependencies = [
 "crossbeam-utils",
 ]

+[[package]]
+name = "crossbeam-epoch"
+version = "0.9.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e"
+dependencies = [
+ "crossbeam-utils",
+]
+
 [[package]]
 name = "crossbeam-queue"
 version = "0.3.12"
@@ -977,6 +1081,12 @@ version = "0.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2"

+[[package]]
+name = "foldhash"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb"
+
 [[package]]
 name = "form_urlencoded"
 version = "1.2.2"
@@ -1209,7 +1319,7 @@ checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1"
 dependencies = [
 "allocator-api2",
 "equivalent",
- "foldhash",
+ "foldhash 0.1.5",
 ]

 [[package]]
@@ -1217,6 +1327,9 @@ name = "hashbrown"
 version = "0.16.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100"
+dependencies = [
+ "foldhash 0.2.0",
+]

 [[package]]
 name = "hashlink"
@@ -1343,9 +1456,9 @@ checksum = "135b12329e5e3ce057a9f972339ea52bc954fe1e9358ef27f95e89716fbc5424"

 [[package]]
 name = "hyper"
-version = "1.8.1"
+version = "1.9.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2ab2d4f250c3d7b1c9fcdff1cece94ea4e2dfbec68614f7b87cb205f24ca9d11"
+checksum = "6299f016b246a94207e63da54dbe807655bf9e00044f73ded42c3ac5305fbcca"
 dependencies = [
 "atomic-waker",
 "bytes",
@@ -1358,7 +1471,6 @@ dependencies = [
 "httpdate",
 "itoa",
 "pin-project-lite",
- "pin-utils",
 "smallvec",
 "tokio",
 "want",
@@ -1393,6 +1505,20 @@ dependencies = [
 "tower-service",
 ]

+[[package]]
+name = "hyper-unix-socket"
+version = "0.6.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "88978f1d73da0eb87d86555fcc40cbdd87bc86eb6525710b89db8c9278ec6a59"
+dependencies = [
+ "bytes",
+ "hyper",
+ "hyper-util",
+ "pin-project-lite",
+ "tokio",
+ "tower-service",
+]
+
 [[package]]
 name = "hyper-util"
 version = "0.1.20"
@@ -1850,6 +1976,46 @@ version = "2.8.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79"

+[[package]]
+name = "metrics"
+version = "0.24.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5d5312e9ba3771cfa961b585728215e3d972c950a3eed9252aa093d6301277e8"
+dependencies = [
+ "ahash",
+ "portable-atomic",
+]
+
+[[package]]
+name = "metrics-exporter-prometheus"
+version = "0.18.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3589659543c04c7dc5526ec858591015b87cd8746583b51b48ef4353f99dbcda"
+dependencies = [
+ "base64 0.22.1",
+ "indexmap",
+ "metrics",
+ "metrics-util",
+ "quanta",
+ "thiserror 2.0.18",
+]
+
+[[package]]
+name = "metrics-util"
+version = "0.20.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cdfb1365fea27e6dd9dc1dbc19f570198bc86914533ad639dae939635f096be4"
+dependencies = [
+ "crossbeam-epoch",
+ "crossbeam-utils",
+ "hashbrown 0.16.1",
+ "metrics",
+ "quanta",
+ "rand 0.9.2",
+ "rand_xoshiro",
+ "sketches-ddsketch",
+]
+
 [[package]]
 name = "mime"
 version = "0.3.17"
@@ -1981,7 +2147,7 @@ dependencies = [
 "num-integer",
 "num-iter",
 "num-traits",
- "rand 0.8.5",
+ "rand 0.8.6",
 "smallvec",
 "zeroize",
 ]
@@ -2233,6 +2399,12 @@ dependencies = [
 "windows-sys 0.61.2",
 ]

+[[package]]
+name = "owo-colors"
+version = "4.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d211803b9b6b570f68772237e415a029d5a50c65d382910b879fb19d3271f94d"
+
 [[package]]
 name = "parking"
 version = "2.2.1"
@@ -2309,12 +2481,6 @@ version = "0.2.17"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd"

-[[package]]
-name = "pin-utils"
-version = "0.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"
-
 [[package]]
 name = "pkcs1"
 version = "0.7.5"
@@ -2348,6 +2514,12 @@ version = "0.2.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b4596b6d070b27117e987119b4dac604f3c58cfb0b191112e24771b2faeac1a6"

+[[package]]
+name = "portable-atomic"
+version = "1.13.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c33a9471896f1c69cecef8d20cbe2f7accd12527ce60845ff44c153bb2a21b49"
+
 [[package]]
 name = "potential_utf"
 version = "0.1.4"
@@ -2423,6 +2595,79 @@ dependencies = [
 "prost",
 ]

+[[package]]
+name = "pyo3"
+version = "0.28.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "91fd8e38a3b50ed1167fb981cd6fd60147e091784c427b8f7183a7ee32c31c12"
+dependencies = [
+ "libc",
+ "once_cell",
+ "portable-atomic",
+ "pyo3-build-config",
+ "pyo3-ffi",
+ "pyo3-macros",
+]
+
+[[package]]
+name = "pyo3-build-config"
+version = "0.28.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e368e7ddfdeb98c9bca7f8383be1648fd84ab466bf2bc015e94008db6d35611e"
+dependencies = [
+ "target-lexicon",
+]
+
+[[package]]
+name = "pyo3-ffi"
+version = "0.28.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7f29e10af80b1f7ccaf7f69eace800a03ecd13e883acfacc1e5d0988605f651e"
+dependencies = [
+ "libc",
+ "pyo3-build-config",
+]
+
+[[package]]
+name = "pyo3-macros"
+version = "0.28.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "df6e520eff47c45997d2fc7dd8214b25dd1310918bbb2642156ef66a67f29813"
+dependencies = [
+ "proc-macro2",
+ "pyo3-macros-backend",
+ "quote",
+ "syn",
+]
+
+[[package]]
+name = "pyo3-macros-backend"
+version = "0.28.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c4cdc218d835738f81c2338f822078af45b4afdf8b2e33cbb5916f108b813acb"
+dependencies = [
+ "heck",
+ "proc-macro2",
+ "pyo3-build-config",
+ "quote",
+ "syn",
+]
+
+[[package]]
+name = "quanta"
+version = "0.12.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f3ab5a9d756f0d97bdc89019bd2e4ea098cf9cde50ee7564dde6b81ccc8f06c7"
+dependencies = [
+ "crossbeam-utils",
+ "libc",
+ "once_cell",
+ "raw-cpuid",
+ "wasi",
+ "web-sys",
+ "winapi",
+]
+
 [[package]]
 name = "quinn"
 version = "0.11.9"
@@ -2502,9 +2747,9 @@ checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf"

 [[package]]
 name = "rand"
-version = "0.8.5"
+version = "0.8.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "34af8d1a0e25924bc5b7c43c079c942339d8f0a8b57c39049bef581b46327404"
+checksum = "5ca0ecfa931c29007047d1bc58e623ab12e5590e8c7cc53200d5202b69266d8a"
 dependencies = [
 "libc",
 "rand_chacha 0.3.1",
@@ -2559,6 +2804,24 @@ dependencies = [
 "getrandom 0.3.4",
 ]

+[[package]]
+name = "rand_xoshiro"
+version = "0.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f703f4665700daf5512dcca5f43afa6af89f09db47fb56be587f80636bda2d41"
+dependencies = [
+ "rand_core 0.9.5",
+]
+
+[[package]]
+name = "raw-cpuid"
+version = "11.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "498cd0dc59d73224351ee52a95fee0f1a617a2eae0e7d9d720cc622c73a54186"
+dependencies = [
+ "bitflags 2.11.0",
+]
+
 [[package]]
 name = "redox_syscall"
 version = "0.5.18"
@@ -2737,9 +3000,9 @@ dependencies = [

 [[package]]
 name = "rustls"
-version = "0.23.37"
+version = "0.23.40"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "758025cb5fccfd3bc2fd74708fd4682be41d99e5dff73c377c0646c6012c73a4"
+checksum = "ef86cd5876211988985292b91c96a8f2d298df24e75989a43a3c73f2d4d8168b"
 dependencies = [
 "aws-lc-rs",
 "log",
@@ -2802,9 +3065,9 @@ checksum = "f87165f0995f63a9fbeea62b64d10b4d9d8e78ec6d7d51fb2125fda7bb36788f"

 [[package]]
 name = "rustls-webpki"
-version = "0.103.12"
+version = "0.103.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8279bb85272c9f10811ae6a6c547ff594d6a7f3c6c6b02ee9726d1d0dcfcdd06"
+checksum = "61c429a8649f110dddef65e2a5ad240f747e85f7758a6bccc7e5777bd33f756e"
 dependencies = [
 "aws-lc-rs",
 "ring",
@@ -3151,6 +3414,12 @@ version = "0.3.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "703d5c7ef118737c72f1af64ad2f6f8c5e1921f818cdcb97b8fe6fc69bf66214"

+[[package]]
+name = "sketches-ddsketch"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0c6f73aeb92d671e0cc4dca167e59b2deb6387c375391bc99ee743f326994a2b"
+
 [[package]]
 name = "slab"
 version = "0.4.12"
@@ -3315,7 +3584,7 @@ dependencies = [
 "memchr",
 "once_cell",
 "percent-encoding",
- "rand 0.8.5",
+ "rand 0.8.6",
 "rsa",
 "serde",
 "sha1",
@@ -3356,7 +3625,7 @@ dependencies = [
 "md-5",
 "memchr",
 "once_cell",
- "rand 0.8.5",
+ "rand 0.8.6",
 "serde",
 "serde_json",
 "sha2",
@@ -3476,6 +3745,12 @@ dependencies = [
 "libc",
 ]

+[[package]]
+name = "target-lexicon"
+version = "0.13.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "adb6935a6f5c20170eeceb1a3835a49e12e19d792f6dd344ccc76a985ca5a6ca"
+
 [[package]]
 name = "tempfile"
 version = "3.27.0"
@@ -3598,9 +3873,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20"

 [[package]]
 name = "tokio"
-version = "1.51.1"
+version = "1.52.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f66bf9585cda4b724d3e78ab34b73fb2bbaba9011b9bfdf69dc836382ea13b8c"
+checksum = "b67dee974fe86fd92cc45b7a95fdd2f99a36a6d7b0d431a231178d3d670bbcc6"
 dependencies = [
 "bytes",
 "libc",
@@ -3658,9 +3933,9 @@ dependencies = [

 [[package]]
 name = "tokio-tungstenite"
-version = "0.28.0"
+version = "0.29.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d25a406cddcc431a75d3d9afc6a7c0f7428d4891dd973e4d54c56b46127bf857"
+checksum = "8f72a05e828585856dacd553fba484c242c46e391fb0e58917c942ee9202915c"
 dependencies = [
 "futures-util",
 "log",
@@ -3869,9 +4144,9 @@ checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b"

 [[package]]
 name = "tungstenite"
-version = "0.28.0"
+version = "0.29.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8628dcc84e5a09eb3d8423d6cb682965dea9133204e8fb3efee74c2a0c259442"
+checksum = "6c01152af293afb9c7c2a57e4b559c5620b421f6d133261c60dd2d0cdb38e6b8"
 dependencies = [
 "bytes",
 "data-encoding",
@@ -3881,7 +4156,6 @@ dependencies = [
 "rand 0.9.2",
 "sha1",
 "thiserror 2.0.18",
- "utf-8",
 ]

 [[package]]
@@ -3991,12 +4265,6 @@ dependencies = [
 "serde_derive",
 ]

-[[package]]
-name = "utf-8"
-version = "0.7.6"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9"
-
 [[package]]
 name = "utf8-zero"
 version = "0.8.1"
@@ -4017,9 +4285,9 @@ checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821"

 [[package]]
 name = "uuid"
-version = "1.23.0"
+version = "1.23.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5ac8b6f42ead25368cf5b098aeb3dc8a1a2c05a3eee8a9a1a68c640edbfc79d9"
+checksum = "ddd74a9687298c6858e9b88ec8935ec45d22e8fd5e6394fa1bd4e99a87789c76"
 dependencies = [
 "getrandom 0.4.2",
 "js-sys",
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -20,11 +20,13 @@ publish = false

 [workspace.dependencies]
 arc-swap = "= 1.9.1"
+argh = "= 0.1.19"
 axum-server = { version = "= 0.8.0", features = ["tls-rustls-no-provider"] }
-aws-lc-rs = { version = "= 1.16.2", features = ["fips"] }
-axum = { version = "= 0.8.8", features = ["http2", "macros", "ws"] }
-clap = { version = "= 4.6.0", features = ["derive", "env"] }
+aws-lc-rs = { version = "= 1.16.3", features = ["fips"] }
+axum = { version = "= 0.8.9", features = ["http2", "macros", "ws"] }
+clap = { version = "= 4.6.1", features = ["derive", "env"] }
 client-ip = { version = "0.2.1", features = ["forwarded-header"] }
+color-eyre = "= 0.6.5"
 colored = "= 3.1.1"
 config-rs = { package = "config", version = "= 0.15.22", default-features = false, features = [
  "json",
@@ -37,11 +39,17 @@ eyre = "= 0.6.12"
 forwarded-header-value = "= 0.1.1"
 futures = "= 0.3.32"
 glob = "= 0.3.3"
+hyper-unix-socket = "= 0.6.1"
+hyper-util = "= 0.1.20"
 ipnet = { version = "= 2.12.0", features = ["serde"] }
 json-subscriber = "= 0.2.8"
-nix = { version = "= 0.31.2", features = ["signal"] }
+metrics = "= 0.24.3"
+metrics-exporter-prometheus = { version = "= 0.18.1", default-features = false }
+nix = { version = "= 0.31.2", features = ["hostname", "signal"] }
 notify = "= 8.2.0"
 pin-project-lite = "= 0.2.17"
+pyo3 = "= 0.28.3"
+pyo3-build-config = "= 0.28.3"
 regex = "= 1.12.3"
 reqwest = { version = "= 0.13.2", features = [
  "form",
@@ -58,7 +66,7 @@ reqwest-middleware = { version = "= 0.5.1", features = [
  "query",
  "rustls",
 ] }
-rustls = { version = "= 0.23.37", features = ["fips"] }
+rustls = { version = "= 0.23.40", features = ["fips"] }
 sentry = { version = "= 0.47.0", default-features = false, features = [
  "backtrace",
  "contexts",
@@ -89,7 +97,7 @@ sqlx = { version = "= 0.8.6", default-features = false, features = [
 tempfile = "= 3.27.0"
 thiserror = "= 2.0.18"
 time = { version = "= 0.3.47", features = ["macros"] }
-tokio = { version = "= 1.51.1", features = ["full", "tracing"] }
+tokio = { version = "= 1.52.1", features = ["full", "tracing"] }
 tokio-retry2 = "= 0.9.1"
 tokio-rustls = "= 0.26.4"
 tokio-util = { version = "= 0.7.18", features = ["full"] }
@@ -104,18 +112,12 @@ tracing-subscriber = { version = "= 0.3.23", features = [
  "tracing-log",
 ] }
 url = "= 2.5.8"
-uuid = { version = "= 1.23.0", features = ["serde", "v4"] }
+uuid = { version = "= 1.23.1", features = ["serde", "v4"] }

+ak-axum = { package = "authentik-axum", version = "2026.5.0-rc1", path = "./packages/ak-axum" }
 ak-client = { package = "authentik-client", version = "2026.5.0-rc1", path = "./packages/client-rust" }
 ak-common = { package = "authentik-common", version = "2026.5.0-rc1", path = "./packages/ak-common", default-features = false }

-[profile.dev.package.backtrace]
-opt-level = 3
-
-[profile.release]
-lto = true
-debug = 2
-
 [workspace.lints.rust]
 ambiguous_negative_literals = "warn"
 closure_returning_async_block = "warn"
@@ -229,3 +231,57 @@ unused_trait_names = "warn"
 unwrap_in_result = "warn"
 unwrap_used = "warn"
 verbose_file_reads = "warn"
+
+[profile.dev.package.backtrace]
+opt-level = 3
+
+[profile.dev]
+panic = "abort"
+
+[profile.release]
+debug = 2
+lto = "fat"
+# Because of the async runtime, we want to die straightaway if we panic.
+panic = "abort"
+strip = true
+
+[package]
+name = "authentik"
+version.workspace = true
+authors.workspace = true
+edition.workspace = true
+readme.workspace = true
+homepage.workspace = true
+repository.workspace = true
+license-file.workspace = true
+publish.workspace = true
+
+[features]
+default = ["core", "proxy"]
+core = ["ak-common/core", "dep:pyo3", "dep:sqlx"]
+proxy = ["ak-common/proxy"]
+
+[build-dependencies]
+pyo3-build-config.workspace = true
+
+[dependencies]
+ak-axum.workspace = true
+ak-common.workspace = true
+arc-swap.workspace = true
+argh.workspace = true
+axum.workspace = true
+color-eyre.workspace = true
+eyre.workspace = true
+hyper-unix-socket.workspace = true
+hyper-util.workspace = true
+metrics.workspace = true
+metrics-exporter-prometheus.workspace = true
+nix.workspace = true
+pyo3 = { workspace = true, optional = true }
+sqlx = { workspace = true, optional = true }
+tokio.workspace = true
+tracing.workspace = true
+uuid.workspace = true
+
+[lints]
+workspace = true
--- a/3
+++ b/3
@@ -115,6 +115,9 @@ run-server:  ## Run the main authentik server process
 run-worker:  ## Run the main authentik worker process
 	$(UV) run ak worker

+run-worker-watch:  ## Run the authentik worker, with auto reloading
+	watchexec --on-busy-update=restart --stop-signal=SIGINT --exts py,rs --no-meta --notify -- $(UV) run ak worker
+
 core-i18n-extract:
 	$(UV) run ak makemessages \
 		--add-location file \
--- a/authentik/admin/api/meta.py
+++ b/authentik/admin/api/meta.py
@@ -1,16 +1,13 @@
 """Meta API"""

-from django.apps import apps
 from drf_spectacular.utils import extend_schema
-from rest_framework.fields import BooleanField, CharField
+from rest_framework.fields import CharField
 from rest_framework.permissions import IsAuthenticated
 from rest_framework.request import Request
 from rest_framework.response import Response
 from rest_framework.viewsets import ViewSet

-from authentik.api.validation import validate
 from authentik.core.api.utils import PassiveSerializer
-from authentik.core.models import AttributesMixin
 from authentik.lib.api import Models
 from authentik.lib.utils.reflection import get_apps

@@ -39,19 +36,12 @@ class AppsViewSet(ViewSet):
 class ModelViewSet(ViewSet):
    """Read-only view list all installed models"""

-    class ModelFilterSerializer(PassiveSerializer):
-        filter_has_attributes = BooleanField(allow_null=True, default=None)
-
    permission_classes = [IsAuthenticated]

-    @extend_schema(responses={200: AppSerializer(many=True)}, parameters=[ModelFilterSerializer])
-    @validate(ModelFilterSerializer, "query")
-    def list(self, request: Request, query: ModelFilterSerializer) -> Response:
+    @extend_schema(responses={200: AppSerializer(many=True)})
+    def list(self, request: Request) -> Response:
        """Read-only view list all installed models"""
        data = []
        for name, label in Models.choices:
-            if query.validated_data["filter_has_attributes"]:
-                if not issubclass(apps.get_model(name), AttributesMixin):
-                    continue
            data.append({"name": name, "label": label})
        return Response(AppSerializer(data, many=True).data)
--- a/authentik/admin/files/backends/s3.py
+++ b/authentik/admin/files/backends/s3.py
@@ -1,7 +1,7 @@
 from collections.abc import Generator, Iterator
 from contextlib import contextmanager
 from tempfile import SpooledTemporaryFile
-from urllib.parse import urlsplit
+from urllib.parse import urlsplit, urlunsplit

 import boto3
 from botocore.config import Config
@@ -164,16 +164,19 @@ class S3Backend(ManageableBackend):
        )

        def _file_url(name: str, request: HttpRequest | None) -> str:
+            client = self.client
            params = {
                "Bucket": self.bucket_name,
                "Key": f"{self.base_path}/{name}",
            }

-            url = self.client.generate_presigned_url(
-                "get_object",
-                Params=params,
-                ExpiresIn=expires_in,
-                HttpMethod="GET",
+            operation_name = "GetObject"
+            operation_model = client.meta.service_model.operation_model(operation_name)
+            request_dict = client._convert_to_request_dict(
+                params,
+                operation_model,
+                endpoint_url=client.meta.endpoint_url,
+                context={"is_presign_request": True},
            )

            # Support custom domain for S3-compatible storage (so not AWS)
@@ -183,9 +186,8 @@ class S3Backend(ManageableBackend):
                CONFIG.get(f"storage.{self.name}.custom_domain", None),
            )
            if custom_domain:
-                parsed = urlsplit(url)
                scheme = "https" if use_https else "http"
-                path = parsed.path
+                path = request_dict["url_path"]

                # When using path-style addressing, the presigned URL contains the bucket
                # name in the path (e.g., /bucket-name/key). Since custom_domain must
@@ -200,9 +202,22 @@ class S3Backend(ManageableBackend):
                if not path.startswith("/"):
                    path = f"/{path}"

-                url = f"{scheme}://{custom_domain}{path}?{parsed.query}"
+                custom_base = urlsplit(f"{scheme}://{custom_domain}")

-            return url
+                # Sign the final public URL instead of signing the internal S3 endpoint and
+                # rewriting it afterwards. Presigned SigV4 URLs include the host header in the
+                # canonical request, so post-sign host changes break strict backends like RustFS.
+                public_path = f"{custom_base.path.rstrip('/')}{path}" if custom_base.path else path
+                request_dict["url_path"] = public_path
+                request_dict["url"] = urlunsplit(
+                    (custom_base.scheme, custom_base.netloc, public_path, "", "")
+                )
+
+            return client._request_signer.generate_presigned_url(
+                request_dict,
+                operation_name,
+                expires_in=expires_in,
+            )

        if use_cache:
            return self._cache_get_or_set(name, request, _file_url, expires_in)
--- a/authentik/admin/files/backends/tests/test_s3_backend.py
+++ b/authentik/admin/files/backends/tests/test_s3_backend.py
@@ -1,4 +1,5 @@
 from unittest import skipUnless
+from urllib.parse import parse_qs, urlsplit

 from botocore.exceptions import UnsupportedSignatureVersionError
 from django.test import TestCase
@@ -168,6 +169,44 @@ class TestS3Backend(FileTestS3BackendMixin, TestCase):
            f"URL: {url}",
        )

+    @CONFIG.patch("storage.s3.secure_urls", False)
+    @CONFIG.patch("storage.s3.addressing_style", "path")
+    def test_file_url_custom_domain_resigns_for_custom_host(self):
+        """Test presigned URLs are signed for the custom domain host.
+
+        Host-changing custom domains must produce a signature query string for
+        the public host, not reuse the internal endpoint signature.
+        """
+        bucket_name = self.media_s3_bucket_name
+        key_name = "application-icons/test.svg"
+        custom_domain = f"files.example.test:8020/{bucket_name}"
+
+        endpoint_signed_url = self.media_s3_backend.client.generate_presigned_url(
+            "get_object",
+            Params={
+                "Bucket": bucket_name,
+                "Key": f"{self.media_s3_backend.base_path}/{key_name}",
+            },
+            ExpiresIn=900,
+            HttpMethod="GET",
+        )
+
+        with CONFIG.patch("storage.media.s3.custom_domain", custom_domain):
+            custom_url = self.media_s3_backend.file_url(key_name, use_cache=False)
+
+        endpoint_parts = urlsplit(endpoint_signed_url)
+        custom_parts = urlsplit(custom_url)
+
+        self.assertEqual(custom_parts.scheme, "http")
+        self.assertEqual(custom_parts.netloc, "files.example.test:8020")
+        self.assertEqual(parse_qs(custom_parts.query)["X-Amz-SignedHeaders"], ["host"])
+        self.assertNotEqual(
+            custom_parts.query,
+            endpoint_parts.query,
+            "Custom-domain URLs must be signed for the public host, not reuse the endpoint "
+            "signature query string.",
+        )
+
    def test_themed_urls_without_theme_variable(self):
        """Test themed_urls returns None when filename has no %(theme)s"""
        result = self.media_s3_backend.themed_urls("logo.png")
--- a/authentik/api/tests/test_viewsets.py
+++ b/authentik/api/tests/test_viewsets.py
@@ -1,31 +1,73 @@
 """authentik API Modelviewset tests"""

 from collections.abc import Callable
+from urllib.parse import urlencode

 from django.test import TestCase
 from rest_framework.viewsets import ModelViewSet, ReadOnlyModelViewSet

+from authentik.admin.api.version_history import VersionHistoryViewSet
 from authentik.api.v3.urls import router
+from authentik.core.tests.utils import RequestFactory, create_test_admin_user
+from authentik.lib.generators import generate_id
+from authentik.tenants.api.domains import DomainViewSet
+from authentik.tenants.api.tenants import TenantViewSet
+from authentik.tenants.utils import get_current_tenant


 class TestModelViewSets(TestCase):
    """Test Viewset"""

+    def setUp(self):
+        self.user = create_test_admin_user()
+        self.factory = RequestFactory()

-def viewset_tester_factory(test_viewset: type[ModelViewSet]) -> Callable:
+
+def viewset_tester_factory(test_viewset: type[ModelViewSet], full=True) -> dict[str, Callable]:
    """Test Viewset"""

-    def tester(self: TestModelViewSets):
-        self.assertIsNotNone(getattr(test_viewset, "search_fields", None))
+    def test_attrs(self: TestModelViewSets) -> None:
+        """Test attributes we require on all viewsets"""
        self.assertIsNotNone(getattr(test_viewset, "ordering", None))
+        self.assertIsNotNone(getattr(test_viewset, "search_fields", None))
        filterset_class = getattr(test_viewset, "filterset_class", None)
        if not filterset_class:
            self.assertIsNotNone(getattr(test_viewset, "filterset_fields", None))

-    return tester
+    def test_ordering(self: TestModelViewSets) -> None:
+        """Test that all ordering fields are correct"""
+        view = test_viewset.as_view({"get": "list"})
+        for ordering_field in test_viewset.ordering:
+            with self.subTest(ordering_field):
+                req = self.factory.get(
+                    f"/?{urlencode({'ordering': ordering_field}, doseq=True)}", user=self.user
+                )
+                req.tenant = get_current_tenant()
+                res = view(req)
+                self.assertEqual(res.status_code, 200)
+
+    def test_search(self: TestModelViewSets) -> None:
+        """Test that search fields are correct"""
+        view = test_viewset.as_view({"get": "list"})
+        req = self.factory.get(
+            f"/?{urlencode({'search': generate_id()}, doseq=True)}", user=self.user
+        )
+        req.tenant = get_current_tenant()
+        res = view(req)
+        self.assertEqual(res.status_code, 200)
+
+    cases = {
+        "attrs": test_attrs,
+    }
+    if full:
+        cases["ordering"] = test_ordering
+        cases["search"] = test_search
+    return cases


 for _, viewset, _ in router.registry:
    if not issubclass(viewset, ModelViewSet | ReadOnlyModelViewSet):
        continue
-    setattr(TestModelViewSets, f"test_viewset_{viewset.__name__}", viewset_tester_factory(viewset))
+    full = viewset not in [VersionHistoryViewSet, DomainViewSet, TenantViewSet]
+    for test, case in viewset_tester_factory(viewset, full=full).items():
+        setattr(TestModelViewSets, f"test_viewset_{viewset.__name__}_{test}", case)
--- a/authentik/blueprints/management/commands/apply_blueprint.py
+++ b/authentik/blueprints/management/commands/apply_blueprint.py
@@ -1,5 +1,6 @@
 """Apply blueprint from commandline"""

+from argparse import ArgumentParser
 from sys import exit as sys_exit

 from django.core.management.base import BaseCommand, no_translations
@@ -31,5 +32,5 @@ class Command(BaseCommand):
                        sys_exit(1)
                    importer.apply()

-    def add_arguments(self, parser):
+    def add_arguments(self, parser: ArgumentParser):
        parser.add_argument("blueprints", nargs="+", type=str)
--- a/authentik/brands/tests.py
+++ b/authentik/brands/tests.py
@@ -20,11 +20,16 @@ class TestBrands(APITestCase):

    def setUp(self):
        super().setUp()
-        self.default_flags = {}
-        for flag in Flag.available(visibility="public"):
-            self.default_flags[flag().key] = flag.get()
        Brand.objects.all().delete()

+    @property
+    def default_flags(self) -> dict[str, object]:
+        """Get current public flags.
+
+        Some tests define temporary Flag subclasses, so this can't be cached in setUp.
+        """
+        return {flag().key: flag.get() for flag in Flag.available(visibility="public")}
+
    def test_current_brand(self):
        """Test Current brand API"""
        brand = create_test_brand()
--- a/authentik/common/oauth/constants.py
+++ b/authentik/common/oauth/constants.py
@@ -5,6 +5,7 @@ from django.utils.translation import gettext_lazy as _

 GRANT_TYPE_AUTHORIZATION_CODE = "authorization_code"
 GRANT_TYPE_IMPLICIT = "implicit"
+GRANT_TYPE_HYBRID = "hybrid"
 GRANT_TYPE_REFRESH_TOKEN = "refresh_token"  # nosec
 GRANT_TYPE_CLIENT_CREDENTIALS = "client_credentials"
 GRANT_TYPE_PASSWORD = "password"  # nosec
--- a/authentik/common/saml/constants.py
+++ b/authentik/common/saml/constants.py
@@ -30,6 +30,8 @@ SAML_BINDING_REDIRECT = "urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect"

 SAML_STATUS_SUCCESS = "urn:oasis:names:tc:SAML:2.0:status:Success"

+DEFAULT_ISSUER = "authentik"
+
 DSA_SHA1 = "http://www.w3.org/2000/09/xmldsig#dsa-sha1"
 RSA_SHA1 = "http://www.w3.org/2000/09/xmldsig#rsa-sha1"
 # https://datatracker.ietf.org/doc/html/rfc4051#section-2.3.2
--- a/authentik/core/api/application_entitlements.py
+++ b/authentik/core/api/application_entitlements.py
@@ -6,7 +6,6 @@ from rest_framework.exceptions import ValidationError
 from rest_framework.viewsets import ModelViewSet

 from authentik.blueprints.v1.importer import SERIALIZER_CONTEXT_BLUEPRINT
-from authentik.core.api.object_attributes import AttributesMixinSerializer
 from authentik.core.api.used_by import UsedByMixin
 from authentik.core.api.utils import ModelSerializer
 from authentik.core.models import (
@@ -15,7 +14,7 @@ from authentik.core.models import (
 )


-class ApplicationEntitlementSerializer(AttributesMixinSerializer, ModelSerializer):
+class ApplicationEntitlementSerializer(ModelSerializer):
    """ApplicationEntitlement Serializer"""

    def validate_app(self, app: Application) -> Application:
@@ -48,7 +47,8 @@ class ApplicationEntitlementViewSet(UsedByMixin, ModelViewSet):
    search_fields = [
        "pbm_uuid",
        "name",
-        "app",
+        "app__name",
+        "app__slug",
        "attributes",
    ]
    filterset_fields = [
--- a/authentik/core/api/applications.py
+++ b/authentik/core/api/applications.py
@@ -36,9 +36,13 @@ from authentik.rbac.filters import ObjectFilter
 LOGGER = get_logger()


-def user_app_cache_key(user_pk: str, page_number: int | None = None) -> str:
+def user_app_cache_key(
+    user_pk: str, page_number: int | None = None, only_with_launch_url: bool = False
+) -> str:
    """Cache key where application list for user is saved"""
    key = f"{CACHE_PREFIX}app_access/{user_pk}"
+    if only_with_launch_url:
+        key += "/launch"
    if page_number:
        key += f"/{page_number}"
    return key
@@ -116,6 +120,7 @@ class ApplicationSerializer(ModelSerializer):
            "meta_publisher",
            "policy_engine_mode",
            "group",
+            "meta_hide",
        ]
        extra_kwargs = {
            "backchannel_providers": {"required": False},
@@ -274,11 +279,17 @@ class ApplicationViewSet(UsedByMixin, ModelViewSet):
        if superuser_full_list and request.user.is_superuser:
            return super().list(request)

-        only_with_launch_url = str(
-            request.query_params.get("only_with_launch_url", "false")
-        ).lower()
+        only_with_launch_url = (
+            str(request.query_params.get("only_with_launch_url", "false")).lower()
+        ) == "true"

        queryset = self._filter_queryset_for_list(self.get_queryset())
+        queryset = queryset.exclude(meta_hide=True)
+        if only_with_launch_url:
+            # Pre-filter at DB level to skip expensive per-app policy evaluation
+            # for apps that can never appear in the launcher (no meta_launch_url
+            # and no provider, so no possible launch URL).
+            queryset = queryset.exclude(meta_launch_url="", provider__isnull=True)
        paginator: Pagination = self.paginator
        paginated_apps = paginator.paginate_queryset(queryset, request)

@@ -295,7 +306,6 @@ class ApplicationViewSet(UsedByMixin, ModelViewSet):
            except ValueError as exc:
                raise ValidationError from exc
            allowed_applications = self._get_allowed_applications(paginated_apps, user=for_user)
-            allowed_applications = self._expand_applications(allowed_applications)

            serializer = self.get_serializer(allowed_applications, many=True)
            return self.get_paginated_response(serializer.data)
@@ -305,19 +315,26 @@ class ApplicationViewSet(UsedByMixin, ModelViewSet):
            allowed_applications = self._get_allowed_applications(paginated_apps)
        if should_cache:
            allowed_applications = cache.get(
-                user_app_cache_key(self.request.user.pk, paginator.page.number)
+                user_app_cache_key(
+                    self.request.user.pk, paginator.page.number, only_with_launch_url
+                )
            )
-            if not allowed_applications:
+            if allowed_applications:
+                # Re-fetch cached applications since pickled instances lose prefetched
+                # relationships, causing N+1 queries during serialization
+                allowed_applications = self._expand_applications(allowed_applications)
+            else:
                LOGGER.debug("Caching allowed application list", page=paginator.page.number)
                allowed_applications = self._get_allowed_applications(paginated_apps)
                cache.set(
-                    user_app_cache_key(self.request.user.pk, paginator.page.number),
+                    user_app_cache_key(
+                        self.request.user.pk, paginator.page.number, only_with_launch_url
+                    ),
                    allowed_applications,
                    timeout=86400,
                )
-        allowed_applications = self._expand_applications(allowed_applications)

-        if only_with_launch_url == "true":
+        if only_with_launch_url:
            allowed_applications = self._filter_applications_with_launch_url(allowed_applications)

        serializer = self.get_serializer(allowed_applications, many=True)
--- a/authentik/core/api/groups.py
+++ b/authentik/core/api/groups.py
@@ -19,7 +19,7 @@ from rest_framework.authentication import SessionAuthentication
 from rest_framework.decorators import action
 from rest_framework.fields import CharField, IntegerField, SerializerMethodField
 from rest_framework.permissions import IsAuthenticated
-from rest_framework.relations import PrimaryKeyRelatedField
+from rest_framework.relations import ManyRelatedField, PrimaryKeyRelatedField
 from rest_framework.request import Request
 from rest_framework.response import Response
 from rest_framework.serializers import ListSerializer, ValidationError
@@ -30,7 +30,6 @@ from authentik.api.search.fields import (
    JSONSearchField,
 )
 from authentik.api.validation import validate
-from authentik.core.api.object_attributes import AttributesMixinSerializer
 from authentik.core.api.used_by import UsedByMixin
 from authentik.core.api.utils import JSONDictField, ModelSerializer, PassiveSerializer
 from authentik.core.models import Group, User
@@ -38,6 +37,77 @@ from authentik.endpoints.connectors.agent.auth import AgentAuth
 from authentik.rbac.api.roles import RoleSerializer
 from authentik.rbac.decorators import permission_required

+
+class BulkManyRelatedField(ManyRelatedField):
+    """ManyRelatedField that validates all PKs in a single query instead of one per PK."""
+
+    def to_internal_value(self, data):
+        if isinstance(data, str) or not hasattr(data, "__iter__"):
+            self.fail("not_a_list", input_type=type(data).__name__)
+        if not self.allow_empty and len(data) == 0:
+            self.fail("empty")
+
+        child = self.child_relation
+        pk_field = child.pk_field
+        # Coerce PKs through pk_field if defined
+        pk_map = {}
+        for item in data:
+            if isinstance(item, bool):
+                self.fail("incorrect_type", data_type=type(item).__name__)
+            pk = pk_field.to_internal_value(item) if pk_field else item
+            pk_map[pk] = item  # map coerced PK -> original value for error reporting
+
+        queryset = child.get_queryset()
+        # Use count to validate all PKs exist in a single query
+        found_count = queryset.filter(pk__in=pk_map.keys()).count()
+        if found_count < len(pk_map):
+            # Some PKs not found — fall back to per-PK checks for error reporting.
+            # This only runs when there's an actual validation error (rare path).
+            for pk, original in pk_map.items():
+                if not queryset.filter(pk=pk).exists():
+                    child.fail("does_not_exist", pk_value=original)
+
+        # Return raw PKs — Django's M2M set() accepts both objects and PKs,
+        # using get_prep_value() for type coercion. This avoids loading all
+        # objects into memory and avoids triggering post_init signals.
+        return list(pk_map.keys())
+
+    def to_representation(self, iterable):
+        # For non-prefetched querysets, get PKs directly without loading model instances.
+        # When prefetched, _result_cache is a list (possibly empty); when not, it's None.
+        if hasattr(iterable, "values_list") and getattr(iterable, "_result_cache", None) is None:
+            return list(iterable.values_list("pk", flat=True))
+        return super().to_representation(iterable)
+
+
+class BulkPrimaryKeyRelatedField(PrimaryKeyRelatedField):
+    """PrimaryKeyRelatedField that uses bulk validation when many=True."""
+
+    @classmethod
+    def many_init(cls, *args, **kwargs):
+        allow_empty = kwargs.pop("allow_empty", None)
+        max_length = kwargs.pop("max_length", None)
+        min_length = kwargs.pop("min_length", None)
+        child_relation = cls(*args, **kwargs)
+        list_kwargs = {
+            "child_relation": child_relation,
+        }
+        if allow_empty is not None:
+            list_kwargs["allow_empty"] = allow_empty
+        if max_length is not None:
+            list_kwargs["max_length"] = max_length
+        if min_length is not None:
+            list_kwargs["min_length"] = min_length
+        list_kwargs.update(
+            {
+                key: value
+                for key, value in kwargs.items()
+                if key in ("required", "default", "source")
+            }
+        )
+        return BulkManyRelatedField(**list_kwargs)
+
+
 PARTIAL_USER_SERIALIZER_MODEL_FIELDS = [
    "pk",
    "username",
@@ -76,10 +146,11 @@ class RelatedGroupSerializer(ModelSerializer):
        ]


-class GroupSerializer(AttributesMixinSerializer, ModelSerializer):
+class GroupSerializer(ModelSerializer):
    """Group Serializer"""

    attributes = JSONDictField(required=False)
+    users = BulkPrimaryKeyRelatedField(queryset=User.objects.all(), many=True, default=list)
    parents = PrimaryKeyRelatedField(queryset=Group.objects.all(), many=True, required=False)
    parents_obj = SerializerMethodField(allow_null=True)
    children_obj = SerializerMethodField(allow_null=True)
@@ -194,9 +265,6 @@ class GroupSerializer(AttributesMixinSerializer, ModelSerializer):
            "children_obj",
        ]
        extra_kwargs = {
-            "users": {
-                "default": list,
-            },
            "children": {
                "required": False,
                "default": list,
@@ -226,6 +294,7 @@ class GroupFilter(FilterSet):
    members_by_pk = ModelMultipleChoiceFilter(
        field_name="users",
        queryset=User.objects.all(),
+        distinct=False,
    )

    def filter_attributes(self, queryset, name, value):
@@ -277,7 +346,8 @@ class GroupViewSet(UsedByMixin, ModelViewSet):
        ]

    def get_queryset(self):
-        base_qs = Group.objects.all().prefetch_related("roles")
+        # Always prefetch parents and children since their PKs are always serialized
+        base_qs = Group.objects.all().prefetch_related("roles", "parents", "children")

        if self.serializer_class(context={"request": self.request})._should_include_users:
            # Only fetch fields needed by PartialUserSerializer to reduce DB load and instantiation
@@ -288,16 +358,9 @@ class GroupViewSet(UsedByMixin, ModelViewSet):
                    queryset=User.objects.all().only(*PARTIAL_USER_SERIALIZER_MODEL_FIELDS),
                )
            )
-        else:
-            base_qs = base_qs.prefetch_related(
-                Prefetch("users", queryset=User.objects.all().only("id"))
-            )
-
-        if self.serializer_class(context={"request": self.request})._should_include_children:
-            base_qs = base_qs.prefetch_related("children")
-
-        if self.serializer_class(context={"request": self.request})._should_include_parents:
-            base_qs = base_qs.prefetch_related("parents")
+        # When include_users=false, skip users prefetch entirely.
+        # BulkManyRelatedField.to_representation will use values_list to get PKs
+        # directly without loading User instances into memory.

        return base_qs

--- a/authentik/core/api/object_attributes.py
+++ b/authentik/core/api/object_attributes.py
@@ -1,94 +0,0 @@
-from typing import Any
-
-from django.contrib.contenttypes.models import ContentType
-from django.utils.translation import gettext_lazy as _
-from rest_framework.exceptions import ValidationError
-from rest_framework.fields import CharField, SerializerMethodField
-from rest_framework.viewsets import ModelViewSet
-
-from authentik.core.api.utils import ModelSerializer
-from authentik.core.models import AttributesMixin, ObjectAttribute
-from authentik.lib.utils.dict import get_path_from_dict
-
-
-class AttributesMixinSerializer(ModelSerializer):
-
-    def validate(self, data: dict[str, Any]) -> dict[str, Any]:
-        model = self.Meta.model
-        attrs = data.get("attributes", {})
-        attributes = ObjectAttribute.objects.filter(
-            object_type=ContentType.objects.get_for_model(model),
-            enabled=True,
-        )
-        for attr in attributes:
-            value = get_path_from_dict(attrs, attr.key)
-            attr.run_validation(value)
-        return data
-
-
-class ContentTypeSerializer(ModelSerializer):
-    app_label = CharField(read_only=True)
-    model = CharField(read_only=True)
-    verbose_name_plural = SerializerMethodField()
-    fully_qualified_model = SerializerMethodField()
-
-    def get_fully_qualified_model(self, ct: ContentType) -> str:
-        return f"{ct.app_label}.{ct.model}"
-
-    def get_verbose_name_plural(self, ct: ContentType) -> str:
-        return ct.model_class()._meta.verbose_name_plural
-
-    class Meta:
-        model = ContentType
-        fields = ("id", "app_label", "model", "verbose_name_plural", "fully_qualified_model")
-
-
-class ObjectAttributeSerializer(ModelSerializer):
-
-    object_type = CharField()
-    object_type_obj = ContentTypeSerializer(read_only=True, source="object_type")
-
-    def validate_object_type(self, fqm: str) -> ContentType:
-        app_label, _, model = fqm.partition(".")
-        ct = ContentType.objects.filter(app_label=app_label, model=model).first()
-        if not ct or not issubclass(ct.model_class(), AttributesMixin):
-            raise ValidationError("Invalid object type")
-        return ct
-
-    def validate(self, attrs: dict[str, Any]) -> dict[str, Any]:
-        if attrs.get("is_unique") and attrs.get("is_array"):
-            raise ValidationError(_("Unique cannot be enabled for arrays."))
-        return super().validate(attrs)
-
-    class Meta:
-        model = ObjectAttribute
-        fields = [
-            "pk",
-            "object_type",
-            "object_type_obj",
-            "enabled",
-            "created",
-            "key",
-            "label",
-            "last_updated",
-            "regex",
-            "type",
-            "group",
-            "managed",
-            "is_unique",
-            "is_required",
-            "is_array",
-        ]
-        extra_kwargs = {
-            "last_updated": {"read_only": True},
-            "created": {"read_only": True},
-            "pk": {"read_only": True},
-        }
-
-
-class ObjectAttributeViewSet(ModelViewSet):
-    serializer_class = ObjectAttributeSerializer
-    queryset = ObjectAttribute.objects.all()
-    filterset_fields = ["object_type__model", "object_type__app_label", "enabled"]
-    search_fields = ["key", "label", "group", "object_type__model", "object_type__app_label"]
-    ordering = ["key"]
--- a/authentik/core/api/users.py
+++ b/authentik/core/api/users.py
@@ -6,6 +6,7 @@ from typing import Any

 from django.contrib.auth import update_session_auth_hash
 from django.contrib.auth.models import AnonymousUser, Permission
+from django.db.models import Exists, OuterRef, Prefetch, Q
 from django.db.transaction import atomic
 from django.db.utils import IntegrityError
 from django.urls import reverse_lazy
@@ -13,6 +14,7 @@ from django.utils.http import urlencode
 from django.utils.text import slugify
 from django.utils.timezone import now
 from django.utils.translation import gettext as _
+from django.utils.translation import gettext_lazy
 from django_filters.filters import (
    BooleanFilter,
    CharFilter,
@@ -63,7 +65,6 @@ from authentik.api.search.fields import (
 from authentik.api.validation import validate
 from authentik.blueprints.v1.importer import SERIALIZER_CONTEXT_BLUEPRINT
 from authentik.brands.models import Brand
-from authentik.core.api.object_attributes import AttributesMixinSerializer
 from authentik.core.api.used_by import UsedByMixin
 from authentik.core.api.utils import (
    JSONDictField,
@@ -106,6 +107,10 @@ from authentik.stages.email.utils import TemplateEmailMessage

 LOGGER = get_logger()

+INVALID_PASSWORD_HASH_MESSAGE = gettext_lazy(
+    "Invalid password hash format. Must be a valid Django password hash."
+)
+

 class ParamUserSerializer(PassiveSerializer):
    """Partial serializer for query parameters to select a user"""
@@ -129,10 +134,10 @@ class PartialGroupSerializer(ModelSerializer):
        ]


-class UserSerializer(AttributesMixinSerializer, ModelSerializer):
+class UserSerializer(ModelSerializer):
    """User Serializer"""

-    is_superuser = BooleanField(read_only=True)
+    is_superuser = SerializerMethodField()
    avatar = SerializerMethodField()
    attributes = JSONDictField(required=False)
    groups = PrimaryKeyRelatedField(
@@ -169,6 +174,14 @@ class UserSerializer(AttributesMixinSerializer, ModelSerializer):
            return True
        return str(request.query_params.get("include_roles", "true")).lower() == "true"

+    @extend_schema_field(BooleanField)
+    def get_is_superuser(self, instance: User) -> bool:
+        """Use annotation if available to avoid N+1 query"""
+        ann = getattr(instance, "_annotated_is_superuser", None)
+        if ann is not None:
+            return ann
+        return instance.is_superuser
+
    @extend_schema_field(PartialGroupSerializer(many=True))
    def get_groups_obj(self, instance: User) -> list[PartialGroupSerializer] | None:
        if not self._should_include_groups:
@@ -182,47 +195,79 @@ class UserSerializer(AttributesMixinSerializer, ModelSerializer):
        return RoleSerializer(instance.roles, many=True).data

    def __init__(self, *args, **kwargs):
+        """Setting password and permissions directly is allowed only in blueprints."""
        super().__init__(*args, **kwargs)
        if SERIALIZER_CONTEXT_BLUEPRINT in self.context:
            self.fields["password"] = CharField(required=False, allow_null=True)
+            self.fields["password_hash"] = CharField(required=False, allow_null=True)
            self.fields["permissions"] = ListField(
                required=False,
                child=ChoiceField(choices=get_permission_choices()),
            )

    def create(self, validated_data: dict) -> User:
-        """If this serializer is used in the blueprint context, we allow for
-        directly setting a password. However should be done via the `set_password`
-        method instead of directly setting it like rest_framework."""
-        password = validated_data.pop("password", None)
-        perms_qs = Permission.objects.filter(
-            codename__in=[x.split(".")[1] for x in validated_data.pop("permissions", [])]
-        ).values_list("content_type__app_label", "codename")
-        perms_list = [f"{ct}.{name}" for ct, name in list(perms_qs)]
+        """Create a user, with blueprint-only password and permission writes."""
+        is_blueprint = SERIALIZER_CONTEXT_BLUEPRINT in self.context
+        if is_blueprint:
+            password = validated_data.pop("password", None)
+            password_hash = validated_data.pop("password_hash", None)
+            permissions = validated_data.pop("permissions", [])
+            self._validate_password_inputs(password, password_hash)
+
        instance: User = super().create(validated_data)
-        self._set_password(instance, password)
-        instance.assign_perms_to_managed_role(perms_list)
+        if is_blueprint:
+            self._set_password(instance, password, password_hash)
+            perms_qs = Permission.objects.filter(
+                codename__in=[permission.split(".")[1] for permission in permissions]
+            ).values_list("content_type__app_label", "codename")
+            perms_list = [f"{ct}.{name}" for ct, name in perms_qs]
+            instance.assign_perms_to_managed_role(perms_list)
+        self._ensure_password_not_empty(instance)
        return instance

    def update(self, instance: User, validated_data: dict) -> User:
-        """Same as `create` above, set the password directly if we're in a blueprint
-        context"""
-        password = validated_data.pop("password", None)
-        perms_qs = Permission.objects.filter(
-            codename__in=[x.split(".")[1] for x in validated_data.pop("permissions", [])]
-        ).values_list("content_type__app_label", "codename")
-        perms_list = [f"{ct}.{name}" for ct, name in list(perms_qs)]
+        """Update a user, with blueprint-only password and permission writes."""
+        is_blueprint = SERIALIZER_CONTEXT_BLUEPRINT in self.context
+        if is_blueprint:
+            password = validated_data.pop("password", None)
+            password_hash = validated_data.pop("password_hash", None)
+            permissions = validated_data.pop("permissions", [])
+            self._validate_password_inputs(password, password_hash)
+
        instance = super().update(instance, validated_data)
-        self._set_password(instance, password)
-        instance.assign_perms_to_managed_role(perms_list)
+        if is_blueprint:
+            self._set_password(instance, password, password_hash)
+            perms_qs = Permission.objects.filter(
+                codename__in=[permission.split(".")[1] for permission in permissions]
+            ).values_list("content_type__app_label", "codename")
+            perms_list = [f"{ct}.{name}" for ct, name in perms_qs]
+            instance.assign_perms_to_managed_role(perms_list)
+        self._ensure_password_not_empty(instance)
        return instance

-    def _set_password(self, instance: User, password: str | None):
-        """Set password of user if we're in a blueprint context, and if it's an empty
-        string then use an unusable password"""
-        if SERIALIZER_CONTEXT_BLUEPRINT in self.context and password:
+    def _validate_password_inputs(self, password: str | None, password_hash: str | None):
+        """Validate mutually-exclusive password inputs before any model mutation."""
+        if password is not None and password_hash is not None:
+            raise ValidationError(_("Cannot set both password and password_hash. Use only one."))
+        if password_hash is None:
+            return
+        try:
+            User.validate_password_hash(password_hash)
+        except ValueError as exc:
+            LOGGER.warning("Failed to identify password hash format", exc_info=exc)
+            raise ValidationError(INVALID_PASSWORD_HASH_MESSAGE) from exc
+
+    def _set_password(self, instance: User, password: str | None, password_hash: str | None = None):
+        """Set password from plain text or hash."""
+        if password_hash is not None:
+            instance.set_password_from_hash(password_hash)
+            instance.save()
+        elif password:
            instance.set_password(password)
            instance.save()
+
+    def _ensure_password_not_empty(self, instance: User):
+        """Store an explicit unusable password instead of an empty password field."""
        if len(instance.password) == 0:
            instance.set_unusable_password()
            instance.save()
@@ -391,6 +436,12 @@ class UserPasswordSetSerializer(PassiveSerializer):
    password = CharField(required=True)


+class UserPasswordHashSetSerializer(PassiveSerializer):
+    """Payload to set a users' password hash directly"""
+
+    password = CharField(required=True)
+
+
 class UserServiceAccountSerializer(PassiveSerializer):
    """Payload to create a service account"""

@@ -542,10 +593,30 @@ class UserViewSet(

    def get_queryset(self):
        base_qs = User.objects.all().exclude_anonymous()
+        # Always prefetch groups since group PKs are always serialized.
+        # Use full prefetch when include_groups=true (for groups_obj), ID-only otherwise.
        if self.serializer_class(context={"request": self.request})._should_include_groups:
            base_qs = base_qs.prefetch_related("groups")
+        else:
+            base_qs = base_qs.prefetch_related(
+                Prefetch("groups", queryset=Group.objects.all().only("group_uuid"))
+            )
        if self.serializer_class(context={"request": self.request})._should_include_roles:
            base_qs = base_qs.prefetch_related("roles")
+        else:
+            base_qs = base_qs.prefetch_related(
+                Prefetch("roles", queryset=Role.objects.all().only("uuid"))
+            )
+        # Annotate is_superuser to avoid N+1 query per user
+        base_qs = base_qs.annotate(
+            _annotated_is_superuser=Exists(
+                Group.objects.filter(
+                    is_superuser=True,
+                ).filter(
+                    Q(users=OuterRef("pk")) | Q(descendant_nodes__descendant__users=OuterRef("pk"))
+                )
+            )
+        )
        return base_qs

    @extend_schema(
@@ -714,6 +785,11 @@ class UserViewSet(
        self.request.session.modified = True
        return Response(serializer.initial_data)

+    def _update_session_hash_after_password_change(self, request: Request, user: User):
+        if user.pk == request.user.pk and SESSION_KEY_IMPERSONATE_USER not in self.request.session:
+            LOGGER.debug("Updating session hash after password change")
+            update_session_auth_hash(self.request, user)
+
    @permission_required("authentik_core.reset_user_password")
    @extend_schema(
        request=UserPasswordSetSerializer,
@@ -737,9 +813,45 @@ class UserViewSet(
        except (ValidationError, IntegrityError) as exc:
            LOGGER.debug("Failed to set password", exc=exc)
            return Response(status=400)
-        if user.pk == request.user.pk and SESSION_KEY_IMPERSONATE_USER not in self.request.session:
-            LOGGER.debug("Updating session hash after password change")
-            update_session_auth_hash(self.request, user)
+        self._update_session_hash_after_password_change(request, user)
+        return Response(status=204)
+
+    @permission_required("authentik_core.reset_user_password")
+    @extend_schema(
+        request=UserPasswordHashSetSerializer,
+        responses={
+            204: OpenApiResponse(description="Successfully changed password"),
+            400: OpenApiResponse(description="Bad request"),
+        },
+    )
+    @action(
+        detail=True,
+        methods=["POST"],
+        permission_classes=[IsAuthenticated],
+    )
+    @validate(UserPasswordHashSetSerializer)
+    def set_password_hash(
+        self, request: Request, pk: int, body: UserPasswordHashSetSerializer
+    ) -> Response:
+        """Set a user's password from a pre-hashed Django password value.
+
+        Submit the Django password hash in the shared ``password`` request field.
+
+        This updates authentik's local password verifier only. It does not attempt
+        to propagate the password change to LDAP or Kerberos because no raw password
+        is available from the request payload.
+        """
+        user: User = self.get_object()
+        try:
+            user.set_password_from_hash(body.validated_data["password"], request=request)
+            user.save()
+        except ValueError as exc:
+            LOGGER.debug("Failed to set password hash", exc=exc)
+            return Response(data={"password": [INVALID_PASSWORD_HASH_MESSAGE]}, status=400)
+        except (ValidationError, IntegrityError) as exc:
+            LOGGER.debug("Failed to set password hash", exc=exc)
+            return Response(status=400)
+        self._update_session_hash_after_password_change(request, user)
        return Response(status=204)

    @permission_required("authentik_core.reset_user_password")
--- a/authentik/core/apps.py
+++ b/authentik/core/apps.py
@@ -7,6 +7,12 @@ from authentik.tasks.schedules.common import ScheduleSpec
 from authentik.tenants.flags import Flag


+class Setup(Flag[bool], key="setup"):
+
+    default = False
+    visibility = "system"
+
+
 class AppAccessWithoutBindings(Flag[bool], key="core_default_app_access"):

    default = True
@@ -26,6 +32,10 @@ class AuthentikCoreConfig(ManagedAppConfig):
    mountpoint = ""
    default = True

+    def import_related(self):
+        super().import_related()
+        self.import_module("authentik.core.setup.signals")
+
    @ManagedAppConfig.reconcile_tenant
    def source_inbuilt(self):
        """Reconcile inbuilt source"""
--- a/authentik/core/management/commands/hash_password.py
+++ b/authentik/core/management/commands/hash_password.py
@@ -0,0 +1,28 @@
+"""Hash password using Django's password hashers"""
+
+from django.contrib.auth.hashers import make_password
+from django.core.management.base import BaseCommand, CommandError
+
+
+class Command(BaseCommand):
+    """Hash a password using Django's password hashers"""
+
+    help = "Hash a password for use with AUTHENTIK_BOOTSTRAP_PASSWORD_HASH"
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "password",
+            type=str,
+            help="Password to hash",
+        )
+
+    def handle(self, *args, **options):
+        password = options["password"]
+
+        if not password:
+            raise CommandError("Password cannot be empty")
+        try:
+            hashed = make_password(password)
+            self.stdout.write(hashed)
+        except ValueError as exc:
+            raise CommandError(f"Error hashing password: {exc}") from exc
--- a/authentik/core/migrations/0058_objectattribute.py
+++ b/authentik/core/migrations/0058_objectattribute.py
@@ -1,62 +0,0 @@
-# Generated by Django 5.2.13 on 2026-04-11 18:35
-
-import django.db.models.deletion
-import uuid
-from django.db import migrations, models
-
-
-class Migration(migrations.Migration):
-
-    dependencies = [
-        ("authentik_core", "0057_remove_user_groups_remove_user_user_permissions_and_more"),
-        ("contenttypes", "0002_remove_content_type_name"),
-    ]
-
-    operations = [
-        migrations.CreateModel(
-            name="ObjectAttribute",
-            fields=[
-                ("created", models.DateTimeField(auto_now_add=True)),
-                ("last_updated", models.DateTimeField(auto_now=True)),
-                (
-                    "managed",
-                    models.TextField(
-                        default=None,
-                        help_text="Objects that are managed by authentik. These objects are created and updated automatically. This flag only indicates that an object can be overwritten by migrations. You can still modify the objects via the API, but expect changes to be overwritten in a later update.",
-                        null=True,
-                        unique=True,
-                        verbose_name="Managed by authentik",
-                    ),
-                ),
-                (
-                    "attribute_id",
-                    models.UUIDField(default=uuid.uuid4, primary_key=True, serialize=False),
-                ),
-                ("enabled", models.BooleanField(default=True)),
-                ("label", models.TextField()),
-                ("group", models.TextField(blank=True)),
-                ("key", models.TextField()),
-                (
-                    "type",
-                    models.TextField(
-                        choices=[("text", "Text"), ("number", "Number"), ("boolean", "Boolean")]
-                    ),
-                ),
-                ("is_unique", models.BooleanField(default=False)),
-                ("is_required", models.BooleanField(default=False)),
-                ("regex", models.TextField(blank=True)),
-                ("is_array", models.BooleanField(default=False)),
-                (
-                    "object_type",
-                    models.ForeignKey(
-                        on_delete=django.db.models.deletion.CASCADE, to="contenttypes.contenttype"
-                    ),
-                ),
-            ],
-            options={
-                "verbose_name": "Object Attribute",
-                "verbose_name_plural": "Object Attributes",
-                "unique_together": {("object_type", "key", "enabled")},
-            },
-        ),
-    ]
--- a/authentik/core/migrations/0058_setup.py
+++ b/authentik/core/migrations/0058_setup.py
@@ -0,0 +1,61 @@
+# Generated by Django 5.2.13 on 2026-04-21 18:49
+from django.apps.registry import Apps
+
+from django.db.backends.base.schema import BaseDatabaseSchemaEditor
+
+from django.db import migrations
+
+
+def check_is_already_setup(apps: Apps, schema_editor: BaseDatabaseSchemaEditor):
+    from django.conf import settings
+    from authentik.flows.models import FlowAuthenticationRequirement
+
+    VersionHistory = apps.get_model("authentik_admin", "VersionHistory")
+    Flow = apps.get_model("authentik_flows", "Flow")
+    User = apps.get_model("authentik_core", "User")
+
+    db_alias = schema_editor.connection.alias
+
+    # Upgrading from a previous version
+    if not settings.TEST and VersionHistory.objects.using(db_alias).count() > 1:
+        return True
+    # OOBE flow sets itself to this authentication requirement once finished
+    if (
+        Flow.objects.using(db_alias)
+        .filter(
+            slug="initial-setup", authentication=FlowAuthenticationRequirement.REQUIRE_SUPERUSER
+        )
+        .exists()
+    ):
+        return True
+    # non-akadmin and non-guardian anonymous user exist
+    if (
+        User.objects.using(db_alias)
+        .exclude(username="akadmin")
+        .exclude(username="AnonymousUser")
+        .exists()
+    ):
+        return True
+    return False
+
+
+def update_setup_flag(apps: Apps, schema_editor: BaseDatabaseSchemaEditor):
+    from authentik.core.apps import Setup
+    from authentik.tenants.utils import get_current_tenant
+
+    is_already_setup = check_is_already_setup(apps, schema_editor)
+    if is_already_setup:
+        tenant = get_current_tenant()
+        tenant.flags[Setup().key] = True
+        tenant.save()
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("authentik_core", "0057_remove_user_groups_remove_user_user_permissions_and_more"),
+        # 0024_flow_authentication adds the `authentication` field.
+        ("authentik_flows", "0024_flow_authentication"),
+    ]
+
+    operations = [migrations.RunPython(update_setup_flag, migrations.RunPython.noop)]
--- a/authentik/core/migrations/0059_add_application_meta_hide.py
+++ b/authentik/core/migrations/0059_add_application_meta_hide.py
@@ -0,0 +1,33 @@
+# Generated by Django 5.2.12 on 2026-04-09 18:04
+
+from django.apps.registry import Apps
+from django.db import migrations, models
+from django.db.backends.base.schema import BaseDatabaseSchemaEditor
+
+
+def migrate_blank_launch_url(apps: Apps, schema_editor: BaseDatabaseSchemaEditor):
+    db_alias = schema_editor.connection.alias
+    Application = apps.get_model("authentik_core", "Application")
+
+    Application.objects.using(db_alias).filter(meta_launch_url="blank://blank").update(
+        meta_hide=True, meta_launch_url=""
+    )
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("authentik_core", "0058_setup"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="application",
+            name="meta_hide",
+            field=models.BooleanField(
+                default=False,
+                help_text="Hide this application from the user's My applications page.",
+            ),
+        ),
+        migrations.RunPython(migrate_blank_launch_url, migrations.RunPython.noop),
+    ]
--- a/authentik/core/models.py
+++ b/authentik/core/models.py
@@ -10,10 +10,9 @@ from uuid import uuid4

 import pgtrigger
 from deepmerge import always_merger
-from django.contrib.auth.hashers import check_password
+from django.contrib.auth.hashers import check_password, identify_hasher
 from django.contrib.auth.models import AbstractUser, Permission
 from django.contrib.auth.models import UserManager as DjangoUserManager
-from django.contrib.contenttypes.models import ContentType
 from django.contrib.sessions.base_session import AbstractBaseSession
 from django.core.validators import validate_slug
 from django.db import models
@@ -27,7 +26,6 @@ from guardian.models import RoleModelPermission, RoleObjectPermission
 from model_utils.managers import InheritanceManager
 from psqlextra.indexes import UniqueIndex
 from psqlextra.models import PostgresMaterializedViewModel
-from rest_framework.exceptions import ValidationError
 from rest_framework.serializers import Serializer
 from structlog.stdlib import get_logger

@@ -562,6 +560,33 @@ class User(SerializerModel, AttributesMixin, AbstractUser):
        self.password_change_date = now()
        return super().set_password(raw_password)

+    @staticmethod
+    def validate_password_hash(password_hash: str):
+        """Validate that the value is a recognized Django password hash."""
+        identify_hasher(password_hash)  # Raises ValueError if invalid
+
+    def set_password_from_hash(self, password_hash: str, signal=True, sender=None, request=None):
+        """Set password directly from a pre-hashed value.
+
+        Unlike set_password(), this does not hash the input again. The provided value
+        must already be a valid Django password hash, and it is stored directly on the
+        user after validation.
+
+        Because no raw password is available, downstream password sync integrations
+        such as LDAP and Kerberos cannot be updated from this code path.
+
+        Raises ValueError if the hash format is not recognized.
+        """
+        self.validate_password_hash(password_hash)
+        if self.pk and signal:
+            from authentik.core.signals import password_hash_changed
+
+            if not sender:
+                sender = self
+            password_hash_changed.send(sender=sender, user=self, request=request)
+        self.password = password_hash
+        self.password_change_date = now()
+
    def check_password(self, raw_password: str) -> bool:
        """
        Return a boolean of whether the raw_password was correct. Handles
@@ -737,6 +762,9 @@ class Application(SerializerModel, PolicyBindingModel):
    meta_icon = FileField(default="", blank=True)
    meta_description = models.TextField(default="", blank=True)
    meta_publisher = models.TextField(default="", blank=True)
+    meta_hide = models.BooleanField(
+        default=False, help_text=_("Hide this application from the user's My applications page.")
+    )

    objects = ApplicationQuerySet.as_manager()

@@ -792,9 +820,13 @@ class Application(SerializerModel, PolicyBindingModel):

    def get_provider(self) -> Provider | None:
        """Get casted provider instance. Needs Application queryset with_provider"""
+        if hasattr(self, "_cached_provider"):
+            return self._cached_provider
        if not self.provider:
+            self._cached_provider = None
            return None
-        return get_deepest_child(self.provider)
+        self._cached_provider = get_deepest_child(self.provider)
+        return self._cached_provider

    def backchannel_provider_for[T: Provider](self, provider_type: type[T], **kwargs) -> T | None:
        """Get Backchannel provider for a specific type"""
@@ -1360,61 +1392,3 @@ class AuthenticatedSession(SerializerModel):
            session=Session.objects.filter(session_key=request.session.session_key).first(),
            user=user,
        )
-
-
-class ObjectAttribute(SerializerModel, ManagedModel, CreatedUpdatedModel):
-    """User-defined schema for models' `attributes` JSON field."""
-
-    class AttributeType(models.TextChoices):
-        TEXT = "text"
-        NUMBER = "number"
-        BOOLEAN = "boolean"
-
-    attribute_id = models.UUIDField(default=uuid4, primary_key=True)
-
-    enabled = models.BooleanField(default=True)
-
-    object_type = models.ForeignKey(ContentType, on_delete=models.CASCADE)
-    label = models.TextField()
-    group = models.TextField(blank=True)
-    key = models.TextField()
-
-    type = models.TextField(choices=AttributeType.choices)
-    is_unique = models.BooleanField(default=False)
-    is_required = models.BooleanField(default=False)
-    regex = models.TextField(blank=True)
-    is_array = models.BooleanField(default=False)
-
-    def run_validation(self, value: Any) -> None:
-        err_key = f"attributes_{self.key.replace(".", "_")}"
-        if self.is_required and value is None:
-            raise ValidationError({err_key: _("This field is required")})
-        if self.is_array:
-            if not isinstance(value, (list, tuple)):
-                raise ValidationError({err_key: _("Value must be an array.")})
-            if self.regex != "":
-                if not all(re.fullmatch(self.regex, v) for v in value):
-                    raise ValidationError({err_key: _("Value does not match configured pattern.")})
-        else:
-            if self.is_unique:
-                model: type[models.Model] = self.object_type.model_class()
-                lookup_key = f"attributes__{self.key.replace(".", "__")}"
-                if model.objects.filter(**{lookup_key: value}).exists():
-                    raise ValidationError({err_key: _("Value is not unique.")})
-            if self.regex != "":
-                if not re.fullmatch(self.regex, value):
-                    raise ValidationError({err_key: _("Value does not match configured pattern.")})
-
-    @property
-    def serializer(self) -> type[Serializer]:
-        from authentik.core.api.object_attributes import ObjectAttributeSerializer
-
-        return ObjectAttributeSerializer
-
-    def __str__(self):
-        return f"Object attribute '{self.key}' for content type {self.object_type_id}"
-
-    class Meta:
-        verbose_name = _("Object Attribute")
-        verbose_name_plural = _("Object Attributes")
-        unique_together = (("object_type", "key", "enabled"),)
--- a/packages/django-dramatiq-postgres/django_dramatiq_postgres/management/init.py
+++ b/packages/django-dramatiq-postgres/django_dramatiq_postgres/management/init.py
--- a/authentik/core/setup/signals.py
+++ b/authentik/core/setup/signals.py
@@ -0,0 +1,42 @@
+from os import getenv
+
+from django.dispatch import receiver
+from structlog.stdlib import get_logger
+
+from authentik.blueprints.models import BlueprintInstance
+from authentik.blueprints.v1.importer import Importer
+from authentik.core.apps import Setup
+from authentik.root.signals import post_startup
+from authentik.tenants.models import Tenant
+
+BOOTSTRAP_BLUEPRINT = "system/bootstrap.yaml"
+
+LOGGER = get_logger()
+
+
+@receiver(post_startup)
+def post_startup_setup_bootstrap(sender, **_):
+    if (
+        not getenv("AUTHENTIK_BOOTSTRAP_PASSWORD")
+        and not getenv("AUTHENTIK_BOOTSTRAP_PASSWORD_HASH")
+        and not getenv("AUTHENTIK_BOOTSTRAP_TOKEN")
+    ):
+        return
+    LOGGER.info("Configuring authentik through bootstrap environment variables")
+    content = BlueprintInstance(path=BOOTSTRAP_BLUEPRINT).retrieve()
+    # If we have bootstrap credentials set, run bootstrap tasks outside of main server
+    # sync, so that we can sure the first start actually has working bootstrap
+    # credentials
+    for tenant in Tenant.objects.filter(ready=True):
+        if Setup.get(tenant=tenant):
+            LOGGER.info("Tenant is already setup, skipping", tenant=tenant.schema_name)
+            continue
+        with tenant:
+            importer = Importer.from_string(content)
+            valid, logs = importer.validate()
+            if not valid:
+                LOGGER.warning("Blueprint invalid", tenant=tenant.schema_name)
+                for log in logs:
+                    log.log()
+            importer.apply()
+            Setup.set(True, tenant=tenant)
--- a/authentik/core/setup/views.py
+++ b/authentik/core/setup/views.py
@@ -0,0 +1,80 @@
+from functools import lru_cache
+from http import HTTPMethod, HTTPStatus
+
+from django.contrib.staticfiles import finders
+from django.db import transaction
+from django.http import HttpRequest, HttpResponse
+from django.shortcuts import redirect
+from django.urls import reverse
+from django.views import View
+from structlog.stdlib import get_logger
+
+from authentik.blueprints.models import BlueprintInstance
+from authentik.core.apps import Setup
+from authentik.flows.models import Flow, FlowAuthenticationRequirement, in_memory_stage
+from authentik.flows.planner import FlowPlanner
+from authentik.flows.stage import StageView
+
+LOGGER = get_logger()
+FLOW_CONTEXT_START_BY = "goauthentik.io/core/setup/started-by"
+
+
+@lru_cache
+def read_static(path: str) -> str | None:
+    result = finders.find(path)
+    if not result:
+        return None
+    with open(result, encoding="utf8") as _file:
+        return _file.read()
+
+
+class SetupView(View):
+
+    setup_flow_slug = "initial-setup"
+
+    def dispatch(self, request: HttpRequest, *args, **kwargs):
+        if request.method != HTTPMethod.HEAD and Setup.get():
+            return redirect(reverse("authentik_core:root-redirect"))
+        return super().dispatch(request, *args, **kwargs)
+
+    def head(self, request: HttpRequest, *args, **kwargs):
+        if Setup.get():
+            return HttpResponse(status=HTTPStatus.SERVICE_UNAVAILABLE)
+        if not Flow.objects.filter(slug=self.setup_flow_slug).exists():
+            return HttpResponse(status=HTTPStatus.SERVICE_UNAVAILABLE)
+        return HttpResponse(status=HTTPStatus.OK)
+
+    def get(self, request: HttpRequest):
+        flow = Flow.objects.filter(slug=self.setup_flow_slug).first()
+        if not flow:
+            LOGGER.info("Setup flow does not exist yet, waiting for worker to finish")
+            return HttpResponse(
+                read_static("dist/standalone/loading/startup.html"),
+                status=HTTPStatus.SERVICE_UNAVAILABLE,
+            )
+        planner = FlowPlanner(flow)
+        plan = planner.plan(request, {FLOW_CONTEXT_START_BY: "setup"})
+        plan.append_stage(in_memory_stage(PostSetupStageView))
+        return plan.to_redirect(request, flow)
+
+
+class PostSetupStageView(StageView):
+    """Run post-setup tasks"""
+
+    def post(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
+        """Wrapper when this stage gets hit with a post request"""
+        return self.get(request, *args, **kwargs)
+
+    def get(self, requeset: HttpRequest, *args, **kwargs):
+        with transaction.atomic():
+            # Remember we're setup
+            Setup.set(True)
+            # Disable OOBE Blueprints
+            BlueprintInstance.objects.filter(
+                **{"metadata__labels__blueprints.goauthentik.io/system-oobe": "true"}
+            ).update(enabled=False)
+            # Make flow inaccessible
+            Flow.objects.filter(slug="initial-setup").update(
+                authentication=FlowAuthenticationRequirement.REQUIRE_SUPERUSER
+            )
+        return self.executor.stage_ok()
--- a/authentik/core/signals.py
+++ b/authentik/core/signals.py
@@ -24,6 +24,8 @@ from authentik.root.ws.consumer import build_device_group

 # Arguments: user: User, password: str
 password_changed = Signal()
+# Arguments: user: User, request: HttpRequest | None
+password_hash_changed = Signal()
 # Arguments: credentials: dict[str, any], request: HttpRequest,
 #            stage: Stage, context: dict[str, any]
 login_failed = Signal()
--- a/authentik/core/templates/login/base_full.html
+++ b/authentik/core/templates/login/base_full.html
@@ -12,7 +12,7 @@
 {% block head %}
 <style data-id="static-styles">
    :root {
-        --ak-global--background-image: url("{{ request.brand.branding_default_flow_background_url }}");
+        --ak-global--background-image: url("{{ request.brand.branding_default_flow_background_url|iriencode|safe }}");
    }
 </style>

--- a/authentik/core/tests/test_applications_api.py
+++ b/authentik/core/tests/test_applications_api.py
@@ -129,6 +129,7 @@ class TestApplicationsAPI(APITestCase):
                        "meta_icon_url": None,
                        "meta_icon_themed_urls": None,
                        "meta_description": "",
+                        "meta_hide": False,
                        "meta_publisher": "",
                        "policy_engine_mode": "any",
                    },
@@ -187,12 +188,14 @@ class TestApplicationsAPI(APITestCase):
                        "meta_icon_url": None,
                        "meta_icon_themed_urls": None,
                        "meta_description": "",
+                        "meta_hide": False,
                        "meta_publisher": "",
                        "policy_engine_mode": "any",
                    },
                    {
                        "launch_url": None,
                        "meta_description": "",
+                        "meta_hide": False,
                        "meta_icon": "",
                        "meta_icon_url": None,
                        "meta_icon_themed_urls": None,
--- a/authentik/core/tests/test_hash_password_command.py
+++ b/authentik/core/tests/test_hash_password_command.py
@@ -0,0 +1,28 @@
+"""Tests for hash_password management command."""
+
+from io import StringIO
+
+from django.contrib.auth.hashers import check_password
+from django.core.management import call_command
+from django.core.management.base import CommandError
+from django.test import TestCase
+
+
+class TestHashPasswordCommand(TestCase):
+    """Test hash_password management command."""
+
+    def test_hash_password(self):
+        """Test hashing a password."""
+        out = StringIO()
+        call_command("hash_password", "test123", stdout=out)
+        hashed = out.getvalue().strip()
+
+        self.assertTrue(hashed.startswith("pbkdf2_sha256$"))
+        self.assertTrue(check_password("test123", hashed))
+
+    def test_hash_password_empty_fails(self):
+        """Test that empty password raises error."""
+        with self.assertRaises(CommandError) as ctx:
+            call_command("hash_password", "")
+
+        self.assertIn("Password cannot be empty", str(ctx.exception))
--- a/authentik/core/tests/test_interface_views.py
+++ b/authentik/core/tests/test_interface_views.py
@@ -4,6 +4,7 @@ from django.test import TestCase
 from django.urls import reverse

 from authentik.brands.models import Brand
+from authentik.core.apps import Setup
 from authentik.core.models import Application, UserTypes
 from authentik.core.tests.utils import create_test_brand, create_test_user

@@ -12,6 +13,7 @@ class TestInterfaceRedirects(TestCase):
    """Test RootRedirectView and BrandDefaultRedirectView redirect logic by user type"""

    def setUp(self):
+        Setup.set(True)
        self.app = Application.objects.create(name="test-app", slug="test-app")
        self.brand: Brand = create_test_brand(default_application=self.app)

--- a/authentik/core/tests/test_object_attributes_api.py
+++ b/authentik/core/tests/test_object_attributes_api.py
@@ -1,198 +0,0 @@
-"""Test object attributes API"""
-
-from django.urls import reverse
-from rest_framework.test import APITestCase
-
-from authentik.core.api.object_attributes import ContentType
-from authentik.core.models import ObjectAttribute, User
-from authentik.core.tests.utils import create_test_admin_user, create_test_user
-from authentik.lib.generators import generate_id
-
-
-class TestObjectAttributesAPI(APITestCase):
-    """Test object attributes API"""
-
-    def setUp(self) -> None:
-        super().setUp()
-        self.user = create_test_admin_user()
-        self.client.force_login(self.user)
-
-    def test_create(self):
-        res = self.client.post(
-            reverse("authentik_api:objectattribute-list"),
-            data={
-                "object_type": "authentik_core.user",
-                "enabled": False,
-                "key": "employeeNumber",
-                "label": "Employee Number",
-                "type": "text",
-                "group": "Employee",
-                "is_unique": False,
-                "is_required": False,
-            },
-        )
-        self.assertEqual(res.status_code, 201)
-        attr = ObjectAttribute.objects.filter(key="employeeNumber").first()
-        self.assertIsNotNone(attr)
-
-    def test_create_invalid(self):
-        res = self.client.post(
-            reverse("authentik_api:objectattribute-list"),
-            data={
-                "object_type": "authentik_core.objectattribute",
-                "enabled": False,
-                "key": "employeeNumber",
-                "label": "Employee Number",
-                "type": "text",
-                "group": "Employee",
-                "is_unique": False,
-                "is_required": False,
-            },
-        )
-        self.assertEqual(res.status_code, 400)
-        self.assertJSONEqual(res.content, {"object_type": ["Invalid object type"]})
-
-    def test_create_invalid_array_unique(self):
-        res = self.client.post(
-            reverse("authentik_api:objectattribute-list"),
-            data={
-                "object_type": "authentik_core.user",
-                "enabled": False,
-                "key": "employeeNumber",
-                "label": "Employee Number",
-                "type": "text",
-                "group": "Employee",
-                "is_unique": True,
-                "is_required": False,
-                "is_array": True,
-            },
-        )
-        self.assertEqual(res.status_code, 400)
-        self.assertJSONEqual(
-            res.content, {"non_field_errors": ["Unique cannot be enabled for arrays."]}
-        )
-
-    def test_update(self):
-        attr = ObjectAttribute.objects.create(
-            object_type=ContentType.objects.get_for_model(User),
-            label="foo",
-            key=generate_id(),
-            type=ObjectAttribute.AttributeType.TEXT,
-        )
-        res = self.client.put(
-            reverse("authentik_api:objectattribute-detail", kwargs={"pk": attr.pk}),
-            data={
-                "object_type": "authentik_core.user",
-                "enabled": False,
-                "key": attr.key,
-                "label": "Employee Number",
-                "type": "text",
-                "group": "Employee",
-                "is_unique": False,
-                "is_required": False,
-            },
-        )
-        self.assertEqual(res.status_code, 200)
-        attr.refresh_from_db()
-        self.assertEqual(attr.label, "Employee Number")
-
-    def test_user_attrib_validation_required(self):
-        attr = ObjectAttribute.objects.create(
-            object_type=ContentType.objects.get_for_model(User),
-            label="foo",
-            key=generate_id(),
-            type=ObjectAttribute.AttributeType.TEXT,
-            is_required=True,
-        )
-        res = self.client.patch(
-            reverse("authentik_api:user-detail", kwargs={"pk": self.user.pk}),
-            data={
-                "attributes": {},
-            },
-        )
-        self.assertEqual(res.status_code, 400)
-        self.assertJSONEqual(res.content, {f"attributes_{attr.key}": ["This field is required"]})
-
-    def test_user_attrib_validation_unique(self):
-        attr = ObjectAttribute.objects.create(
-            object_type=ContentType.objects.get_for_model(User),
-            label="foo",
-            key=generate_id(),
-            type=ObjectAttribute.AttributeType.TEXT,
-            is_unique=True,
-        )
-        other_user = create_test_user()
-        other_user.attributes[attr.key] = "foo"
-        other_user.save()
-        res = self.client.patch(
-            reverse("authentik_api:user-detail", kwargs={"pk": self.user.pk}),
-            data={
-                "attributes": {attr.key: "foo"},
-            },
-        )
-        self.assertEqual(res.status_code, 400)
-        self.assertJSONEqual(res.content, {f"attributes_{attr.key}": ["Value is not unique."]})
-
-    def test_user_attrib_validation_regex(self):
-        attr = ObjectAttribute.objects.create(
-            object_type=ContentType.objects.get_for_model(User),
-            label="foo",
-            key=generate_id(),
-            type=ObjectAttribute.AttributeType.TEXT,
-            regex="bar",
-        )
-        res = self.client.patch(
-            reverse("authentik_api:user-detail", kwargs={"pk": self.user.pk}),
-            data={
-                "attributes": {attr.key: "foo"},
-            },
-        )
-        self.assertEqual(res.status_code, 400)
-        self.assertJSONEqual(
-            res.content, {f"attributes_{attr.key}": ["Value does not match configured pattern."]}
-        )
-
-    def test_user_attrib_validation_array(self):
-        attr = ObjectAttribute.objects.create(
-            object_type=ContentType.objects.get_for_model(User),
-            label="foo",
-            key=generate_id(),
-            type=ObjectAttribute.AttributeType.TEXT,
-            is_array=True,
-        )
-        res = self.client.patch(
-            reverse("authentik_api:user-detail", kwargs={"pk": self.user.pk}),
-            data={
-                "attributes": {attr.key: "foo"},
-            },
-        )
-        self.assertEqual(res.status_code, 400)
-        self.assertJSONEqual(res.content, {f"attributes_{attr.key}": ["Value must be an array."]})
-
-        res = self.client.patch(
-            reverse("authentik_api:user-detail", kwargs={"pk": self.user.pk}),
-            data={
-                "attributes": {attr.key: ["foo"]},
-            },
-        )
-        self.assertEqual(res.status_code, 200)
-
-    def test_user_attrib_validation_array_regex(self):
-        attr = ObjectAttribute.objects.create(
-            object_type=ContentType.objects.get_for_model(User),
-            label="foo",
-            key=generate_id(),
-            type=ObjectAttribute.AttributeType.TEXT,
-            is_array=True,
-            regex="bar",
-        )
-        res = self.client.patch(
-            reverse("authentik_api:user-detail", kwargs={"pk": self.user.pk}),
-            data={
-                "attributes": {attr.key: ["foo"]},
-            },
-        )
-        self.assertEqual(res.status_code, 400)
-        self.assertJSONEqual(
-            res.content, {f"attributes_{attr.key}": ["Value does not match configured pattern."]}
-        )
--- a/authentik/core/tests/test_setup.py
+++ b/authentik/core/tests/test_setup.py
@@ -0,0 +1,174 @@
+from http import HTTPStatus
+from os import environ
+
+from django.contrib.auth.hashers import make_password
+from django.urls import reverse
+
+from authentik.blueprints.tests import apply_blueprint
+from authentik.core.apps import Setup
+from authentik.core.models import Token, TokenIntents, User
+from authentik.flows.models import Flow
+from authentik.flows.tests import FlowTestCase
+from authentik.lib.generators import generate_id
+from authentik.root.signals import post_startup, pre_startup
+from authentik.tenants.flags import patch_flag
+
+
+class TestSetup(FlowTestCase):
+    def tearDown(self):
+        environ.pop("AUTHENTIK_BOOTSTRAP_PASSWORD", None)
+        environ.pop("AUTHENTIK_BOOTSTRAP_PASSWORD_HASH", None)
+        environ.pop("AUTHENTIK_BOOTSTRAP_TOKEN", None)
+
+    @patch_flag(Setup, True)
+    def test_setup(self):
+        """Test existing instance"""
+        res = self.client.get(reverse("authentik_core:root-redirect"))
+        self.assertEqual(res.status_code, HTTPStatus.FOUND)
+        self.assertRedirects(
+            res,
+            reverse("authentik_flows:default-authentication") + "?next=/",
+            fetch_redirect_response=False,
+        )
+
+        res = self.client.head(reverse("authentik_core:setup"))
+        self.assertEqual(res.status_code, HTTPStatus.SERVICE_UNAVAILABLE)
+
+        res = self.client.get(reverse("authentik_core:setup"))
+        self.assertEqual(res.status_code, HTTPStatus.FOUND)
+        self.assertRedirects(
+            res,
+            reverse("authentik_core:root-redirect"),
+            fetch_redirect_response=False,
+        )
+
+    @patch_flag(Setup, False)
+    def test_not_setup_no_flow(self):
+        """Test case on initial startup; setup flag is not set and oobe flow does
+        not exist yet"""
+        Flow.objects.filter(slug="initial-setup").delete()
+        res = self.client.get(reverse("authentik_core:root-redirect"))
+        self.assertEqual(res.status_code, HTTPStatus.FOUND)
+        self.assertRedirects(res, reverse("authentik_core:setup"), fetch_redirect_response=False)
+        # Flow does not exist, hence 503
+        res = self.client.get(reverse("authentik_core:setup"))
+        self.assertEqual(res.status_code, HTTPStatus.SERVICE_UNAVAILABLE)
+        res = self.client.head(reverse("authentik_core:setup"))
+        self.assertEqual(res.status_code, HTTPStatus.SERVICE_UNAVAILABLE)
+
+    @patch_flag(Setup, False)
+    @apply_blueprint("default/flow-oobe.yaml")
+    def test_not_setup(self):
+        """Test case for when worker comes up, and has created flow"""
+        res = self.client.get(reverse("authentik_core:root-redirect"))
+        self.assertEqual(res.status_code, HTTPStatus.FOUND)
+        self.assertRedirects(res, reverse("authentik_core:setup"), fetch_redirect_response=False)
+        # Flow does not exist, hence 503
+        res = self.client.head(reverse("authentik_core:setup"))
+        self.assertEqual(res.status_code, HTTPStatus.OK)
+        res = self.client.get(reverse("authentik_core:setup"))
+        self.assertEqual(res.status_code, HTTPStatus.FOUND)
+        self.assertRedirects(
+            res,
+            reverse("authentik_core:if-flow", kwargs={"flow_slug": "initial-setup"}),
+            fetch_redirect_response=False,
+        )
+
+    @apply_blueprint("default/flow-oobe.yaml")
+    @apply_blueprint("system/bootstrap.yaml")
+    def test_setup_flow_full(self):
+        """Test full setup flow"""
+        Setup.set(False)
+
+        res = self.client.get(reverse("authentik_core:setup"))
+        self.assertEqual(res.status_code, HTTPStatus.FOUND)
+        self.assertRedirects(
+            res,
+            reverse("authentik_core:if-flow", kwargs={"flow_slug": "initial-setup"}),
+            fetch_redirect_response=False,
+        )
+
+        res = self.client.get(
+            reverse("authentik_api:flow-executor", kwargs={"flow_slug": "initial-setup"}),
+        )
+        self.assertEqual(res.status_code, HTTPStatus.OK)
+        self.assertStageResponse(res, component="ak-stage-prompt")
+
+        pw = generate_id()
+        res = self.client.post(
+            reverse("authentik_api:flow-executor", kwargs={"flow_slug": "initial-setup"}),
+            {
+                "email": f"{generate_id()}@t.goauthentik.io",
+                "password": pw,
+                "password_repeat": pw,
+                "component": "ak-stage-prompt",
+            },
+        )
+        self.assertEqual(res.status_code, HTTPStatus.FOUND)
+
+        res = self.client.get(
+            reverse("authentik_api:flow-executor", kwargs={"flow_slug": "initial-setup"}),
+        )
+        self.assertEqual(res.status_code, HTTPStatus.FOUND)
+
+        res = self.client.get(
+            reverse("authentik_api:flow-executor", kwargs={"flow_slug": "initial-setup"}),
+        )
+        self.assertEqual(res.status_code, HTTPStatus.FOUND)
+
+        res = self.client.get(
+            reverse("authentik_api:flow-executor", kwargs={"flow_slug": "initial-setup"}),
+        )
+        self.assertEqual(res.status_code, HTTPStatus.OK)
+
+        self.assertTrue(Setup.get())
+        user = User.objects.get(username="akadmin")
+        self.assertTrue(user.check_password(pw))
+
+    @patch_flag(Setup, False)
+    @apply_blueprint("default/flow-oobe.yaml")
+    @apply_blueprint("system/bootstrap.yaml")
+    def test_setup_flow_direct(self):
+        """Test setup flow, directly accessing the flow"""
+        res = self.client.get(
+            reverse("authentik_api:flow-executor", kwargs={"flow_slug": "initial-setup"})
+        )
+        self.assertStageResponse(
+            res,
+            component="ak-stage-access-denied",
+            error_message="Access the authentik setup by navigating to http://testserver/",
+        )
+
+    def test_setup_bootstrap_env(self):
+        """Test setup with env vars"""
+        User.objects.filter(username="akadmin").delete()
+        Setup.set(False)
+
+        environ["AUTHENTIK_BOOTSTRAP_PASSWORD"] = generate_id()
+        environ["AUTHENTIK_BOOTSTRAP_TOKEN"] = generate_id()
+        pre_startup.send(sender=self)
+        post_startup.send(sender=self)
+
+        self.assertTrue(Setup.get())
+        user = User.objects.get(username="akadmin")
+        self.assertTrue(user.check_password(environ["AUTHENTIK_BOOTSTRAP_PASSWORD"]))
+
+        token = Token.objects.filter(identifier="authentik-bootstrap-token").first()
+        self.assertEqual(token.intent, TokenIntents.INTENT_API)
+        self.assertEqual(token.key, environ["AUTHENTIK_BOOTSTRAP_TOKEN"])
+
+    def test_setup_bootstrap_env_password_hash(self):
+        """Test setup with password hash env var"""
+        User.objects.filter(username="akadmin").delete()
+        Setup.set(False)
+
+        password = generate_id()
+        password_hash = make_password(password)
+        environ["AUTHENTIK_BOOTSTRAP_PASSWORD_HASH"] = password_hash
+        pre_startup.send(sender=self)
+        post_startup.send(sender=self)
+
+        self.assertTrue(Setup.get())
+        user = User.objects.get(username="akadmin")
+        self.assertEqual(user.password, password_hash)
+        self.assertTrue(user.check_password(password))
--- a/authentik/core/tests/test_users.py
+++ b/authentik/core/tests/test_users.py
@@ -1,8 +1,15 @@
 """user tests"""

-from django.test.testcases import TestCase
+from unittest.mock import patch

+from django.contrib.auth.hashers import make_password
+from django.test.testcases import TestCase
+from rest_framework.exceptions import ValidationError
+
+from authentik.blueprints.v1.importer import SERIALIZER_CONTEXT_BLUEPRINT
+from authentik.core.api.users import UserSerializer
 from authentik.core.models import User
+from authentik.core.signals import password_changed, password_hash_changed
 from authentik.events.models import Event
 from authentik.lib.generators import generate_id

@@ -33,3 +40,99 @@ class TestUsers(TestCase):
        self.assertEqual(Event.objects.count(), 1)
        user.ak_groups.all()
        self.assertEqual(Event.objects.count(), 1)
+
+    def test_set_password_from_hash_signal_skips_source_sync_receivers(self):
+        """Test hash password updates do not expose a raw password to sync receivers."""
+        user = User.objects.create(
+            username=generate_id(),
+            attributes={"distinguishedName": "cn=test,ou=users,dc=example,dc=com"},
+        )
+        password_changed_captured = []
+        password_hash_changed_captured = []
+        dispatch_uid = generate_id()
+        hash_dispatch_uid = generate_id()
+
+        def password_changed_receiver(sender, **kwargs):
+            password_changed_captured.append(kwargs)
+
+        def password_hash_changed_receiver(sender, **kwargs):
+            password_hash_changed_captured.append(kwargs)
+
+        password_changed.connect(password_changed_receiver, dispatch_uid=dispatch_uid)
+        password_hash_changed.connect(
+            password_hash_changed_receiver, dispatch_uid=hash_dispatch_uid
+        )
+        try:
+            with (
+                patch(
+                    "authentik.sources.ldap.signals.LDAPSource.objects.filter"
+                ) as ldap_sources_filter,
+                patch(
+                    "authentik.sources.kerberos.signals."
+                    "UserKerberosSourceConnection.objects.select_related"
+                ) as kerberos_connections_select,
+            ):
+                user.set_password_from_hash(make_password("new-password"))  # nosec
+                user.save()
+        finally:
+            password_changed.disconnect(dispatch_uid=dispatch_uid)
+            password_hash_changed.disconnect(dispatch_uid=hash_dispatch_uid)
+
+        self.assertEqual(password_changed_captured, [])
+        self.assertEqual(len(password_hash_changed_captured), 1)
+        ldap_sources_filter.assert_not_called()
+        kerberos_connections_select.assert_not_called()
+
+
+class TestUserSerializerPasswordHash(TestCase):
+    """Test UserSerializer password_hash support in blueprint context."""
+
+    def test_password_hash_sets_password_directly(self):
+        """Test a valid password hash is stored without re-hashing."""
+        password = "test-password-123"  # nosec
+        password_hash = make_password(password)
+        serializer = UserSerializer(
+            data={
+                "username": generate_id(),
+                "name": "Test User",
+                "password_hash": password_hash,
+            },
+            context={SERIALIZER_CONTEXT_BLUEPRINT: True},
+        )
+
+        self.assertTrue(serializer.is_valid(), serializer.errors)
+        user = serializer.save()
+
+        self.assertEqual(user.password, password_hash)
+        self.assertTrue(user.check_password(password))
+        self.assertIsNotNone(user.password_change_date)
+
+    def test_password_hash_rejects_invalid_format(self):
+        """Test invalid password hash values are rejected."""
+        serializer = UserSerializer(
+            data={
+                "username": generate_id(),
+                "name": "Test User",
+                "password_hash": "not-a-valid-hash",
+            },
+            context={SERIALIZER_CONTEXT_BLUEPRINT: True},
+        )
+
+        self.assertTrue(serializer.is_valid(), serializer.errors)
+        with self.assertRaises(ValidationError) as ctx:
+            serializer.save()
+
+        self.assertIn("Invalid password hash format", str(ctx.exception))
+
+    def test_password_hash_ignored_outside_blueprint_context(self):
+        """Test password_hash is not accepted by the regular serializer."""
+        serializer = UserSerializer(
+            data={
+                "username": generate_id(),
+                "name": "Test User",
+                "password_hash": make_password("test"),  # nosec
+            }
+        )
+
+        self.assertTrue(serializer.is_valid(), serializer.errors)
+        self.assertNotIn("password_hash", serializer.validated_data)
--- a/authentik/core/tests/test_users_api.py
+++ b/authentik/core/tests/test_users_api.py
@@ -3,6 +3,7 @@
 from datetime import datetime, timedelta
 from json import loads

+from django.contrib.auth.hashers import make_password
 from django.urls.base import reverse
 from django.utils.timezone import now
 from rest_framework.test import APITestCase
@@ -26,6 +27,9 @@ from authentik.flows.models import FlowAuthenticationRequirement, FlowDesignatio
 from authentik.lib.generators import generate_id, generate_key
 from authentik.stages.email.models import EmailStage

+INVALID_PASSWORD_HASH = "not-a-valid-hash"
+INVALID_PASSWORD_HASH_ERROR = "Invalid password hash format. Must be a valid Django password hash."
+

 class TestUsersAPI(APITestCase):
    """Test Users API"""
@@ -34,6 +38,20 @@ class TestUsersAPI(APITestCase):
        self.admin = create_test_admin_user()
        self.user = create_test_user()

+    def _set_password_hash(self, user: User, password_hash: str, client=None):
+        return (client or self.client).post(
+            reverse("authentik_api:user-set-password-hash", kwargs={"pk": user.pk}),
+            data={"password": password_hash},
+        )
+
+    def _assert_password_hash_set(
+        self, user: User, password: str, password_hash: str, response
+    ) -> None:
+        self.assertEqual(response.status_code, 204, response.data)
+        user.refresh_from_db()
+        self.assertEqual(user.password, password_hash)
+        self.assertTrue(user.check_password(password))
+
    def test_filter_type(self):
        """Test API filtering by type"""
        self.client.force_login(self.admin)
@@ -113,6 +131,26 @@ class TestUsersAPI(APITestCase):
        self.assertEqual(response.status_code, 400)
        self.assertJSONEqual(response.content, {"password": ["This field may not be blank."]})

+    def test_set_password_hash(self):
+        """Test setting a user's password from a hash."""
+        self.client.force_login(self.admin)
+        password = generate_key()
+        password_hash = make_password(password)
+        response = self._set_password_hash(self.user, password_hash)
+
+        self._assert_password_hash_set(self.user, password, password_hash, response)
+
+    def test_set_password_hash_invalid(self):
+        """Test invalid password hashes are rejected."""
+        self.client.force_login(self.admin)
+        response = self._set_password_hash(self.user, INVALID_PASSWORD_HASH)
+
+        self.assertEqual(response.status_code, 400)
+        self.assertJSONEqual(
+            response.content,
+            {"password": [INVALID_PASSWORD_HASH_ERROR]},
+        )
+
    def test_recovery(self):
        """Test user recovery link"""
        flow = create_test_flow(
@@ -261,6 +299,29 @@ class TestUsersAPI(APITestCase):
        self.assertTrue(token_filter.exists())
        self.assertTrue(token_filter.first().expiring)

+    def test_service_account_set_password_hash(self):
+        """Service account password hash can be set through the API."""
+        self.client.force_login(self.admin)
+        response = self.client.post(
+            reverse("authentik_api:user-service-account"),
+            data={
+                "name": "test-sa",
+                "create_group": False,
+            },
+        )
+        self.assertEqual(response.status_code, 200, response.data)
+        body = loads(response.content)
+
+        user = User.objects.get(pk=body["user_pk"])
+        self.assertEqual(user.type, UserTypes.SERVICE_ACCOUNT)
+        self.assertFalse(user.has_usable_password())
+
+        password = generate_key()
+        password_hash = make_password(password)
+        response = self._set_password_hash(user, password_hash)
+
+        self._assert_password_hash_set(user, password, password_hash, response)
+
    def test_service_account_no_expire(self):
        """Service account creation without token expiration"""
        self.client.force_login(self.admin)
--- a/authentik/core/urls.py
+++ b/authentik/core/urls.py
@@ -1,7 +1,6 @@
 """authentik URL Configuration"""

 from django.conf import settings
-from django.contrib.auth.decorators import login_required
 from django.urls import path

 from authentik.core.api.application_entitlements import ApplicationEntitlementViewSet
@@ -9,7 +8,6 @@ from authentik.core.api.applications import ApplicationViewSet
 from authentik.core.api.authenticated_sessions import AuthenticatedSessionViewSet
 from authentik.core.api.devices import AdminDeviceViewSet, DeviceViewSet
 from authentik.core.api.groups import GroupViewSet
-from authentik.core.api.object_attributes import ObjectAttributeViewSet
 from authentik.core.api.property_mappings import PropertyMappingViewSet
 from authentik.core.api.providers import ProviderViewSet
 from authentik.core.api.sources import (
@@ -20,6 +18,7 @@ from authentik.core.api.sources import (
 from authentik.core.api.tokens import TokenViewSet
 from authentik.core.api.transactional_applications import TransactionalApplicationView
 from authentik.core.api.users import UserViewSet
+from authentik.core.setup.views import SetupView
 from authentik.core.views.apps import RedirectToAppLaunch
 from authentik.core.views.debug import AccessDeniedView
 from authentik.core.views.interface import (
@@ -36,7 +35,7 @@ from authentik.tenants.channels import TenantsAwareMiddleware
 urlpatterns = [
    path(
        "",
-        login_required(RootRedirectView.as_view()),
+        RootRedirectView.as_view(),
        name="root-redirect",
    ),
    path(
@@ -63,6 +62,11 @@ urlpatterns = [
        FlowInterfaceView.as_view(),
        name="if-flow",
    ),
+    path(
+        "setup",
+        SetupView.as_view(),
+        name="setup",
+    ),
    # Fallback for WS
    path("ws/outpost/<uuid:pk>/", InterfaceView.as_view(template_name="if/admin.html")),
    path(
@@ -83,7 +87,6 @@ api_urlpatterns = [
    ("core/groups", GroupViewSet),
    ("core/users", UserViewSet),
    ("core/tokens", TokenViewSet),
-    ("core/object_attributes", ObjectAttributeViewSet),
    ("sources/all", SourceViewSet),
    ("sources/user_connections/all", UserSourceConnectionViewSet),
    ("sources/group_connections/all", GroupSourceConnectionViewSet),
--- a/authentik/core/views/interface.py
+++ b/authentik/core/views/interface.py
@@ -3,6 +3,7 @@
 from json import dumps
 from typing import Any

+from django.contrib.auth.mixins import AccessMixin
 from django.http import HttpRequest
 from django.http.response import HttpResponse
 from django.shortcuts import redirect
@@ -14,12 +15,13 @@ from authentik.admin.tasks import LOCAL_VERSION
 from authentik.api.v3.config import ConfigView
 from authentik.brands.api import CurrentBrandSerializer
 from authentik.brands.models import Brand
+from authentik.core.apps import Setup
 from authentik.core.models import UserTypes
 from authentik.lib.config import CONFIG
 from authentik.policies.denied import AccessDeniedResponse


-class RootRedirectView(RedirectView):
+class RootRedirectView(AccessMixin, RedirectView):
    """Root redirect view, redirect to brand's default application if set"""

    pattern_name = "authentik_core:if-user"
@@ -40,6 +42,10 @@ class RootRedirectView(RedirectView):
        return None

    def dispatch(self, request: HttpRequest, *args: Any, **kwargs: Any) -> HttpResponse:
+        if not Setup.get():
+            return redirect("authentik_core:setup")
+        if not request.user.is_authenticated:
+            return self.handle_no_permission()
        if redirect_response := RootRedirectView().redirect_to_app(request):
            return redirect_response
        return super().dispatch(request, *args, **kwargs)
--- a/authentik/endpoints/api/device_access_group.py
+++ b/authentik/endpoints/api/device_access_group.py
@@ -1,12 +1,11 @@
 from rest_framework.viewsets import ModelViewSet

-from authentik.core.api.object_attributes import AttributesMixinSerializer
 from authentik.core.api.used_by import UsedByMixin
 from authentik.core.api.utils import ModelSerializer
 from authentik.endpoints.models import DeviceAccessGroup


-class DeviceAccessGroupSerializer(AttributesMixinSerializer, ModelSerializer):
+class DeviceAccessGroupSerializer(ModelSerializer):

    class Meta:
        model = DeviceAccessGroup
--- a/authentik/endpoints/connectors/agent/controller.py
+++ b/authentik/endpoints/connectors/agent/controller.py
@@ -138,13 +138,7 @@ class AgentConnectorController(BaseController[AgentConnector]):
                            "AllowDeviceIdentifiersInAttestation": True,
                            "AuthenticationMethod": "UserSecureEnclaveKey",
                            "EnableAuthorization": True,
-                            "EnableCreateUserAtLogin": True,
-                            "FileVaultPolicy": ["RequireAuthentication"],
-                            "LoginPolicy": ["RequireAuthentication"],
-                            "NewUserAuthorizationMode": "Standard",
-                            "UnlockPolicy": ["RequireAuthentication"],
                            "UseSharedDeviceKeys": True,
-                            "UserAuthorizationMode": "Standard",
                        },
                    },
                ],
--- a/authentik/endpoints/connectors/agent/migrations/0005_appleindependentsecureenclave.py
+++ b/authentik/endpoints/connectors/agent/migrations/0005_appleindependentsecureenclave.py
@@ -0,0 +1,54 @@
+# Generated by Django 5.2.12 on 2026-03-06 14:38
+
+import django.db.models.deletion
+import uuid
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        (
+            "authentik_endpoints_connectors_agent",
+            "0004_agentconnector_challenge_idle_timeout_and_more",
+        ),
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="AppleIndependentSecureEnclave",
+            fields=[
+                ("created", models.DateTimeField(auto_now_add=True)),
+                ("last_updated", models.DateTimeField(auto_now=True)),
+                (
+                    "name",
+                    models.CharField(
+                        help_text="The human-readable name of this device.", max_length=64
+                    ),
+                ),
+                (
+                    "confirmed",
+                    models.BooleanField(default=True, help_text="Is this device ready for use?"),
+                ),
+                ("last_used", models.DateTimeField(null=True)),
+                ("uuid", models.UUIDField(default=uuid.uuid4, primary_key=True, serialize=False)),
+                ("apple_secure_enclave_key", models.TextField()),
+                ("apple_enclave_key_id", models.TextField()),
+                ("device_type", models.TextField()),
+                (
+                    "user",
+                    models.ForeignKey(
+                        help_text="The user that this device belongs to.",
+                        on_delete=django.db.models.deletion.CASCADE,
+                        to=settings.AUTH_USER_MODEL,
+                    ),
+                ),
+            ],
+            options={
+                "verbose_name": "Apple Independent Secure Enclave",
+                "verbose_name_plural": "Apple Independent Secure Enclaves",
+            },
+        ),
+    ]
--- a/authentik/endpoints/connectors/agent/models.py
+++ b/authentik/endpoints/connectors/agent/models.py
@@ -19,6 +19,7 @@ from authentik.flows.stage import StageView
 from authentik.lib.generators import generate_key
 from authentik.lib.models import InternallyManagedMixin, SerializerModel
 from authentik.lib.utils.time import timedelta_string_validator
+from authentik.stages.authenticator.models import Device as Authenticator

 if TYPE_CHECKING:
    from authentik.endpoints.connectors.agent.controller import AgentConnectorController
@@ -172,3 +173,17 @@ class AppleNonce(InternallyManagedMixin, ExpiringModel):
    class Meta(ExpiringModel.Meta):
        verbose_name = _("Apple Nonce")
        verbose_name_plural = _("Apple Nonces")
+
+
+class AppleIndependentSecureEnclave(Authenticator):
+    """A device-independent secure enclave key, used by Tap-to-login"""
+
+    uuid = models.UUIDField(primary_key=True, default=uuid4)
+
+    apple_secure_enclave_key = models.TextField()
+    apple_enclave_key_id = models.TextField()
+    device_type = models.TextField()
+
+    class Meta:
+        verbose_name = _("Apple Independent Secure Enclave")
+        verbose_name_plural = _("Apple Independent Secure Enclaves")
--- a/authentik/endpoints/tests/test_api.py
+++ b/authentik/endpoints/tests/test_api.py
@@ -1,10 +1,12 @@
+from unittest.mock import PropertyMock, patch
+
 from django.urls import reverse
 from rest_framework.test import APITestCase

 from authentik.core.tests.utils import create_test_admin_user
 from authentik.endpoints.connectors.agent.models import AgentConnector
+from authentik.endpoints.controller import BaseController
 from authentik.endpoints.models import StageMode
-from authentik.enterprise.endpoints.connectors.fleet.models import FleetConnector
 from authentik.lib.generators import generate_id


@@ -25,16 +27,22 @@ class TestAPI(APITestCase):
        )
        self.assertEqual(res.status_code, 201)

-    def test_endpoint_stage_fleet(self):
-        connector = FleetConnector.objects.create(name=generate_id())
-        res = self.client.post(
-            reverse("authentik_api:stages-endpoint-list"),
-            data={
-                "name": generate_id(),
-                "connector": str(connector.pk),
-                "mode": StageMode.REQUIRED,
-            },
-        )
+    def test_endpoint_stage_agent_no_stage(self):
+        connector = AgentConnector.objects.create(name=generate_id())
+
+        class controller(BaseController):
+            def capabilities(self):
+                return []
+
+        with patch.object(AgentConnector, "controller", PropertyMock(return_value=controller)):
+            res = self.client.post(
+                reverse("authentik_api:stages-endpoint-list"),
+                data={
+                    "name": generate_id(),
+                    "connector": str(connector.pk),
+                    "mode": StageMode.REQUIRED,
+                },
+            )
        self.assertEqual(res.status_code, 400)
        self.assertJSONEqual(
            res.content, {"connector": ["Selected connector is not compatible with this stage."]}
--- a/authentik/enterprise/endpoints/connectors/agent/api/secure_enclave.py
+++ b/authentik/enterprise/endpoints/connectors/agent/api/secure_enclave.py
@@ -0,0 +1,28 @@
+from rest_framework.viewsets import ModelViewSet
+
+from authentik.core.api.used_by import UsedByMixin
+from authentik.core.api.utils import ModelSerializer
+from authentik.endpoints.connectors.agent.models import AppleIndependentSecureEnclave
+
+
+class AppleIndependentSecureEnclaveSerializer(ModelSerializer):
+    class Meta:
+        model = AppleIndependentSecureEnclave
+        fields = [
+            "uuid",
+            "user",
+            "apple_secure_enclave_key",
+            "apple_enclave_key_id",
+            "device_type",
+        ]
+
+
+class AppleIndependentSecureEnclaveViewSet(UsedByMixin, ModelViewSet):
+    queryset = AppleIndependentSecureEnclave.objects.all()
+    serializer_class = AppleIndependentSecureEnclaveSerializer
+    search_fields = [
+        "name",
+        "user__name",
+    ]
+    ordering = ["uuid"]
+    filterset_fields = ["user", "apple_enclave_key_id"]
--- a/authentik/enterprise/endpoints/connectors/agent/tests/test_apple_token.py
+++ b/authentik/enterprise/endpoints/connectors/agent/tests/test_apple_token.py
@@ -11,6 +11,7 @@ from authentik.endpoints.connectors.agent.models import (
    AgentConnector,
    AgentDeviceConnection,
    AgentDeviceUserBinding,
+    AppleIndependentSecureEnclave,
    AppleNonce,
    DeviceToken,
    EnrollmentToken,
@@ -25,7 +26,7 @@ class TestAppleToken(TestCase):

    def setUp(self):
        self.apple_sign_key = create_test_cert(PrivateKeyAlg.ECDSA)
-        sign_key_pem = self.apple_sign_key.public_key.public_bytes(
+        self.sign_key_pem = self.apple_sign_key.public_key.public_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PublicFormat.SubjectPublicKeyInfo,
        ).decode()
@@ -50,7 +51,7 @@ class TestAppleToken(TestCase):
            device=self.device,
            connector=self.connector,
            apple_sign_key_id=self.apple_sign_key.kid,
-            apple_signing_key=sign_key_pem,
+            apple_signing_key=self.sign_key_pem,
            apple_encryption_key=self.enc_pub,
        )
        self.user = create_test_user()
@@ -59,7 +60,7 @@ class TestAppleToken(TestCase):
            user=self.user,
            order=0,
            apple_enclave_key_id=self.apple_sign_key.kid,
-            apple_secure_enclave_key=sign_key_pem,
+            apple_secure_enclave_key=self.sign_key_pem,
        )
        self.device_token = DeviceToken.objects.create(device=self.connection)

@@ -113,3 +114,62 @@ class TestAppleToken(TestCase):
        ).first()
        self.assertIsNotNone(event)
        self.assertEqual(event.context["device"]["name"], self.device.name)
+
+    @reconcile_app("authentik_crypto")
+    def test_token_independent(self):
+        nonce = generate_id()
+
+        AgentDeviceUserBinding.objects.all().delete()
+        AppleIndependentSecureEnclave.objects.create(
+            user=self.user,
+            apple_enclave_key_id=self.apple_sign_key.kid,
+            apple_secure_enclave_key=self.sign_key_pem,
+        )
+
+        AppleNonce.objects.create(
+            device_token=self.device_token,
+            nonce=nonce,
+        )
+        embedded = encode(
+            {"iss": str(self.connector.pk), "aud": str(self.device.pk), "request_nonce": nonce},
+            self.apple_sign_key.private_key,
+            headers={
+                "kid": self.apple_sign_key.kid,
+            },
+            algorithm=JWTAlgorithms.from_private_key(self.apple_sign_key.private_key),
+        )
+        assertion = encode(
+            {
+                "iss": str(self.connector.pk),
+                "aud": "http://testserver/endpoints/agent/psso/token/",
+                "request_nonce": nonce,
+                "assertion": embedded,
+                "jwe_crypto": {
+                    "apv": (
+                        "AAAABUFwcGxlAAAAQQTFgZOospN6KbkhXhx1lfa-AKYxjEfJhTJrkpdEY_srMmkPzS7VN0Bzt2AtNBEXE"
+                        "aphDONiP2Mq6Oxytv5JKOxHAAAAJDgyOThERkY5LTVFMUUtNEUwMS04OEUwLUI3QkQzOUM4QjA3Qw"
+                    )
+                },
+            },
+            self.apple_sign_key.private_key,
+            headers={
+                "kid": self.apple_sign_key.kid,
+            },
+            algorithm=JWTAlgorithms.from_private_key(self.apple_sign_key.private_key),
+        )
+        res = self.client.post(
+            reverse("authentik_enterprise_endpoints_connectors_agent:psso-token"),
+            data={
+                "assertion": assertion,
+                "platform_sso_version": "1.0",
+                "grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
+            },
+        )
+
+        self.assertEqual(res.status_code, 200)
+        event = Event.objects.filter(
+            action=EventAction.LOGIN,
+            app="authentik.endpoints.connectors.agent",
+        ).first()
+        self.assertIsNotNone(event)
+        self.assertEqual(event.context["device"]["name"], self.device.name)
--- a/authentik/enterprise/endpoints/connectors/agent/urls.py
+++ b/authentik/enterprise/endpoints/connectors/agent/urls.py
@@ -1,5 +1,8 @@
 from django.urls import path

+from authentik.enterprise.endpoints.connectors.agent.api.secure_enclave import (
+    AppleIndependentSecureEnclaveViewSet,
+)
 from authentik.enterprise.endpoints.connectors.agent.views.apple_jwks import AppleJWKSView
 from authentik.enterprise.endpoints.connectors.agent.views.apple_nonce import NonceView
 from authentik.enterprise.endpoints.connectors.agent.views.apple_register import (
@@ -23,6 +26,7 @@ urlpatterns = [
 ]

 api_urlpatterns = [
+    ("endpoints/agents/psso/ise", AppleIndependentSecureEnclaveViewSet),
    path(
        "endpoints/agents/psso/register/device/",
        RegisterDeviceView.as_view(),
--- a/authentik/enterprise/endpoints/connectors/agent/views/apple_token.py
+++ b/authentik/enterprise/endpoints/connectors/agent/views/apple_token.py
@@ -19,6 +19,7 @@ from authentik.endpoints.connectors.agent.models import (
    AgentConnector,
    AgentDeviceConnection,
    AgentDeviceUserBinding,
+    AppleIndependentSecureEnclave,
    AppleNonce,
    DeviceAuthenticationToken,
 )
@@ -103,7 +104,9 @@ class TokenView(View):
        nonce.delete()
        return decoded

-    def validate_embedded_assertion(self, assertion: str) -> tuple[AgentDeviceUserBinding, dict]:
+    def validate_embedded_assertion(
+        self, assertion: str
+    ) -> tuple[AgentDeviceUserBinding | AppleIndependentSecureEnclave, dict]:
        """Decode an embedded assertion and validate it by looking up the matching device user"""
        decode_unvalidated = get_unverified_header(assertion)
        expected_kid = decode_unvalidated["kid"]
@@ -112,8 +115,13 @@ class TokenView(View):
            target=self.device_connection.device, apple_enclave_key_id=expected_kid
        ).first()
        if not device_user:
-            LOGGER.warning("Could not find device user binding for user")
-            raise ValidationError("Invalid request")
+            independent_user = AppleIndependentSecureEnclave.objects.filter(
+                apple_enclave_key_id=expected_kid
+            ).first()
+            if not independent_user:
+                LOGGER.warning("Could not find device user binding or independent enclave for user")
+                raise ValidationError("Invalid request")
+            device_user = independent_user
        decoded: dict[str, Any] = decode(
            assertion,
            device_user.apple_secure_enclave_key,
--- a/authentik/enterprise/endpoints/connectors/fleet/controller.py
+++ b/authentik/enterprise/endpoints/connectors/fleet/controller.py
@@ -1,11 +1,15 @@
 import re
+from plistlib import loads
 from typing import Any

+from cryptography.hazmat.primitives import serialization
+from cryptography.x509 import load_der_x509_certificate
 from django.db import transaction
 from requests import RequestException
 from rest_framework.exceptions import ValidationError

 from authentik.core.models import User
+from authentik.crypto.models import CertificateKeyPair
 from authentik.endpoints.controller import BaseController, Capabilities, ConnectorSyncException
 from authentik.endpoints.facts import (
    DeviceFacts,
@@ -44,7 +48,7 @@ class FleetController(BaseController[DBC]):
        return "fleetdm.com"

    def capabilities(self) -> list[Capabilities]:
-        return [Capabilities.ENROLL_AUTOMATIC_API]
+        return [Capabilities.STAGE_ENDPOINTS, Capabilities.ENROLL_AUTOMATIC_API]

    def _url(self, path: str) -> str:
        return f"{self.connector.url}{path}"
@@ -76,8 +80,44 @@ class FleetController(BaseController[DBC]):
        except RequestException as exc:
            raise ConnectorSyncException(exc) from exc

+    @property
+    def mtls_ca_managed(self) -> str:
+        return f"goauthentik.io/endpoints/connectors/fleet/{self.connector.pk}"
+
+    def _sync_mtls_ca(self):
+        """Sync conditional access Root CA for mTLS"""
+        try:
+            # Fleet doesn't have an API to just get the Conditional Access Root CA Cert (yet),
+            # hence we fetch the apple config profile and extract it
+            res = self._session.get(self._url("/api/v1/fleet/conditional_access/idp/apple/profile"))
+            res.raise_for_status()
+            profile = loads(res.text).get("PayloadContent", [])
+            raw_cert = None
+            for payload in profile:
+                if payload.get("PayloadIdentifier", "") != "com.fleetdm.conditional-access-ca":
+                    continue
+                raw_cert = payload.get("PayloadContent")
+            if not raw_cert:
+                raise ConnectorSyncException("Failed to get conditional acccess CA")
+        except RequestException as exc:
+            raise ConnectorSyncException(exc) from exc
+        cert = load_der_x509_certificate(raw_cert)
+        CertificateKeyPair.objects.update_or_create(
+            managed=self.mtls_ca_managed,
+            defaults={
+                "name": f"Fleet Endpoint connector {self.connector.name}",
+                "certificate_data": cert.public_bytes(
+                    encoding=serialization.Encoding.PEM,
+                ).decode("utf-8"),
+            },
+        )
+
    @transaction.atomic
    def sync_endpoints(self) -> None:
+        try:
+            self._sync_mtls_ca()
+        except ConnectorSyncException as exc:
+            self.logger.warning("Failed to sync conditional access CA", exc=exc)
        for host in self._paginate_hosts():
            serial = host["hardware_serial"]
            device, _ = Device.objects.get_or_create(
@@ -198,6 +238,8 @@ class FleetController(BaseController[DBC]):
                        for policy in host.get("policies", [])
                    ],
                    "agent_version": fleet_version,
+                    # Host UUID is required for conditional access matching
+                    "uuid": host.get("uuid", "").lower(),
                },
            },
        }
--- a/authentik/enterprise/endpoints/connectors/fleet/models.py
+++ b/authentik/enterprise/endpoints/connectors/fleet/models.py
@@ -51,6 +51,12 @@ class FleetConnector(Connector):
    def component(self) -> str:
        return "ak-endpoints-connector-fleet-form"

+    @property
+    def stage(self):
+        from authentik.enterprise.endpoints.connectors.fleet.stage import FleetStageView
+
+        return FleetStageView
+
    class Meta:
        verbose_name = _("Fleet Connector")
        verbose_name_plural = _("Fleet Connectors")
--- a/authentik/enterprise/endpoints/connectors/fleet/stage.py
+++ b/authentik/enterprise/endpoints/connectors/fleet/stage.py
@@ -0,0 +1,51 @@
+from cryptography.x509 import (
+    Certificate,
+    Extension,
+    SubjectAlternativeName,
+    UniformResourceIdentifier,
+)
+from rest_framework.exceptions import PermissionDenied
+
+from authentik.crypto.models import CertificateKeyPair, fingerprint_sha256
+from authentik.endpoints.models import Device, EndpointStage, StageMode
+from authentik.enterprise.endpoints.connectors.fleet.models import FleetConnector
+from authentik.enterprise.stages.mtls.stage import PLAN_CONTEXT_CERTIFICATE, MTLSStageView
+from authentik.flows.planner import PLAN_CONTEXT_DEVICE
+
+FLEET_CONDITIONAL_ACCESS_URI_PREFIX = "urn:device:apple:uuid:"
+
+
+class FleetStageView(MTLSStageView):
+    def get_authorities(self):
+        stage: EndpointStage = self.executor.current_stage
+        connector = FleetConnector.objects.filter(pk=stage.connector_id).first()
+        controller = connector.controller(connector)
+        kp = CertificateKeyPair.objects.filter(managed=controller.mtls_ca_managed).first()
+        return [kp] if kp else None
+
+    def lookup_device(self, cert: Certificate, mode: StageMode):
+        san_ext: Extension[SubjectAlternativeName] = cert.extensions.get_extension_for_oid(
+            SubjectAlternativeName.oid
+        )
+        raw_values = san_ext.value.get_values_for_type(UniformResourceIdentifier)
+        values = [x.removeprefix(FLEET_CONDITIONAL_ACCESS_URI_PREFIX).lower() for x in raw_values]
+        self.logger.debug("Looking for devices with uuid", fleet_device_uuid=values)
+        device = Device.objects.filter(
+            **{"deviceconnection__devicefactsnapshot__data__vendor__fleetdm.com__uuid__in": values}
+        ).first()
+        if not device and mode == StageMode.REQUIRED:
+            raise PermissionDenied("Failed to find device")
+        self.executor.plan.context[PLAN_CONTEXT_DEVICE] = device
+        self.executor.plan.context[PLAN_CONTEXT_CERTIFICATE] = self._cert_to_dict(cert)
+        return self.executor.stage_ok()
+
+    def dispatch(self, request, *args, **kwargs):
+        stage: EndpointStage = self.executor.current_stage
+        try:
+            cert = self.get_cert(stage.mode)
+            if not cert:
+                return self.executor.stage_ok()
+            self.logger.debug("Received certificate", cert=fingerprint_sha256(cert))
+            return self.lookup_device(cert, stage.mode)
+        except PermissionDenied as exc:
+            return self.executor.stage_invalid(error_message=exc.detail)
--- a/authentik/enterprise/endpoints/connectors/fleet/tests/fixtures/cond_acc_host.pem
+++ b/authentik/enterprise/endpoints/connectors/fleet/tests/fixtures/cond_acc_host.pem
@@ -0,0 +1,23 @@
+-----BEGIN CERTIFICATE-----
+MIIDwDCCAqigAwIBAgIBBDANBgkqhkiG9w0BAQsFADBpMQkwBwYDVQQGEwAxJDAi
+BgNVBAoTG0xvY2FsIGNlcnRpZmljYXRlIGF1dGhvcml0eTEQMA4GA1UECxMHU0NF
+UCBDQTEkMCIGA1UEAxMbRmxlZXQgY29uZGl0aW9uYWwgYWNjZXNzIENBMB4XDTI2
+MDMxODExMTc1NFoXDTI3MDQyMDExMjc1NFowLDEqMCgGA1UEAxMhRmxlZXQgY29u
+ZGl0aW9uYWwgYWNjZXNzIGZvciBPa3RhMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8A
+MIIBCgKCAQEA3xuKxQQ8JSA4qCJ6RfOB7tbQurhwXiaJSLUDG7R5ncdRcd9LH/9y
+5ZyI5kQACOwfICHmv02zR4/CrurfzXabo3CCpvcMdS7JI/FzP1GIIZ5RsR7oPFC6
+JJg3m5BHuoHsUtCD7w0D52WiE7XVfbw47h2ChKmGMhkSrBvQnp3dHFEt8ntbl1/q
+zCSuQaLeR2sQFurBDVBdinEgsvb1YHaYHi4tdFx5joG64Q/nJXyA2OM4hO9uBF+G
+c4UVTzubx5sxwONcPhC9H+eLMpF1VHeU9gAGBlruVusUEYDmlqYQuA+bW5fTr4Zd
+ZmJ5e+CzzUBYHduAML9a5S+1jbxSPZFBSwIDAQABo4GvMIGsMA4GA1UdDwEB/wQE
+AwIHgDATBgNVHSUEDDAKBggrBgEFBQcDAjAdBgNVHQ4EFgQUPrc1+LvbR9WoJIWZ
+7YQa/3IX2w8wHwYDVR0jBBgwFoAUfl92kU2qcH4e+hypez4kEnqMbk4wRQYDVR0R
+BD4wPIY6dXJuOmRldmljZTphcHBsZTp1dWlkOjVCRjQyMkQ2LTZFQUItNTE1Ni1B
+QzVBLTlFQURDOTUyNDcxMzANBgkqhkiG9w0BAQsFAAOCAQEAGfxJ/u4271tnUUTB
+J39YU6z2Ciav+9G3BtbvxBXI57Po7zCE6Z1sVkvYq6Xd0CcItPWRjbSPEy78ZzS0
+By+gPy5fkKc8HHJ5I1wK890xbLBUS1P4EbdVBzI9ggouEa3B2asE10asnzLoKE4C
+0FYWQwrzCsso8yxsJj1S8RKtd6MMbCis/9OQSC8om2tu6cLO+OftVn5DHtNWFidw
+tAl/oHn2cZPUfZGpJGrHNZlp5w1c1dYfQeiPayoQIbsF+8eMV424G76z/8UPhMBs
+R23LByv4TlUOPAGn2TRa2WtLIXs7FgqXRIFW4CjsPsEpXSVNlkYcn/VHY7Jl13zz
+CRQ1Pg==
+-----END CERTIFICATE-----
--- a/authentik/enterprise/endpoints/connectors/fleet/tests/fixtures/cond_acc_profile.mobileconfig
+++ b/authentik/enterprise/endpoints/connectors/fleet/tests/fixtures/cond_acc_profile.mobileconfig
@@ -0,0 +1,46 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
+<plist version="1.0">
+    <dict>
+        <key>PayloadContent</key>
+        <array>
+            <!-- Trusted CA certificate -->
+            <dict>
+                <key>PayloadCertificateFileName</key>
+                <string>conditional_access_ca.der</string>
+                <key>PayloadContent</key>
+                <data>MIIDjzCCAnegAwIBAgIBATANBgkqhkiG9w0BAQsFADBpMQkwBwYDVQQGEwAxJDAiBgNVBAoTG0xvY2FsIGNlcnRpZmljYXRlIGF1dGhvcml0eTEQMA4GA1UECxMHU0NFUCBDQTEkMCIGA1UEAxMbRmxlZXQgY29uZGl0aW9uYWwgYWNjZXNzIENBMB4XDTI1MTIwOTEyMjI1MVoXDTM1MTIwOTEyMjI1MVowaTEJMAcGA1UEBhMAMSQwIgYDVQQKExtMb2NhbCBjZXJ0aWZpY2F0ZSBhdXRob3JpdHkxEDAOBgNVBAsTB1NDRVAgQ0ExJDAiBgNVBAMTG0ZsZWV0IGNvbmRpdGlvbmFsIGFjY2VzcyBDQTCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBANrgCcpzQci2UhH+Dn0eHopnnbx3HbMabMCHXm6xteMVFLrdQJDTFrZCQzcexUgbpPJ0az6mn4szo+E3stn0y2PPWsiAiVhFwp5M9HwNg18rPgDmITv2pM3l/hlEsfggjq6TEVO2gRcq4NujEGagcYX6kp6nWxh6bbRngQ/hlK6mXItWV3x0G9eTcbFObwZhbuC2dNbccytdqbVEIpBjp6fftQnQwAaUVjoyZBFlf1C1cDV4+1jpaVsIj11U1olA33GJCHcZQ4CJEsgh8yiSsvkH5RNf94CGINB5ixsMfppjSXV/vNkWDKEfmUXW2q4ft7KK/L/SRq8QSB4VqTAp2GsCAwEAAaNCMEAwDgYDVR0PAQH/BAQDAgGGMA8GA1UdEwEB/wQFMAMBAf8wHQYDVR0OBBYEFH5fdpFNqnB+HvocqXs+JBJ6jG5OMA0GCSqGSIb3DQEBCwUAA4IBAQAJr4bTGlrANoHStu4Y+OXjGbEQjZOe546Bcln4eWrEB16eaVzfKuZgjJYdcOmp36/v34QY/OCXEIsixrBU5aW/Sr53IK6UQSZV3O3xbBc4Aert7AbeJ4NVGZyelfVQo/5G0qM6k9p0+zpIZqNAzFbhcSPIzuE7ig2OGsFoQU+bXhzk09bsZ+u4BXibzVNfMuMG+DHNv0PRjll272nEPI3bGwHF5tdrnfJG6e9t+qK9j9UqmSlBknHQJNeU5o8IDcmWYjWtOuBzecYsg8pZzXabJqlHTBIz/h7waRe7jtrK+XopK3jghRf9JTL+i0Y8NbVjoNkIoS3xMeRhnNbR9lw1</data>
+                <key>PayloadDescription</key>
+                <string>Fleet conditional access CA certificate</string>
+                <key>PayloadDisplayName</key>
+                <string>Fleet conditional access CA</string>
+                <key>PayloadIdentifier</key>
+                <string>com.fleetdm.conditional-access-ca</string>
+                <key>PayloadType</key>
+                <string>com.apple.security.root</string>
+                <key>PayloadUUID</key>
+                <string>ef1b2231-ad80-5511-9893-1f9838295147</string>
+                <key>PayloadVersion</key>
+                <integer>1</integer>
+            </dict>
+        </array>
+        <key>PayloadDescription</key>
+        <string>Configures SCEP enrollment for Okta conditional access</string>
+        <key>PayloadDisplayName</key>
+        <string>Fleet conditional access for Okta</string>
+        <key>PayloadIdentifier</key>
+        <string>com.fleetdm.conditional-access-okta</string>
+        <key>PayloadOrganization</key>
+        <string>Fleet Device Management</string>
+        <key>PayloadRemovalDisallowed</key>
+        <false/>
+        <key>PayloadScope</key>
+        <string>User</string>
+        <key>PayloadType</key>
+        <string>Configuration</string>
+        <key>PayloadUUID</key>
+        <string>6fa509a3-feca-56f7-a283-d6a81c733ed2</string>
+        <key>PayloadVersion</key>
+        <integer>1</integer>
+    </dict>
+</plist>
--- a/authentik/enterprise/endpoints/connectors/fleet/tests/fixtures/host_macos.json
+++ b/authentik/enterprise/endpoints/connectors/fleet/tests/fixtures/host_macos.json
@@ -1,27 +1,27 @@
 {
-    "created_at": "2025-06-25T22:21:35Z",
-    "updated_at": "2025-12-20T11:42:09Z",
+    "created_at": "2026-02-18T16:31:34Z",
+    "updated_at": "2026-03-18T11:29:18Z",
    "software": null,
-    "software_updated_at": "2025-10-22T02:24:25Z",
-    "id": 1,
-    "detail_updated_at": "2025-10-23T23:30:31Z",
-    "label_updated_at": "2025-10-23T23:30:31Z",
-    "policy_updated_at": "2025-10-23T23:02:11Z",
-    "last_enrolled_at": "2025-06-25T22:21:37Z",
-    "seen_time": "2025-10-23T23:59:08Z",
+    "software_updated_at": "2026-03-18T11:29:17Z",
+    "id": 19,
+    "detail_updated_at": "2026-03-18T11:29:18Z",
+    "label_updated_at": "2026-03-18T11:29:18Z",
+    "policy_updated_at": "2026-03-18T11:29:18Z",
+    "last_enrolled_at": "2026-02-18T16:31:45Z",
+    "seen_time": "2026-03-18T11:31:34Z",
    "refetch_requested": false,
    "hostname": "jens-mac-vm.local",
-    "uuid": "C8B98348-A0A6-5838-A321-57B59D788269",
+    "uuid": "5BF422D6-6EAB-5156-AC5A-9EADC9524713",
    "platform": "darwin",
-    "osquery_version": "5.19.0",
+    "osquery_version": "5.21.0",
    "orbit_version": null,
    "fleet_desktop_version": null,
    "scripts_enabled": null,
-    "os_version": "macOS 26.0.1",
-    "build": "25A362",
+    "os_version": "macOS 26.3",
+    "build": "25D125",
    "platform_like": "darwin",
    "code_name": "",
-    "uptime": 256356000000000,
+    "uptime": 653014000000000,
    "memory": 4294967296,
    "cpu_type": "arm64e",
    "cpu_subtype": "ARM64E",
@@ -31,38 +31,41 @@
    "hardware_vendor": "Apple Inc.",
    "hardware_model": "VirtualMac2,1",
    "hardware_version": "",
-    "hardware_serial": "Z5DDF07GK6",
+    "hardware_serial": "ZV35VFDD50",
    "computer_name": "jens-mac-vm",
+    "timezone": null,
    "public_ip": "92.116.179.252",
-    "primary_ip": "192.168.85.3",
-    "primary_mac": "e6:9d:21:c2:2f:19",
+    "primary_ip": "192.168.64.7",
+    "primary_mac": "5e:72:1c:89:98:29",
    "distributed_interval": 10,
    "config_tls_refresh": 60,
    "logger_tls_period": 10,
-    "team_id": 2,
+    "team_id": 5,
    "pack_stats": null,
-    "team_name": "prod",
-    "gigs_disk_space_available": 23.82,
-    "percent_disk_space_available": 37,
+    "team_name": "dev",
+    "gigs_disk_space_available": 16.52,
+    "percent_disk_space_available": 26,
    "gigs_total_disk_space": 62.83,
    "gigs_all_disk_space": null,
    "issues": {
        "failing_policies_count": 1,
-        "critical_vulnerabilities_count": 2,
-        "total_issues_count": 3
+        "critical_vulnerabilities_count": 0,
+        "total_issues_count": 1
    },
    "device_mapping": null,
    "mdm": {
        "enrollment_status": "On (manual)",
        "dep_profile_error": false,
-        "server_url": "https://fleet.beryjuio-home.k8s.beryju.io/mdm/apple/mdm",
+        "server_url": "https://fleet.beryjuio-prod.k8s.beryju.io/mdm/apple/mdm",
        "name": "Fleet",
        "encryption_key_available": false,
        "connected_to_fleet": true
    },
    "refetch_critical_queries_until": null,
-    "last_restarted_at": "2025-10-21T00:17:55Z",
-    "status": "offline",
+    "last_restarted_at": "2026-03-10T22:05:44.00887Z",
+    "status": "online",
    "display_text": "jens-mac-vm.local",
-    "display_name": "jens-mac-vm"
+    "display_name": "jens-mac-vm",
+    "fleet_id": 5,
+    "fleet_name": "dev"
 }
--- a/authentik/enterprise/endpoints/connectors/fleet/tests/test_connector.py
+++ b/authentik/enterprise/endpoints/connectors/fleet/tests/test_connector.py
@@ -21,12 +21,19 @@ TEST_HOST = {"hosts": [TEST_HOST_UBUNTU, TEST_HOST_MACOS, TEST_HOST_WINDOWS, TES
 class TestFleetConnector(APITestCase):
    def setUp(self):
        self.connector = FleetConnector.objects.create(
-            name=generate_id(), url="http://localhost", token=generate_id()
+            name=generate_id(),
+            url="http://localhost",
+            token=generate_id(),
+            map_teams_access_group=True,
        )

    def test_sync(self):
        controller = self.connector.controller(self.connector)
        with Mocker() as mock:
+            mock.get(
+                "http://localhost/api/v1/fleet/conditional_access/idp/apple/profile",
+                text=load_fixture("fixtures/cond_acc_profile.mobileconfig"),
+            )
            mock.get(
                "http://localhost/api/v1/fleet/hosts?order_key=hardware_serial&page=0&per_page=50&device_mapping=true&populate_software=true&populate_users=true",
                json=TEST_HOST,
@@ -40,6 +47,9 @@ class TestFleetConnector(APITestCase):
            identifier="VMware-56 4d 4a 5a b0 22 7b d7-9b a5 0b dc 8f f2 3b 60"
        ).first()
        self.assertIsNotNone(device)
+        group = device.access_group
+        self.assertIsNotNone(group)
+        self.assertEqual(group.name, "prod")
        self.assertEqual(
            device.cached_facts.data,
            {
@@ -50,7 +60,13 @@ class TestFleetConnector(APITestCase):
                    "version": "24.04.3 LTS",
                },
                "disks": [],
-                "vendor": {"fleetdm.com": {"policies": [], "agent_version": ""}},
+                "vendor": {
+                    "fleetdm.com": {
+                        "policies": [],
+                        "agent_version": "",
+                        "uuid": "5a4a4d56-22b0-d77b-9ba5-0bdc8ff23b60",
+                    }
+                },
                "network": {"hostname": "ubuntu-desktop", "interfaces": []},
                "hardware": {
                    "model": "VMware20,1",
@@ -72,6 +88,10 @@ class TestFleetConnector(APITestCase):
        self.connector.save()
        controller = self.connector.controller(self.connector)
        with Mocker() as mock:
+            mock.get(
+                "http://localhost/api/v1/fleet/conditional_access/idp/apple/profile",
+                text=load_fixture("fixtures/cond_acc_profile.mobileconfig"),
+            )
            mock.get(
                "http://localhost/api/v1/fleet/hosts?order_key=hardware_serial&page=0&per_page=50&device_mapping=true&populate_software=true&populate_users=true",
                json=TEST_HOST,
@@ -81,11 +101,13 @@ class TestFleetConnector(APITestCase):
                json={"hosts": []},
            )
            controller.sync_endpoints()
-        self.assertEqual(mock.call_count, 2)
+        self.assertEqual(mock.call_count, 3)
        self.assertEqual(mock.request_history[0].method, "GET")
        self.assertEqual(mock.request_history[0].headers["foo"], "bar")
        self.assertEqual(mock.request_history[1].method, "GET")
        self.assertEqual(mock.request_history[1].headers["foo"], "bar")
+        self.assertEqual(mock.request_history[2].method, "GET")
+        self.assertEqual(mock.request_history[2].headers["foo"], "bar")

    def test_map_host_linux(self):
        controller = self.connector.controller(self.connector)
@@ -128,6 +150,6 @@ class TestFleetConnector(APITestCase):
                "arch": "arm64e",
                "family": OSFamily.macOS,
                "name": "macOS",
-                "version": "26.0.1",
+                "version": "26.3",
            },
        )
--- a/authentik/enterprise/endpoints/connectors/fleet/tests/test_stage.py
+++ b/authentik/enterprise/endpoints/connectors/fleet/tests/test_stage.py
@@ -0,0 +1,84 @@
+from json import loads
+from ssl import PEM_FOOTER, PEM_HEADER
+
+from django.urls import reverse
+from requests_mock import Mocker
+
+from authentik.core.tests.utils import (
+    create_test_flow,
+)
+from authentik.endpoints.models import Device, EndpointStage, StageMode
+from authentik.enterprise.endpoints.connectors.fleet.models import FleetConnector
+from authentik.enterprise.stages.mtls.stage import PLAN_CONTEXT_CERTIFICATE
+from authentik.flows.models import FlowDesignation, FlowStageBinding
+from authentik.flows.planner import PLAN_CONTEXT_DEVICE
+from authentik.flows.tests import FlowTestCase
+from authentik.lib.generators import generate_id
+from authentik.lib.tests.utils import load_fixture
+
+
+class FleetConnectorStageTests(FlowTestCase):
+    def setUp(self):
+        super().setUp()
+        self.connector = FleetConnector.objects.create(
+            name=generate_id(), url="http://localhost", token=generate_id()
+        )
+
+        controller = self.connector.controller(self.connector)
+        with Mocker() as mock:
+            mock.get(
+                "http://localhost/api/v1/fleet/conditional_access/idp/apple/profile",
+                text=load_fixture("fixtures/cond_acc_profile.mobileconfig"),
+            )
+            mock.get(
+                "http://localhost/api/v1/fleet/hosts?order_key=hardware_serial&page=0&per_page=50&device_mapping=true&populate_software=true&populate_users=true",
+                json={"hosts": [loads(load_fixture("fixtures/host_macos.json"))]},
+            )
+            mock.get(
+                "http://localhost/api/v1/fleet/hosts?order_key=hardware_serial&page=1&per_page=50&device_mapping=true&populate_software=true&populate_users=true",
+                json={"hosts": []},
+            )
+            controller.sync_endpoints()
+
+        self.flow = create_test_flow(FlowDesignation.AUTHENTICATION)
+        self.stage = EndpointStage.objects.create(
+            name=generate_id(),
+            mode=StageMode.REQUIRED,
+            connector=self.connector,
+        )
+
+        self.binding = FlowStageBinding.objects.create(target=self.flow, stage=self.stage, order=0)
+
+        self.host_cert = load_fixture("fixtures/cond_acc_host.pem")
+
+    def _format_traefik(self, cert: str | None = None):
+        cert = cert if cert else self.host_cert
+        return cert.replace(PEM_HEADER, "").replace(PEM_FOOTER, "").replace("\n", "")
+
+    def test_assoc(self):
+        dev = Device.objects.get(identifier="ZV35VFDD50")
+        with self.assertFlowFinishes() as plan:
+            res = self.client.get(
+                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
+                headers={"X-Forwarded-TLS-Client-Cert": self._format_traefik()},
+            )
+            self.assertEqual(res.status_code, 200)
+        plan = plan()
+        self.assertEqual(plan.context[PLAN_CONTEXT_DEVICE], dev)
+        self.assertEqual(
+            plan.context[PLAN_CONTEXT_CERTIFICATE]["subject"],
+            "CN=Fleet conditional access for Okta",
+        )
+
+    def test_assoc_not_found(self):
+        dev = Device.objects.get(identifier="ZV35VFDD50")
+        dev.delete()
+        with self.assertFlowFinishes() as plan:
+            res = self.client.get(
+                reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
+                headers={"X-Forwarded-TLS-Client-Cert": self._format_traefik()},
+            )
+            self.assertEqual(res.status_code, 200)
+            self.assertStageResponse(res, self.flow, component="ak-stage-access-denied")
+        plan = plan()
+        self.assertNotIn(PLAN_CONTEXT_DEVICE, plan.context)
--- a/authentik/enterprise/providers/ssf/signals.py
+++ b/authentik/enterprise/providers/ssf/signals.py
@@ -12,7 +12,7 @@ from authentik.core.models import (
    User,
    UserTypes,
 )
-from authentik.core.signals import password_changed
+from authentik.core.signals import password_changed, password_hash_changed
 from authentik.enterprise.providers.ssf.models import (
    EventTypes,
    SSFProvider,
@@ -84,14 +84,13 @@ def ssf_user_session_delete_session_revoked(sender, instance: AuthenticatedSessi
    )


-@receiver(password_changed)
-def ssf_password_changed_cred_change(sender, user: User, password: str | None, **_):
+def _send_password_credential_change(user: User, change_type: str):
    """Credential change trigger (password changed)"""
    send_ssf_events(
        EventTypes.CAEP_CREDENTIAL_CHANGE,
        {
            "credential_type": "password",
-            "change_type": "revoke" if password is None else "update",
+            "change_type": change_type,
        },
        sub_id={
            "format": "complex",
@@ -103,6 +102,16 @@ def ssf_password_changed_cred_change(sender, user: User, password: str | None, *
    )


+@receiver(password_hash_changed)
+@receiver(password_changed)
+def ssf_password_changed_cred_change(signal, sender, user: User, password: str | None = None, **_):
+    """Credential change trigger (password changed)"""
+    if signal is password_hash_changed:
+        _send_password_credential_change(user, "update")
+        return
+    _send_password_credential_change(user, "revoke" if password is None else "update")
+
+
 device_type_map = {
    StaticDevice: "pin",
    TOTPDevice: "pin",
--- a/authentik/enterprise/providers/ssf/tests/test_signals.py
+++ b/authentik/enterprise/providers/ssf/tests/test_signals.py
@@ -1,5 +1,6 @@
 from uuid import uuid4

+from django.contrib.auth.hashers import make_password
 from django.urls import reverse
 from rest_framework.test import APITestCase

@@ -52,6 +53,21 @@ class TestSignals(APITestCase):
        )
        self.assertEqual(res.status_code, 201, res.content)

+    def _assert_password_credential_change(self, user, change_type: str):
+        stream = Stream.objects.filter(provider=self.provider).first()
+        self.assertIsNotNone(stream)
+        event = StreamEvent.objects.filter(stream=stream).first()
+        self.assertIsNotNone(event)
+        self.assertEqual(event.status, SSFEventStatus.PENDING_FAILED)
+        event_payload = event.payload["events"][
+            "https://schemas.openid.net/secevent/caep/event-type/credential-change"
+        ]
+        self.assertEqual(event_payload["change_type"], change_type)
+        self.assertEqual(event_payload["credential_type"], "password")
+        self.assertEqual(event.payload["sub_id"]["format"], "complex")
+        self.assertEqual(event.payload["sub_id"]["user"]["format"], "email")
+        self.assertEqual(event.payload["sub_id"]["user"]["email"], user.email)
+
    def test_signal_logout(self):
        """Test user logout"""
        user = create_test_user()
@@ -79,19 +95,25 @@ class TestSignals(APITestCase):
        user.set_password(generate_id())
        user.save()

-        stream = Stream.objects.filter(provider=self.provider).first()
-        self.assertIsNotNone(stream)
-        event = StreamEvent.objects.filter(stream=stream).first()
-        self.assertIsNotNone(event)
-        self.assertEqual(event.status, SSFEventStatus.PENDING_FAILED)
-        event_payload = event.payload["events"][
-            "https://schemas.openid.net/secevent/caep/event-type/credential-change"
-        ]
-        self.assertEqual(event_payload["change_type"], "update")
-        self.assertEqual(event_payload["credential_type"], "password")
-        self.assertEqual(event.payload["sub_id"]["format"], "complex")
-        self.assertEqual(event.payload["sub_id"]["user"]["format"], "email")
-        self.assertEqual(event.payload["sub_id"]["user"]["email"], user.email)
+        self._assert_password_credential_change(user, "update")
+
+    def test_signal_password_change_from_hash(self):
+        """Test user password change from a pre-hashed password."""
+        user = create_test_user()
+        self.client.force_login(user)
+        user.set_password_from_hash(make_password(generate_id()))
+        user.save()
+
+        self._assert_password_credential_change(user, "update")
+
+    def test_signal_password_revoke(self):
+        """Test explicit password revoke."""
+        user = create_test_user()
+        self.client.force_login(user)
+        user.set_password(None)
+        user.save()
+
+        self._assert_password_credential_change(user, "revoke")

    def test_signal_authenticator_added(self):
        """Test authenticator creation signal"""
--- a/authentik/enterprise/reports/api/reports.py
+++ b/authentik/enterprise/reports/api/reports.py
@@ -4,13 +4,13 @@ from django.urls import reverse
 from drf_spectacular.utils import extend_schema
 from rest_framework import mixins
 from rest_framework.decorators import action
+from rest_framework.fields import CharField, SerializerMethodField
 from rest_framework.permissions import BasePermission
 from rest_framework.request import Request
 from rest_framework.response import Response
 from rest_framework.viewsets import GenericViewSet

 from authentik.core.api.groups import PartialUserSerializer
-from authentik.core.api.object_attributes import ContentTypeSerializer
 from authentik.core.api.utils import ModelSerializer
 from authentik.enterprise.api import EnterpriseRequiredMixin
 from authentik.enterprise.reports.models import DataExport
@@ -18,6 +18,19 @@ from authentik.enterprise.reports.tasks import generate_export
 from authentik.rbac.permissions import HasPermission


+class ContentTypeSerializer(ModelSerializer):
+    app_label = CharField(read_only=True)
+    model = CharField(read_only=True)
+    verbose_name_plural = SerializerMethodField()
+
+    def get_verbose_name_plural(self, ct: ContentType) -> str:
+        return ct.model_class()._meta.verbose_name_plural
+
+    class Meta:
+        model = ContentType
+        fields = ("id", "app_label", "model", "verbose_name_plural")
+
+
 class DataExportSerializer(EnterpriseRequiredMixin, ModelSerializer):
    requested_by = PartialUserSerializer(read_only=True)
    content_type = ContentTypeSerializer(read_only=True)
--- a/authentik/enterprise/stages/mtls/stage.py
+++ b/authentik/enterprise/stages/mtls/stage.py
@@ -15,6 +15,7 @@ from cryptography.x509 import (
 )
 from cryptography.x509.verification import PolicyBuilder, Store, VerificationError
 from django.utils.translation import gettext_lazy as _
+from rest_framework.exceptions import PermissionDenied

 from authentik.brands.models import Brand
 from authentik.core.models import User
@@ -25,7 +26,6 @@ from authentik.enterprise.stages.mtls.models import (
    MutualTLSStage,
    UserAttributes,
 )
-from authentik.flows.challenge import AccessDeniedChallenge
 from authentik.flows.models import FlowDesignation
 from authentik.flows.planner import PLAN_CONTEXT_PENDING_USER
 from authentik.flows.stage import ChallengeStageView
@@ -217,8 +217,7 @@ class MTLSStageView(ChallengeStageView):
            return None
        return str(_cert_attr[0])

-    def dispatch(self, request, *args, **kwargs):
-        stage: MutualTLSStage = self.executor.current_stage
+    def get_cert(self, mode: StageMode):
        certs = [
            *self._parse_cert_xfcc(),
            *self._parse_cert_nginx(),
@@ -228,21 +227,26 @@ class MTLSStageView(ChallengeStageView):
        authorities = self.get_authorities()
        if not authorities:
            self.logger.warning("No Certificate authority found")
-            if stage.mode == StageMode.OPTIONAL:
-                return self.executor.stage_ok()
-            if stage.mode == StageMode.REQUIRED:
-                return super().dispatch(request, *args, **kwargs)
+            if mode == StageMode.OPTIONAL:
+                return None
+            if mode == StageMode.REQUIRED:
+                raise PermissionDenied("Unknown error")
        cert = self.validate_cert(authorities, certs)
-        if not cert and stage.mode == StageMode.REQUIRED:
+        if not cert and mode == StageMode.REQUIRED:
            self.logger.warning("Client certificate required but no certificates given")
-            return super().dispatch(
-                request,
-                *args,
-                error_message=_("Certificate required but no certificate was given."),
-                **kwargs,
-            )
-        if not cert and stage.mode == StageMode.OPTIONAL:
+            raise PermissionDenied(str(_("Certificate required but no certificate was given.")))
+        if not cert and mode == StageMode.OPTIONAL:
            self.logger.info("No certificate given, continuing")
+            return None
+        return cert
+
+    def dispatch(self, request, *args, **kwargs):
+        stage: MutualTLSStage = self.executor.current_stage
+        try:
+            cert = self.get_cert(stage.mode)
+        except PermissionDenied as exc:
+            return self.executor.stage_invalid(error_message=exc.detail)
+        if not cert:
            return self.executor.stage_ok()
        self.logger.debug("Received certificate", cert=fingerprint_sha256(cert))
        existing_user = self.check_if_user(cert)
@@ -251,15 +255,5 @@ class MTLSStageView(ChallengeStageView):
        elif existing_user:
            self.auth_user(existing_user, cert)
        else:
-            return super().dispatch(
-                request, *args, error_message=_("No user found for certificate."), **kwargs
-            )
+            return self.executor.stage_invalid(_("No user found for certificate."))
        return self.executor.stage_ok()
-
-    def get_challenge(self, *args, error_message: str | None = None, **kwargs):
-        return AccessDeniedChallenge(
-            data={
-                "component": "ak-stage-access-denied",
-                "error_message": str(error_message or "Unknown error"),
-            }
-        )
--- a/authentik/events/signals.py
+++ b/authentik/events/signals.py
@@ -11,7 +11,7 @@ from django.http import HttpRequest
 from rest_framework.request import Request

 from authentik.core.models import AuthenticatedSession, User
-from authentik.core.signals import login_failed, password_changed
+from authentik.core.signals import login_failed, password_changed, password_hash_changed
 from authentik.events.models import Event, EventAction
 from authentik.flows.models import Stage
 from authentik.flows.planner import (
@@ -112,8 +112,15 @@ def on_invitation_used(sender, request: HttpRequest, invitation: Invitation, **_
    )


+@receiver(password_hash_changed)
@receiver(password_changed)
-def on_password_changed(sender, user: User, password: str, request: HttpRequest | None, **_):
+def on_password_changed(
+    sender,
+    user: User,
+    password: str | None = None,
+    request: HttpRequest | None = None,
+    **_,
+):
    """Log password change"""
    Event.new(EventAction.PASSWORD_SET).from_http(request, user=user)

--- a/authentik/events/tests/test_event.py
+++ b/authentik/events/tests/test_event.py
@@ -2,6 +2,7 @@

 from urllib.parse import urlencode

+from django.contrib.auth.hashers import make_password
 from django.contrib.contenttypes.models import ContentType
 from django.test import RequestFactory, TestCase
 from django.views.debug import SafeExceptionReporterFilter
@@ -10,7 +11,7 @@ from guardian.shortcuts import get_anonymous_user
 from authentik.brands.models import Brand
 from authentik.core.models import Group, User
 from authentik.core.tests.utils import create_test_user
-from authentik.events.models import Event
+from authentik.events.models import Event, EventAction
 from authentik.flows.planner import PLAN_CONTEXT_PENDING_USER, FlowPlan
 from authentik.flows.views.executor import QS_QUERY, SESSION_KEY_PLAN
 from authentik.lib.generators import generate_id
@@ -213,3 +214,14 @@ class TestEvents(TestCase):
        event = Event.new("unittest", foo="foo bar \u0000 baz")
        event.save()
        self.assertEqual(event.context["foo"], "foo bar  baz")
+
+    def test_password_set_signal_on_set_password_from_hash(self):
+        """Changing password from hash should still emit an audit event."""
+        user = create_test_user()
+        old_count = Event.objects.filter(action=EventAction.PASSWORD_SET, user__pk=user.pk).count()
+
+        user.set_password_from_hash(make_password(generate_id()))
+        user.save()
+
+        new_count = Event.objects.filter(action=EventAction.PASSWORD_SET, user__pk=user.pk).count()
+        self.assertEqual(new_count, old_count + 1)
--- a/authentik/flows/exceptions.py
+++ b/authentik/flows/exceptions.py
@@ -11,6 +11,10 @@ class FlowNonApplicableException(SentryIgnoredException):

    policy_result: PolicyResult | None = None

+    def __init__(self, policy_result: PolicyResult | None = None, *args):
+        super().__init__(*args)
+        self.policy_result = policy_result
+
    @property
    def messages(self) -> str:
        """Get messages from policy result, fallback to generic reason"""
--- a/authentik/flows/migrations/0027_auto_20231028_1424.py
+++ b/authentik/flows/migrations/0027_auto_20231028_1424.py
@@ -42,6 +42,7 @@ class Migration(migrations.Migration):
                    ("require_superuser", "Require Superuser"),
                    ("require_redirect", "Require Redirect"),
                    ("require_outpost", "Require Outpost"),
+                    ("require_token", "Require Token"),
                ],
                default="none",
                help_text="Required level of authentication and authorization to access a flow.",
--- a/authentik/flows/models.py
+++ b/authentik/flows/models.py
@@ -40,6 +40,7 @@ class FlowAuthenticationRequirement(models.TextChoices):
    REQUIRE_SUPERUSER = "require_superuser"
    REQUIRE_REDIRECT = "require_redirect"
    REQUIRE_OUTPOST = "require_outpost"
+    REQUIRE_TOKEN = "require_token"


 class NotConfiguredAction(models.TextChoices):
--- a/authentik/flows/planner.py
+++ b/authentik/flows/planner.py
@@ -5,6 +5,7 @@ from typing import TYPE_CHECKING, Any

 from django.core.cache import cache
 from django.http import HttpRequest, HttpResponse
+from django.utils.translation import gettext as _
 from sentry_sdk import start_span
 from sentry_sdk.tracing import Span
 from structlog.stdlib import BoundLogger, get_logger
@@ -26,6 +27,7 @@ from authentik.lib.config import CONFIG
 from authentik.lib.utils.urls import redirect_with_qs
 from authentik.outposts.models import Outpost
 from authentik.policies.engine import PolicyEngine
+from authentik.policies.types import PolicyResult
 from authentik.root.middleware import ClientIPMiddleware

 if TYPE_CHECKING:
@@ -226,6 +228,15 @@ class FlowPlanner:
            and context.get(PLAN_CONTEXT_IS_REDIRECTED) is None
        ):
            raise FlowNonApplicableException()
+        if (
+            self.flow.authentication == FlowAuthenticationRequirement.REQUIRE_TOKEN
+            and context.get(PLAN_CONTEXT_IS_RESTORED) is None
+        ):
+            raise FlowNonApplicableException(
+                PolicyResult(
+                    False, _("This link is invalid or has expired. Please request a new one.")
+                )
+            )
        outpost_user = ClientIPMiddleware.get_outpost_user(request)
        if self.flow.authentication == FlowAuthenticationRequirement.REQUIRE_OUTPOST:
            if not outpost_user:
@@ -273,9 +284,7 @@ class FlowPlanner:
            engine.build()
            result = engine.result
            if not result.passing:
-                exc = FlowNonApplicableException()
-                exc.policy_result = result
-                raise exc
+                raise FlowNonApplicableException(result)
            # User is passing so far, check if we have a cached plan
            cached_plan_key = cache_key(self.flow, user)
            cached_plan = cache.get(cached_plan_key, None)
--- a/authentik/flows/templates/if/flow-sfe.html
+++ b/authentik/flows/templates/if/flow-sfe.html
@@ -23,7 +23,7 @@
            height: 100%;
          }
          body {
-            background-image: url("{{ flow_background_url }}");
+            background-image: url("{{ flow_background_url|iriencode|safe }}");
            background-repeat: no-repeat;
            background-size: cover;
          }
--- a/authentik/flows/templates/if/flow.html
+++ b/authentik/flows/templates/if/flow.html
@@ -39,7 +39,7 @@
 <script src="{% versioned_script 'dist/flow/FlowInterface-%v.js' %}" type="module"></script>
 <style data-id="flow-css">
  :root {
-      --ak-global--background-image: url("{{ flow_background_url }}");
+      --ak-global--background-image: url("{{ flow_background_url|iriencode|safe }}");
  }
 </style>
 {% endblock %}
--- a/authentik/flows/tests/test_executor.py
+++ b/authentik/flows/tests/test_executor.py
@@ -1,5 +1,6 @@
 """flow views tests"""

+from datetime import timedelta
 from unittest.mock import MagicMock, PropertyMock, patch
 from urllib.parse import urlencode

@@ -7,6 +8,7 @@ from django.http import HttpRequest, HttpResponse
 from django.test import override_settings
 from django.test.client import RequestFactory
 from django.urls import reverse
+from django.utils.timezone import now
 from rest_framework.exceptions import ParseError

 from authentik.core.models import Group, User
@@ -17,6 +19,7 @@ from authentik.flows.models import (
    FlowDeniedAction,
    FlowDesignation,
    FlowStageBinding,
+    FlowToken,
    InvalidResponseAction,
 )
 from authentik.flows.planner import FlowPlan, FlowPlanner
@@ -24,6 +27,7 @@ from authentik.flows.stage import PLAN_CONTEXT_PENDING_USER_IDENTIFIER, StageVie
 from authentik.flows.tests import FlowTestCase
 from authentik.flows.views.executor import (
    NEXT_ARG_NAME,
+    QS_KEY_TOKEN,
    QS_QUERY,
    SESSION_KEY_PLAN,
    FlowExecutorView,
@@ -740,3 +744,77 @@ class TestFlowExecutor(FlowTestCase):
                "title": flow.title,
            },
        )
+
+    @patch(
+        "authentik.flows.views.executor.to_stage_response",
+        TO_STAGE_RESPONSE_MOCK,
+    )
+    def test_expired_flow_token(self):
+        """Test that an expired flow token shows an appropriate error message"""
+        flow = create_test_flow(
+            FlowDesignation.RECOVERY,
+            authentication=FlowAuthenticationRequirement.REQUIRE_TOKEN,
+        )
+        user = create_test_user()
+        plan = FlowPlan(flow_pk=flow.pk.hex, bindings=[], markers=[])
+
+        token = FlowToken.objects.create(
+            user=user,
+            identifier=generate_id(),
+            flow=flow,
+            _plan=FlowToken.pickle(plan),
+            expires=now() - timedelta(hours=1),
+        )
+
+        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug})
+        response = self.client.get(
+            url + f"?{urlencode({QS_QUERY: urlencode({QS_KEY_TOKEN: token.key})})}"
+        )
+        self.assertStageResponse(
+            response,
+            flow,
+            component="ak-stage-access-denied",
+            error_message="This link is invalid or has expired. Please request a new one.",
+        )
+
+    @patch(
+        "authentik.flows.views.executor.to_stage_response",
+        TO_STAGE_RESPONSE_MOCK,
+    )
+    def test_invalid_flow_token_require_token(self):
+        """Test that an invalid/nonexistent token on a REQUIRE_TOKEN flow shows error"""
+        flow = create_test_flow(
+            FlowDesignation.RECOVERY,
+            authentication=FlowAuthenticationRequirement.REQUIRE_TOKEN,
+        )
+
+        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug})
+        response = self.client.get(
+            url + f"?{urlencode({QS_QUERY: urlencode({QS_KEY_TOKEN: 'invalid-token'})})}"
+        )
+        self.assertStageResponse(
+            response,
+            flow,
+            component="ak-stage-access-denied",
+            error_message="This link is invalid or has expired. Please request a new one.",
+        )
+
+    @patch(
+        "authentik.flows.views.executor.to_stage_response",
+        TO_STAGE_RESPONSE_MOCK,
+    )
+    def test_no_token_require_token(self):
+        """Test that accessing a REQUIRE_TOKEN flow without any token shows error"""
+        flow = create_test_flow(
+            FlowDesignation.RECOVERY,
+            authentication=FlowAuthenticationRequirement.REQUIRE_TOKEN,
+        )
+
+        url = reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug})
+        response = self.client.get(url)
+        self.assertStageResponse(
+            response,
+            flow,
+            component="ak-stage-access-denied",
+            error_message="This link is invalid or has expired. Please request a new one.",
+        )
--- a/authentik/flows/tests/test_planner.py
+++ b/authentik/flows/tests/test_planner.py
@@ -26,6 +26,7 @@ from authentik.flows.models import (
 )
 from authentik.flows.planner import (
    PLAN_CONTEXT_IS_REDIRECTED,
+    PLAN_CONTEXT_IS_RESTORED,
    PLAN_CONTEXT_PENDING_USER,
    FlowPlanner,
    cache_key,
@@ -129,6 +130,22 @@ class TestFlowPlanner(TestCase):
        planner.allow_empty_flows = True
        planner.plan(request)

+    def test_authentication_require_token(self):
+        """Test flow authentication (require_token)"""
+        flow = create_test_flow()
+        flow.authentication = FlowAuthenticationRequirement.REQUIRE_TOKEN
+        request = self.request_factory.get(
+            reverse("authentik_api:flow-executor", kwargs={"flow_slug": flow.slug}),
+        )
+        planner = FlowPlanner(flow)
+        planner.allow_empty_flows = True
+
+        with self.assertRaises(FlowNonApplicableException):
+            planner.plan(request)
+
+        context = {PLAN_CONTEXT_IS_RESTORED: True}
+        planner.plan(request, context)
+
    @patch(
        "authentik.policies.engine.PolicyEngine.result",
        POLICY_RETURN_FALSE,
--- a/authentik/flows/tests/test_stage_views.py
+++ b/authentik/flows/tests/test_stage_views.py
@@ -1,12 +1,14 @@
 """stage view tests"""

 from collections.abc import Callable
+from unittest.mock import patch

 from django.test import RequestFactory, TestCase
+from django.urls import reverse

 from authentik.core.tests.utils import RequestFactory as AuthentikRequestFactory
 from authentik.core.tests.utils import create_test_flow
-from authentik.flows.models import FlowStageBinding
+from authentik.flows.models import Flow, FlowStageBinding
 from authentik.flows.stage import StageView
 from authentik.flows.views.executor import FlowExecutorView
 from authentik.lib.utils.reflection import all_subclasses
@@ -42,6 +44,46 @@ class TestViews(TestCase):
            "/static/dist/assets/images/flow_background.jpg",
        )

+    def test_flow_interface_css_background_preserves_presigned_url_query(self):
+        """Test flow CSS keeps signed URL query separators intact."""
+        flow = create_test_flow()
+        background_url = (
+            "https://s3.ca-central-1.amazonaws.com/example/media/public/background.png"
+            "?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=credential"
+            "&X-Amz-Signature=signature"
+        )
+
+        with patch.object(Flow, "background_url", return_value=background_url):
+            response = self.client.get(
+                reverse("authentik_core:if-flow", kwargs={"flow_slug": flow.slug})
+            )
+
+        self.assertContains(
+            response,
+            f'--ak-global--background-image: url("{background_url}");',
+            html=False,
+        )
+
+    def test_flow_sfe_css_background_preserves_presigned_url_query(self):
+        """Test SFE flow CSS keeps signed URL query separators intact."""
+        flow = create_test_flow()
+        background_url = (
+            "https://s3.ca-central-1.amazonaws.com/example/media/public/background.png"
+            "?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=credential"
+            "&X-Amz-Signature=signature"
+        )
+
+        with patch.object(Flow, "background_url", return_value=background_url):
+            response = self.client.get(
+                reverse("authentik_core:if-flow", kwargs={"flow_slug": flow.slug}) + "?sfe"
+            )
+
+        self.assertContains(
+            response,
+            f'background-image: url("{background_url}");',
+            html=False,
+        )
+

 def view_tester_factory(view_class: type[StageView]) -> Callable:
    """Test a form"""
--- a/authentik/flows/views/executor.py
+++ b/authentik/flows/views/executor.py
@@ -62,6 +62,7 @@ from authentik.policies.engine import PolicyEngine
 LOGGER = get_logger()
 # Argument used to redirect user after login
 NEXT_ARG_NAME = "next"
+
 SESSION_KEY_PLAN = "authentik/flows/plan"
 SESSION_KEY_GET = "authentik/flows/get"
 SESSION_KEY_POST = "authentik/flows/post"
--- a/authentik/flows/views/inspector.py
+++ b/authentik/flows/views/inspector.py
@@ -71,7 +71,11 @@ class FlowInspectorView(APIView):

    flow: Flow
    _logger: BoundLogger
-    permission_classes = [IsAuthenticated]
+
+    def get_permissions(self):
+        if settings.DEBUG:
+            return []
+        return [IsAuthenticated()]

    def setup(self, request: HttpRequest, flow_slug: str):
        super().setup(request, flow_slug=flow_slug)
--- a/authentik/lib/expression/evaluator.py
+++ b/authentik/lib/expression/evaluator.py
@@ -10,7 +10,6 @@ from typing import TYPE_CHECKING, Any

 from cachetools import TLRUCache, cached
 from django.core.exceptions import FieldError
-from django.db.models import Model
 from django.http import HttpRequest
 from django.utils.text import slugify
 from django.utils.timezone import now
@@ -23,7 +22,6 @@ from structlog.stdlib import get_logger
 from authentik.core.models import User
 from authentik.events.models import Event
 from authentik.lib.expression.exceptions import ControlFlowException
-from authentik.lib.utils.dict import get_path_from_dict
 from authentik.lib.utils.email import normalize_addresses
 from authentik.lib.utils.http import get_http_session
 from authentik.lib.utils.time import timedelta_from_string
@@ -72,7 +70,6 @@ class BaseEvaluator:
            "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_obj_attr": BaseEvaluator.expr_obj_attr,
            "ip_address": ip_address,
            "ip_network": ip_network,
            "list_flatten": BaseEvaluator.expr_flatten,
@@ -165,16 +162,6 @@ class BaseEvaluator:
            return False
        return len(list(user_devices)) > 0

-    @staticmethod
-    def expr_obj_attr(obj: Model, attr_key: str, fallback: str) -> Any:
-        """Get an attribute of the given object if set by its dotted path, otherwise
-        return fallback value."""
-        attrs = getattr(obj, "attributes", {})
-        value = get_path_from_dict(attrs, attr_key)
-        if value is None and fallback:
-            return getattr(obj, fallback)
-        return value
-
    def expr_event_create(self, action: str, **kwargs):
        """Create event with supplied data and try to extract as much relevant data
        from the context"""
--- a/authentik/lib/utils/db.py
+++ b/authentik/lib/utils/db.py
@@ -14,7 +14,16 @@ def chunked_queryset[T: Model](queryset: QuerySet[T], chunk_size: int = 1_000) -
    def get_chunks(qs: QuerySet) -> Generator[QuerySet[T]]:
        qs = qs.order_by("pk")
        pks = qs.values_list("pk", flat=True)
-        start_pk = pks[0]
+        # The outer queryset.exists() guard can race with a concurrent
+        # transaction that deletes the last matching row (or with a
+        # different isolation-level snapshot), so by the time this
+        # generator starts iterating the queryset may be empty and
+        # pks[0] would raise IndexError and crash the caller. Using
+        # .first() returns None on an empty queryset, which we bail
+        # out on cleanly. See goauthentik/authentik#21643.
+        start_pk = pks.first()
+        if start_pk is None:
+            return
        while True:
            try:
                end_pk = pks.filter(pk__gte=start_pk)[chunk_size]
--- a/authentik/providers/oauth2/api/providers.py
+++ b/authentik/providers/oauth2/api/providers.py
@@ -65,6 +65,7 @@ class OAuth2ProviderSerializer(ProviderSerializer):
        fields = ProviderSerializer.Meta.fields + [
            "authorization_flow",
            "client_type",
+            "grant_types",
            "client_id",
            "client_secret",
            "access_code_validity",
--- a/authentik/providers/oauth2/errors.py
+++ b/authentik/providers/oauth2/errors.py
@@ -7,7 +7,7 @@ from django.http import HttpRequest, HttpResponse, HttpResponseRedirect
 from authentik.events.models import Event, EventAction
 from authentik.lib.sentry import SentryIgnoredException
 from authentik.lib.views import bad_request_message
-from authentik.providers.oauth2.models import GrantTypes, RedirectURI
+from authentik.providers.oauth2.models import GrantType, RedirectURI


 class OAuth2Error(SentryIgnoredException):
@@ -182,7 +182,7 @@ class AuthorizeError(OAuth2Error):
        # See:
        # http://openid.net/specs/openid-connect-core-1_0.html#ImplicitAuthError
        fragment_or_query = (
-            "#" if self.grant_type in [GrantTypes.IMPLICIT, GrantTypes.HYBRID] else "?"
+            "#" if self.grant_type in [GrantType.IMPLICIT, GrantType.HYBRID] else "?"
        )

        uri = (
@@ -225,7 +225,7 @@ class TokenError(OAuth2Error):
        ),
    }

-    def __init__(self, error):
+    def __init__(self, error: str):
        super().__init__()
        self.error = error
        self.description = self.errors[error]
--- a/authentik/providers/oauth2/migrations/0032_oauth2provider_grant_types.py
+++ b/authentik/providers/oauth2/migrations/0032_oauth2provider_grant_types.py
@@ -0,0 +1,69 @@
+# Generated by Django 5.2.11 on 2026-02-17 11:04
+
+import django.contrib.postgres.fields
+from django.db import migrations, models
+
+
+def migrate_default_grant_types():
+    from authentik.providers.oauth2.models import GrantType
+
+    return [
+        GrantType.AUTHORIZATION_CODE,
+        GrantType.HYBRID,
+        GrantType.IMPLICIT,
+        GrantType.CLIENT_CREDENTIALS,
+        GrantType.PASSWORD,
+        GrantType.DEVICE_CODE,
+        GrantType.REFRESH_TOKEN,
+    ]
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        (
+            "authentik_providers_oauth2",
+            "0031_remove_oauth2provider_backchannel_logout_uri_and_more",
+        ),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="oauth2provider",
+            name="grant_types",
+            field=django.contrib.postgres.fields.ArrayField(
+                base_field=models.TextField(
+                    choices=[
+                        ("authorization_code", "Authorization Code"),
+                        ("implicit", "Implicit"),
+                        ("hybrid", "Hybrid"),
+                        ("refresh_token", "Refresh Token"),
+                        ("client_credentials", "Client Credentials"),
+                        ("password", "Password"),
+                        ("urn:ietf:params:oauth:grant-type:device_code", "Device Code"),
+                    ]
+                ),
+                default=migrate_default_grant_types,
+                size=None,
+            ),
+        ),
+        migrations.AlterField(
+            model_name="oauth2provider",
+            name="grant_types",
+            field=django.contrib.postgres.fields.ArrayField(
+                base_field=models.TextField(
+                    choices=[
+                        ("authorization_code", "Authorization Code"),
+                        ("implicit", "Implicit"),
+                        ("hybrid", "Hybrid"),
+                        ("refresh_token", "Refresh Token"),
+                        ("client_credentials", "Client Credentials"),
+                        ("password", "Password"),
+                        ("urn:ietf:params:oauth:grant-type:device_code", "Device Code"),
+                    ]
+                ),
+                default=list,
+                size=None,
+            ),
+        ),
+    ]
--- a/authentik/providers/oauth2/models.py
+++ b/authentik/providers/oauth2/models.py
@@ -19,6 +19,7 @@ from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey
 from cryptography.hazmat.primitives.asymmetric.types import PrivateKeyTypes
 from dacite import Config
 from dacite.core import from_dict
+from django.contrib.postgres.fields import ArrayField
 from django.contrib.postgres.indexes import HashIndex
 from django.db import models
 from django.http import HttpRequest
@@ -33,7 +34,16 @@ from rest_framework.serializers import Serializer
 from structlog.stdlib import get_logger

 from authentik.brands.models import WebfingerProvider
-from authentik.common.oauth.constants import SubModes
+from authentik.common.oauth.constants import (
+    GRANT_TYPE_AUTHORIZATION_CODE,
+    GRANT_TYPE_CLIENT_CREDENTIALS,
+    GRANT_TYPE_DEVICE_CODE,
+    GRANT_TYPE_HYBRID,
+    GRANT_TYPE_IMPLICIT,
+    GRANT_TYPE_PASSWORD,
+    GRANT_TYPE_REFRESH_TOKEN,
+    SubModes,
+)
 from authentik.core.models import (
    AuthenticatedSession,
    ExpiringModel,
@@ -58,7 +68,7 @@ def generate_client_secret() -> str:
    return generate_id(128)


-class ClientTypes(models.TextChoices):
+class ClientType(models.TextChoices):
    """Confidential clients are capable of maintaining the confidentiality
    of their credentials. Public clients are incapable."""

@@ -66,12 +76,16 @@ class ClientTypes(models.TextChoices):
    PUBLIC = "public", _("Public")


-class GrantTypes(models.TextChoices):
+class GrantType(models.TextChoices):
    """OAuth2 Grant types we support"""

-    AUTHORIZATION_CODE = "authorization_code"
-    IMPLICIT = "implicit"
-    HYBRID = "hybrid"
+    AUTHORIZATION_CODE = GRANT_TYPE_AUTHORIZATION_CODE
+    IMPLICIT = GRANT_TYPE_IMPLICIT
+    HYBRID = GRANT_TYPE_HYBRID
+    REFRESH_TOKEN = GRANT_TYPE_REFRESH_TOKEN
+    CLIENT_CREDENTIALS = GRANT_TYPE_CLIENT_CREDENTIALS
+    PASSWORD = GRANT_TYPE_PASSWORD
+    DEVICE_CODE = GRANT_TYPE_DEVICE_CODE


 class ResponseMode(models.TextChoices):
@@ -188,14 +202,15 @@ class OAuth2Provider(WebfingerProvider, Provider):

    client_type = models.CharField(
        max_length=30,
-        choices=ClientTypes.choices,
-        default=ClientTypes.CONFIDENTIAL,
+        choices=ClientType.choices,
+        default=ClientType.CONFIDENTIAL,
        verbose_name=_("Client Type"),
        help_text=_(
            "Confidential clients are capable of maintaining the confidentiality "
            "of their credentials. Public clients are incapable"
        ),
    )
+    grant_types = ArrayField(models.TextField(choices=GrantType.choices), default=list)
    client_id = models.CharField(
        max_length=255,
        unique=True,
--- a/authentik/providers/oauth2/tests/test_authorize.py
+++ b/authentik/providers/oauth2/tests/test_authorize.py
@@ -22,7 +22,7 @@ from authentik.providers.oauth2.errors import AuthorizeError, ClientIdError, Red
 from authentik.providers.oauth2.models import (
    AccessToken,
    AuthorizationCode,
-    GrantTypes,
+    GrantType,
    OAuth2Provider,
    RedirectURI,
    RedirectURIMatchingMode,
@@ -41,12 +41,34 @@ class TestAuthorize(OAuthTestCase):
        super().setUp()
        self.factory = RequestFactory()

+    def test_disallowed_grant_type(self):
+        """Test with disallowed grant type"""
+        OAuth2Provider.objects.create(
+            name=generate_id(),
+            client_id="test",
+            grant_types=[],
+            authorization_flow=create_test_flow(),
+            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://local.invalid/Foo")],
+        )
+        with self.assertRaises(AuthorizeError) as cm:
+            request = self.factory.get(
+                "/",
+                data={
+                    "response_type": "code",
+                    "client_id": "test",
+                    "redirect_uri": "http://local.invalid/Foo",
+                },
+            )
+            OAuthAuthorizationParams.from_request(request)
+        self.assertEqual(cm.exception.error, "invalid_request")
+
    def test_invalid_grant_type(self):
        """Test with invalid grant type"""
        OAuth2Provider.objects.create(
            name=generate_id(),
            client_id="test",
            authorization_flow=create_test_flow(),
+            grant_types=[GrantType.AUTHORIZATION_CODE],
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://local.invalid/Foo")],
        )
        with self.assertRaises(AuthorizeError) as cm:
@@ -74,6 +96,7 @@ class TestAuthorize(OAuthTestCase):
            client_id="test",
            authorization_flow=create_test_flow(),
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://local.invalid/Foo")],
+            grant_types=[GrantType.AUTHORIZATION_CODE],
        )
        with self.assertRaises(AuthorizeError) as cm:
            request = self.factory.get(
@@ -141,26 +164,6 @@ class TestAuthorize(OAuthTestCase):
            OAuthAuthorizationParams.from_request(request)
        self.assertEqual(cm.exception.cause, "redirect_uri_forbidden_scheme")

-    def test_invalid_redirect_uri_empty(self):
-        """test missing/invalid redirect URI"""
-        provider = OAuth2Provider.objects.create(
-            name=generate_id(),
-            client_id="test",
-            authorization_flow=create_test_flow(),
-            redirect_uris=[],
-        )
-        request = self.factory.get(
-            "/",
-            data={
-                "response_type": "code",
-                "client_id": "test",
-                "redirect_uri": "+",
-            },
-        )
-        OAuthAuthorizationParams.from_request(request)
-        provider.refresh_from_db()
-        self.assertEqual(provider.redirect_uris, [RedirectURI(RedirectURIMatchingMode.STRICT, "+")])
-
    def test_invalid_redirect_uri_regex(self):
        """test missing/invalid redirect URI"""
        OAuth2Provider.objects.create(
@@ -208,6 +211,7 @@ class TestAuthorize(OAuthTestCase):
            client_id="test",
            authorization_flow=create_test_flow(),
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.REGEX, ".+")],
+            grant_types=[GrantType.AUTHORIZATION_CODE],
        )
        request = self.factory.get(
            "/",
@@ -226,6 +230,7 @@ class TestAuthorize(OAuthTestCase):
            name=generate_id(),
            client_id="test",
            authorization_flow=create_test_flow(),
+            grant_types=[GrantType.AUTHORIZATION_CODE],
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://local.invalid/Foo")],
        )
        provider.property_mappings.set(
@@ -247,12 +252,14 @@ class TestAuthorize(OAuthTestCase):
        )
        self.assertEqual(
            OAuthAuthorizationParams.from_request(request).grant_type,
-            GrantTypes.AUTHORIZATION_CODE,
+            GrantType.AUTHORIZATION_CODE,
        )
        self.assertEqual(
            OAuthAuthorizationParams.from_request(request).redirect_uri,
            "http://local.invalid/Foo",
        )
+        provider.grant_types = [GrantType.IMPLICIT]
+        provider.save()
        request = self.factory.get(
            "/",
            data={
@@ -266,7 +273,7 @@ class TestAuthorize(OAuthTestCase):
        )
        self.assertEqual(
            OAuthAuthorizationParams.from_request(request).grant_type,
-            GrantTypes.IMPLICIT,
+            GrantType.IMPLICIT,
        )
        # Implicit without openid scope
        with self.assertRaises(AuthorizeError) as cm:
@@ -281,8 +288,10 @@ class TestAuthorize(OAuthTestCase):
            )
            self.assertEqual(
                OAuthAuthorizationParams.from_request(request).grant_type,
-                GrantTypes.IMPLICIT,
+                GrantType.IMPLICIT,
            )
+        provider.grant_types = [GrantType.HYBRID]
+        provider.save()
        request = self.factory.get(
            "/",
            data={
@@ -294,7 +303,7 @@ class TestAuthorize(OAuthTestCase):
            },
        )
        self.assertEqual(
-            OAuthAuthorizationParams.from_request(request).grant_type, GrantTypes.HYBRID
+            OAuthAuthorizationParams.from_request(request).grant_type, GrantType.HYBRID
        )
        with self.assertRaises(AuthorizeError) as cm:
            request = self.factory.get(
@@ -317,6 +326,7 @@ class TestAuthorize(OAuthTestCase):
            authorization_flow=flow,
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "foo://localhost")],
            access_code_validity="seconds=100",
+            grant_types=[GrantType.AUTHORIZATION_CODE],
        )
        Application.objects.create(name="app", slug="app", provider=provider)
        state = generate_id()
@@ -353,6 +363,7 @@ class TestAuthorize(OAuthTestCase):
            authorization_flow=flow,
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://localhost")],
            signing_key=self.keypair,
+            grant_types=[GrantType.IMPLICIT],
        )
        provider.property_mappings.set(
            ScopeMapping.objects.filter(
@@ -424,6 +435,7 @@ class TestAuthorize(OAuthTestCase):
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://localhost")],
            signing_key=self.keypair,
            encryption_key=self.keypair,
+            grant_types=[GrantType.IMPLICIT],
        )
        provider.property_mappings.set(
            ScopeMapping.objects.filter(
@@ -486,6 +498,7 @@ class TestAuthorize(OAuthTestCase):
            authorization_flow=flow,
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://localhost")],
            signing_key=self.keypair,
+            grant_types=[GrantType.AUTHORIZATION_CODE],
        )
        Application.objects.create(name="app", slug="app", provider=provider)
        state = generate_id()
@@ -535,6 +548,7 @@ class TestAuthorize(OAuthTestCase):
            authorization_flow=flow,
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://localhost")],
            signing_key=self.keypair,
+            grant_types=[GrantType.IMPLICIT],
        )
        provider.property_mappings.set(
            ScopeMapping.objects.filter(
@@ -592,6 +606,7 @@ class TestAuthorize(OAuthTestCase):
            authorization_flow=flow,
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://localhost")],
            signing_key=self.keypair,
+            grant_types=[GrantType.AUTHORIZATION_CODE],
        )
        app = Application.objects.create(name=generate_id(), slug=generate_id(), provider=provider)
        state = generate_id()
@@ -632,6 +647,7 @@ class TestAuthorize(OAuthTestCase):
            name=generate_id(),
            client_id="test",
            authorization_flow=create_test_flow(),
+            grant_types=[GrantType.IMPLICIT],
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://localhost")],
        )
        request = self.factory.get(
@@ -655,6 +671,7 @@ class TestAuthorize(OAuthTestCase):
            client_id="test",
            authorization_flow=create_test_flow(),
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://localhost")],
+            grant_types=[GrantType.IMPLICIT],
        )
        provider.property_mappings.set(
            ScopeMapping.objects.filter(
@@ -687,6 +704,7 @@ class TestAuthorize(OAuthTestCase):
            authorization_flow=flow,
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "foo://localhost")],
            access_code_validity="seconds=100",
+            grant_types=[GrantType.AUTHORIZATION_CODE],
        )
        Application.objects.create(name="app", slug="app", provider=provider)
        state = generate_id()
@@ -717,6 +735,7 @@ class TestAuthorize(OAuthTestCase):
            authorization_flow=flow,
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "foo://localhost")],
            access_code_validity="seconds=100",
+            grant_types=[GrantType.AUTHORIZATION_CODE],
        )
        Application.objects.create(name="app", slug="app", provider=provider)
        state = generate_id()
@@ -756,6 +775,7 @@ class TestAuthorize(OAuthTestCase):
            authentication_flow=auth_flow,
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "foo://localhost")],
            access_code_validity="seconds=100",
+            grant_types=[GrantType.AUTHORIZATION_CODE],
        )
        Application.objects.create(name="app", slug="app", provider=provider)
        state = generate_id()
@@ -782,6 +802,7 @@ class TestAuthorize(OAuthTestCase):
            authorization_flow=flow,
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "foo://localhost")],
            access_code_validity="seconds=100",
+            grant_types=[GrantType.AUTHORIZATION_CODE],
        )
        Application.objects.create(name="app", slug="app", provider=provider)
        state = generate_id()
--- a/authentik/providers/oauth2/tests/test_device_backchannel.py
+++ b/authentik/providers/oauth2/tests/test_device_backchannel.py
@@ -6,10 +6,11 @@ from urllib.parse import quote

 from django.urls import reverse

+from authentik.blueprints.tests import apply_blueprint
 from authentik.core.models import Application
 from authentik.core.tests.utils import create_test_flow
 from authentik.lib.generators import generate_id
-from authentik.providers.oauth2.models import OAuth2Provider
+from authentik.providers.oauth2.models import DeviceToken, GrantType, OAuth2Provider, ScopeMapping
 from authentik.providers.oauth2.tests.utils import OAuthTestCase


@@ -21,6 +22,7 @@ class TesOAuth2DeviceBackchannel(OAuthTestCase):
            name=generate_id(),
            client_id="test",
            authorization_flow=create_test_flow(),
+            grant_types=[GrantType.DEVICE_CODE],
        )
        self.application = Application.objects.create(
            name=generate_id(),
@@ -41,6 +43,21 @@ class TesOAuth2DeviceBackchannel(OAuthTestCase):
            reverse("authentik_providers_oauth2:device"),
        )
        self.assertEqual(res.status_code, 400)
+
+    def test_backchannel_invalid_no_grant(self):
+        """Test backchannel"""
+        self.provider.grant_types = []
+        self.provider.save()
+        res = self.client.post(
+            reverse("authentik_providers_oauth2:device"),
+            data={
+                "client_id": "test",
+            },
+        )
+        self.assertEqual(res.status_code, 400)
+
+    def test_backchannel_invalid_no_app(self):
+        """Test backchannel"""
        # test without application
        self.application.provider = None
        self.application.save()
@@ -110,3 +127,57 @@ class TesOAuth2DeviceBackchannel(OAuthTestCase):
        self.assertEqual(res.status_code, 200)
        body = loads(res.content.decode())
        self.assertEqual(body["expires_in"], 60)
+
+    @apply_blueprint("system/providers-oauth2.yaml")
+    def test_backchannel_scopes(self):
+        """Test backchannel"""
+        self.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",
+                ]
+            )
+        )
+        creds = b64encode(f"{self.provider.client_id}:".encode()).decode()
+        res = self.client.post(
+            reverse("authentik_providers_oauth2:device"),
+            HTTP_AUTHORIZATION=f"Basic {creds}",
+            data={"scope": "openid email"},
+        )
+        self.assertEqual(res.status_code, 200)
+        body = loads(res.content.decode())
+        self.assertEqual(body["expires_in"], 60)
+        token = DeviceToken.objects.filter(device_code=body["device_code"]).first()
+        self.assertIsNotNone(token)
+        self.assertEqual(len(token.scope), 2)
+        self.assertIn("openid", token.scope)
+        self.assertIn("email", token.scope)
+
+    @apply_blueprint("system/providers-oauth2.yaml")
+    def test_backchannel_scopes_extra(self):
+        """Test backchannel"""
+        self.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",
+                ]
+            )
+        )
+        creds = b64encode(f"{self.provider.client_id}:".encode()).decode()
+        res = self.client.post(
+            reverse("authentik_providers_oauth2:device"),
+            HTTP_AUTHORIZATION=f"Basic {creds}",
+            data={"scope": "openid email foo"},
+        )
+        self.assertEqual(res.status_code, 200)
+        body = loads(res.content.decode())
+        self.assertEqual(body["expires_in"], 60)
+        token = DeviceToken.objects.filter(device_code=body["device_code"]).first()
+        self.assertIsNotNone(token)
+        self.assertEqual(len(token.scope), 2)
+        self.assertIn("openid", token.scope)
+        self.assertIn("email", token.scope)
--- a/authentik/providers/oauth2/tests/test_device_init.py
+++ b/authentik/providers/oauth2/tests/test_device_init.py
@@ -9,7 +9,7 @@ from authentik.core.models import Application, Group
 from authentik.core.tests.utils import create_test_admin_user, create_test_brand, create_test_flow
 from authentik.lib.generators import generate_id
 from authentik.policies.models import PolicyBinding
-from authentik.providers.oauth2.models import DeviceToken, OAuth2Provider
+from authentik.providers.oauth2.models import DeviceToken, GrantType, OAuth2Provider
 from authentik.providers.oauth2.tests.utils import OAuthTestCase
 from authentik.providers.oauth2.views.device_init import QS_KEY_CODE

@@ -22,6 +22,7 @@ class TesOAuth2DeviceInit(OAuthTestCase):
            name=generate_id(),
            client_id="test",
            authorization_flow=create_test_flow(),
+            grant_types=[GrantType.DEVICE_CODE],
        )
        self.application = Application.objects.create(
            name=generate_id(),
--- a/authentik/providers/oauth2/tests/test_introspect.py
+++ b/authentik/providers/oauth2/tests/test_introspect.py
@@ -14,7 +14,7 @@ from authentik.lib.generators import generate_id
 from authentik.providers.oauth2.id_token import IDToken
 from authentik.providers.oauth2.models import (
    AccessToken,
-    ClientTypes,
+    ClientType,
    OAuth2Provider,
    RedirectURI,
    RedirectURIMatchingMode,
@@ -173,7 +173,7 @@ class TesOAuth2Introspection(OAuthTestCase):

    def test_introspect_provider_public(self):
        """Test introspect"""
-        self.provider.client_type = ClientTypes.PUBLIC
+        self.provider.client_type = ClientType.PUBLIC
        self.provider.save()
        token = AccessToken.objects.create(
            provider=self.provider,
@@ -208,7 +208,7 @@ class TesOAuth2Introspection(OAuthTestCase):
            authorization_flow=create_test_flow(),
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "")],
            signing_key=create_test_cert(),
-            client_type=ClientTypes.PUBLIC,
+            client_type=ClientType.PUBLIC,
        )
        Application.objects.create(name=generate_id(), slug=generate_id(), provider=other_provider)

--- a/authentik/providers/oauth2/tests/test_revoke.py
+++ b/authentik/providers/oauth2/tests/test_revoke.py
@@ -13,7 +13,7 @@ from authentik.lib.generators import generate_id
 from authentik.providers.oauth2.id_token import IDToken
 from authentik.providers.oauth2.models import (
    AccessToken,
-    ClientTypes,
+    ClientType,
    DeviceToken,
    OAuth2Provider,
    RedirectURI,
@@ -126,7 +126,7 @@ class TesOAuth2Revoke(OAuthTestCase):

    def test_revoke_public(self):
        """Test revoke public client"""
-        self.provider.client_type = ClientTypes.PUBLIC
+        self.provider.client_type = ClientType.PUBLIC
        self.provider.save()
        token = AccessToken.objects.create(
            provider=self.provider,
@@ -241,7 +241,7 @@ class TesOAuth2Revoke(OAuthTestCase):
            authorization_flow=create_test_flow(),
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "")],
            signing_key=create_test_cert(),
-            client_type=ClientTypes.PUBLIC,
+            client_type=ClientType.PUBLIC,
        )
        Application.objects.create(name=generate_id(), slug=generate_id(), provider=other_provider)

@@ -270,14 +270,14 @@ class TesOAuth2Revoke(OAuthTestCase):
    def test_revoke_provider_fed_public(self):
        """Test revoke with federation. self.provider is a public
        client and other_provider is a public client."""
-        self.provider.client_type = ClientTypes.PUBLIC
+        self.provider.client_type = ClientType.PUBLIC
        self.provider.save()
        other_provider = OAuth2Provider.objects.create(
            name=generate_id(),
            authorization_flow=create_test_flow(),
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "")],
            signing_key=create_test_cert(),
-            client_type=ClientTypes.PUBLIC,
+            client_type=ClientType.PUBLIC,
        )
        Application.objects.create(name=generate_id(), slug=generate_id(), provider=other_provider)

--- a/authentik/providers/oauth2/tests/test_token.py
+++ b/authentik/providers/oauth2/tests/test_token.py
@@ -25,6 +25,7 @@ from authentik.providers.oauth2.errors import TokenError
 from authentik.providers.oauth2.models import (
    AccessToken,
    AuthorizationCode,
+    GrantType,
    OAuth2Provider,
    RedirectURI,
    RedirectURIMatchingMode,
@@ -44,11 +45,39 @@ class TestToken(OAuthTestCase):
        self.factory = RequestFactory()
        self.app = Application.objects.create(name=generate_id(), slug="test")

+    def test_invalid_grant_type(self):
+        """test request param"""
+        provider = OAuth2Provider.objects.create(
+            name=generate_id(),
+            authorization_flow=create_test_flow(),
+            grant_types=[],
+            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://TestServer")],
+            signing_key=self.keypair,
+        )
+        header = b64encode(f"{provider.client_id}:{provider.client_secret}".encode()).decode()
+        user = create_test_admin_user()
+        code = AuthorizationCode.objects.create(
+            code="foobar", provider=provider, user=user, auth_time=timezone.now()
+        )
+        request = self.factory.post(
+            "/",
+            data={
+                "grant_type": GRANT_TYPE_AUTHORIZATION_CODE,
+                "code": code.code,
+                "redirect_uri": "http://TestServer",
+            },
+            HTTP_AUTHORIZATION=f"Basic {header}",
+        )
+        with self.assertRaises(TokenError) as cm:
+            TokenParams.parse(request, provider, provider.client_id, provider.client_secret)
+        self.assertEqual(cm.exception.cause, "grant_type_not_configured")
+
    def test_request_auth_code(self):
        """test request param"""
        provider = OAuth2Provider.objects.create(
            name=generate_id(),
            authorization_flow=create_test_flow(),
+            grant_types=[GrantType.AUTHORIZATION_CODE],
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://TestServer")],
            signing_key=self.keypair,
        )
@@ -76,6 +105,7 @@ class TestToken(OAuthTestCase):
        provider = OAuth2Provider.objects.create(
            name=generate_id(),
            authorization_flow=create_test_flow(),
+            grant_types=[GrantType.REFRESH_TOKEN],
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://testserver")],
            signing_key=self.keypair,
        )
@@ -97,6 +127,7 @@ class TestToken(OAuthTestCase):
        provider = OAuth2Provider.objects.create(
            name=generate_id(),
            authorization_flow=create_test_flow(),
+            grant_types=[GrantType.REFRESH_TOKEN],
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://local.invalid")],
            signing_key=self.keypair,
        )
@@ -139,6 +170,7 @@ class TestToken(OAuthTestCase):
        provider = OAuth2Provider.objects.create(
            name=generate_id(),
            authorization_flow=create_test_flow(),
+            grant_types=[GrantType.AUTHORIZATION_CODE],
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://local.invalid")],
            signing_key=self.keypair,
        )
@@ -179,6 +211,7 @@ class TestToken(OAuthTestCase):
        provider = OAuth2Provider.objects.create(
            name=generate_id(),
            authorization_flow=create_test_flow(),
+            grant_types=[GrantType.AUTHORIZATION_CODE],
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://local.invalid")],
            signing_key=self.keypair,
            encryption_key=self.keypair,
@@ -210,6 +243,7 @@ class TestToken(OAuthTestCase):
        provider = OAuth2Provider.objects.create(
            name=generate_id(),
            authorization_flow=create_test_flow(),
+            grant_types=[GrantType.REFRESH_TOKEN],
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://local.invalid")],
            signing_key=self.keypair,
        )
@@ -271,6 +305,7 @@ class TestToken(OAuthTestCase):
        provider = OAuth2Provider.objects.create(
            name=generate_id(),
            authorization_flow=create_test_flow(),
+            grant_types=[GrantType.REFRESH_TOKEN],
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://local.invalid")],
            signing_key=self.keypair,
        )
@@ -328,6 +363,7 @@ class TestToken(OAuthTestCase):
        provider = OAuth2Provider.objects.create(
            name=generate_id(),
            authorization_flow=create_test_flow(),
+            grant_types=[GrantType.REFRESH_TOKEN],
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://testserver")],
            signing_key=self.keypair,
        )
@@ -400,6 +436,7 @@ class TestToken(OAuthTestCase):
        provider = OAuth2Provider.objects.create(
            name=generate_id(),
            authorization_flow=create_test_flow(),
+            grant_types=[GrantType.REFRESH_TOKEN],
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://local.invalid")],
            signing_key=self.keypair,
            refresh_token_threshold="hours=1",  # nosec
@@ -497,6 +534,7 @@ class TestToken(OAuthTestCase):
        provider = OAuth2Provider.objects.create(
            name=generate_id(),
            authorization_flow=create_test_flow(),
+            grant_types=[GrantType.AUTHORIZATION_CODE],
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://local.invalid")],
            signing_key=self.keypair,
            include_claims_in_id_token=True,
--- a/authentik/providers/oauth2/tests/test_token_cc_jwt_provider.py
+++ b/authentik/providers/oauth2/tests/test_token_cc_jwt_provider.py
@@ -22,6 +22,7 @@ from authentik.lib.generators import generate_id
 from authentik.policies.models import PolicyBinding
 from authentik.providers.oauth2.models import (
    AccessToken,
+    GrantType,
    OAuth2Provider,
    RedirectURI,
    RedirectURIMatchingMode,
@@ -55,6 +56,7 @@ class TestTokenClientCredentialsJWTProvider(OAuthTestCase):
            authorization_flow=create_test_flow(),
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://testserver")],
            signing_key=self.cert,
+            grant_types=[GrantType.CLIENT_CREDENTIALS],
        )
        self.provider.jwt_federation_providers.add(self.other_provider)
        self.provider.property_mappings.set(ScopeMapping.objects.all())
--- a/authentik/providers/oauth2/tests/test_token_cc_jwt_source.py
+++ b/authentik/providers/oauth2/tests/test_token_cc_jwt_source.py
@@ -20,6 +20,7 @@ from authentik.core.tests.utils import create_test_cert, create_test_flow
 from authentik.lib.generators import generate_id
 from authentik.policies.models import PolicyBinding
 from authentik.providers.oauth2.models import (
+    GrantType,
    OAuth2Provider,
    RedirectURI,
    RedirectURIMatchingMode,
@@ -68,6 +69,7 @@ class TestTokenClientCredentialsJWTSource(OAuthTestCase):
            authorization_flow=create_test_flow(),
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://testserver")],
            signing_key=self.cert,
+            grant_types=[GrantType.CLIENT_CREDENTIALS],
        )
        self.provider.jwt_federation_sources.add(self.source)
        self.provider.property_mappings.set(ScopeMapping.objects.all())
--- a/Show More
+++ b/Show More