ci: bump taiki-e/install-action in /.github/actions/setup

Bumps [taiki-e/install-action](https://github.com/taiki-e/install-action) from 2.75.23 to 2.75.25. - [Release notes](https://github.com/taiki-e/install-action/releases) - [Changelog](https://github.com/taiki-e/install-action/blob/main/CHANGELOG.md) - [Commits](481c34c1cf...1329c298aa) --- updated-dependencies: - dependency-name: taiki-e/install-action dependency-version: 2.75.25 dependency-type: direct:production update-type: version-update:semver-patch ... Signed-off-by: dependabot[bot] <support@github.com>
enterprise: account lockdown (#18615 )
2026-05-01 11:57:09 +02:00 · 2026-05-01 04:40:21 +00:00 · 2026-04-30 23:02:46 +00:00 · 2026-04-30 14:11:07 -05:00 · 2026-04-30 17:45:56 +02:00 · 2026-04-30 14:58:10 +00:00
417 changed files with 35570 additions and 5659 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
@@ -64,7 +64,7 @@ runs:
        rustflags: ""
    - name: Setup rust dependencies
      if: ${{ contains(inputs.dependencies, 'rust') }}
-      uses: taiki-e/install-action@055f5df8c3f65ea01cd41e9dc855becd88953486 # v2
+      uses: taiki-e/install-action@1329c298aa20c3257846c9b2e0e55967df3e3c37 # v2
      with:
        tool: cargo-deny cargo-machete cargo-llvm-cov nextest
    - name: Setup node (web)
@@ -104,7 +104,7 @@ runs:
      working-directory: ${{ inputs.working-directory }}
      run: |
        export PSQL_TAG=${{ inputs.postgresql_version }}
-        docker compose -f .github/actions/setup/compose.yml up -d
+        docker compose -f .github/actions/setup/compose.yml up -d --wait
        cd web && npm ci
    - name: Generate config
      if: ${{ contains(inputs.dependencies, 'python') }}
--- a/.github/actions/setup/compose.yml
+++ b/.github/actions/setup/compose.yml
@@ -8,8 +8,14 @@ services:
      POSTGRES_USER: authentik
      POSTGRES_PASSWORD: "EK-5jnKfjrGRm<77"
      POSTGRES_DB: authentik
+      PGDATA: /var/lib/postgresql/data/pgdata
    ports:
      - 5432:5432
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U $${POSTGRES_USER} -d $${POSTGRES_DB} -h 127.0.0.1"]
+      interval: 1s
+      timeout: 5s
+      retries: 60
    restart: always
  s3:
    container_name: s3
--- a/.github/dependabot.yml
+++ b/.github/dependabot.yml
@@ -67,6 +67,12 @@ updates:
      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
--- 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@3de37de96c4e900bc3eef9055d3c50abdb4f769d # v2
+      - uses: int128/docker-manifest-create-action@7df7f9e221d927eaadf87db231ddf728047308a4 # v2
        id: build
        with:
          tags: ${{ matrix.tag }}
--- 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/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",
@@ -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.38"
+version = "0.23.40"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "69f9466fb2c14ea04357e91413efb882e2a6d4a406e625449bc0a5d360d53a21"
+checksum = "ef86cd5876211988985292b91c96a8f2d298df24e75989a43a3c73f2d4d8168b"
 dependencies = [
 "aws-lc-rs",
 "log",
@@ -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"
--- 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.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.38", features = ["fips"] }
+rustls = { version = "= 0.23.40", features = ["fips"] }
 sentry = { version = "= 0.47.0", default-features = false, features = [
  "backtrace",
  "contexts",
@@ -106,16 +114,10 @@ tracing-subscriber = { version = "= 0.3.23", features = [
 url = "= 2.5.8"
 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/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/api.py
+++ b/authentik/blueprints/api.py
@@ -126,7 +126,7 @@ class BlueprintInstanceSerializer(ModelSerializer):

 def check_blueprint_perms(blueprint: Blueprint, user: User, explicit_action: str | None = None):
    """Check for individual permissions for each model in a blueprint"""
-    for entry in blueprint.entries:
+    for entry in blueprint.iter_entries():
        full_model = entry.get_model(blueprint)
        app, __, model = full_model.partition(".")
        perms = [
--- 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/blueprints/tests/test_v1_conditions.py
+++ b/authentik/blueprints/tests/test_v1_conditions.py
@@ -1,8 +1,11 @@
 """Test blueprints v1"""

+from unittest.mock import patch
+
 from django.test import TransactionTestCase

 from authentik.blueprints.v1.importer import Importer
+from authentik.enterprise.license import LicenseKey
 from authentik.flows.models import Flow
 from authentik.lib.generators import generate_id
 from authentik.lib.tests.utils import load_fixture
@@ -42,3 +45,45 @@ class TestBlueprintsV1Conditions(TransactionTestCase):
        # Ensure objects do not exist
        self.assertFalse(Flow.objects.filter(slug=flow_slug1))
        self.assertFalse(Flow.objects.filter(slug=flow_slug2))
+
+    def test_enterprise_license_context_unlicensed(self):
+        """Test enterprise license context defaults to a false boolean when unlicensed."""
+        license_key = LicenseKey("test", 0, "Test license", 0, 0)
+
+        with patch("authentik.enterprise.license.LicenseKey.get_total", return_value=license_key):
+            importer = Importer.from_string("""
+version: 1
+entries:
+    - identifiers:
+          name: enterprise-test
+          slug: enterprise-test
+      model: authentik_flows.flow
+      conditions:
+          - !Context goauthentik.io/enterprise/licensed
+      attrs:
+          designation: stage_configuration
+          title: foo
+""")
+
+        self.assertIs(importer.blueprint.context["goauthentik.io/enterprise/licensed"], False)
+
+    def test_enterprise_license_context_licensed(self):
+        """Test enterprise license context defaults to a true boolean when licensed."""
+        license_key = LicenseKey("test", 253402300799, "Test license", 1000, 1000)
+
+        with patch("authentik.enterprise.license.LicenseKey.get_total", return_value=license_key):
+            importer = Importer.from_string("""
+version: 1
+entries:
+    - identifiers:
+          name: enterprise-test
+          slug: enterprise-test
+      model: authentik_flows.flow
+      conditions:
+          - !Context goauthentik.io/enterprise/licensed
+      attrs:
+          designation: stage_configuration
+          title: foo
+""")
+
+        self.assertIs(importer.blueprint.context["goauthentik.io/enterprise/licensed"], True)
--- a/authentik/blueprints/v1/importer.py
+++ b/authentik/blueprints/v1/importer.py
@@ -146,9 +146,7 @@ class Importer:
        try:
            from authentik.enterprise.license import LicenseKey

-            context["goauthentik.io/enterprise/licensed"] = (
-                LicenseKey.get_total().status().is_valid,
-            )
+            context["goauthentik.io/enterprise/licensed"] = LicenseKey.get_total().status().is_valid
        except ModuleNotFoundError:
            pass
        return context
--- a/authentik/brands/api.py
+++ b/authentik/brands/api.py
@@ -64,6 +64,7 @@ class BrandSerializer(ModelSerializer):
            "flow_unenrollment",
            "flow_user_settings",
            "flow_device_code",
+            "flow_lockdown",
            "default_application",
            "web_certificate",
            "client_certificates",
@@ -117,6 +118,7 @@ class CurrentBrandSerializer(PassiveSerializer):
    flow_unenrollment = CharField(source="flow_unenrollment.slug", required=False)
    flow_user_settings = CharField(source="flow_user_settings.slug", required=False)
    flow_device_code = CharField(source="flow_device_code.slug", required=False)
+    flow_lockdown = CharField(source="flow_lockdown.slug", required=False)

    default_locale = CharField(read_only=True)
    flags = SerializerMethodField()
@@ -154,6 +156,7 @@ class BrandViewSet(UsedByMixin, ModelViewSet):
        "flow_unenrollment",
        "flow_user_settings",
        "flow_device_code",
+        "flow_lockdown",
        "web_certificate",
        "client_certificates",
    ]
--- a/authentik/brands/migrations/0012_brand_flow_lockdown.py
+++ b/authentik/brands/migrations/0012_brand_flow_lockdown.py
@@ -0,0 +1,25 @@
+# Generated by Django 5.2.12 on 2026-03-14 02:58
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("authentik_brands", "0011_alter_brand_branding_default_flow_background_and_more"),
+        ("authentik_flows", "0031_alter_flow_layout"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="brand",
+            name="flow_lockdown",
+            field=models.ForeignKey(
+                null=True,
+                on_delete=django.db.models.deletion.SET_NULL,
+                related_name="brand_lockdown",
+                to="authentik_flows.flow",
+            ),
+        ),
+    ]
--- a/authentik/brands/models.py
+++ b/authentik/brands/models.py
@@ -58,6 +58,9 @@ class Brand(SerializerModel):
    flow_device_code = models.ForeignKey(
        Flow, null=True, on_delete=models.SET_NULL, related_name="brand_device_code"
    )
+    flow_lockdown = models.ForeignKey(
+        Flow, null=True, on_delete=models.SET_NULL, related_name="brand_lockdown"
+    )

    default_application = models.ForeignKey(
        "authentik_core.Application",
--- 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
@@ -47,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
@@ -37,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",
@@ -79,6 +150,7 @@ 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)
@@ -193,9 +265,6 @@ class GroupSerializer(ModelSerializer):
            "children_obj",
        ]
        extra_kwargs = {
-            "users": {
-                "default": list,
-            },
            "children": {
                "required": False,
                "default": list,
@@ -225,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):
@@ -276,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
@@ -287,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/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,
@@ -105,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"""
@@ -131,7 +137,7 @@ class PartialGroupSerializer(ModelSerializer):
 class UserSerializer(ModelSerializer):
    """User Serializer"""

-    is_superuser = BooleanField(read_only=True)
+    is_superuser = SerializerMethodField()
    avatar = SerializerMethodField()
    attributes = JSONDictField(required=False)
    groups = PrimaryKeyRelatedField(
@@ -168,6 +174,14 @@ class UserSerializer(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:
@@ -181,47 +195,79 @@ class UserSerializer(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()
@@ -390,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"""

@@ -511,6 +563,9 @@ class UsersFilter(FilterSet):


 class UserViewSet(
+    ConditionalInheritance(
+        "authentik.enterprise.stages.account_lockdown.api.UserAccountLockdownMixin"
+    ),
    ConditionalInheritance("authentik.enterprise.reports.api.reports.ExportMixin"),
    UsedByMixin,
    ModelViewSet,
@@ -541,10 +596,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(
@@ -713,6 +788,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,
@@ -736,9 +816,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_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,7 +10,7 @@ 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.sessions.base_session import AbstractBaseSession
@@ -560,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
@@ -735,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()

@@ -790,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"""
--- 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_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
@@ -19,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 (
@@ -35,7 +35,7 @@ from authentik.tenants.channels import TenantsAwareMiddleware
 urlpatterns = [
    path(
        "",
-        login_required(RootRedirectView.as_view()),
+        RootRedirectView.as_view(),
        name="root-redirect",
    ),
    path(
@@ -62,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(
--- 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/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/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/lifecycle/api/iterations.py
+++ b/authentik/enterprise/lifecycle/api/iterations.py
@@ -1,7 +1,6 @@
 from datetime import datetime

-from django.db.models import BooleanField as ModelBooleanField
-from django.db.models import Case, Q, Value, When
+from django.db.models import Exists, OuterRef, Q, Subquery
 from django_filters.rest_framework import BooleanFilter, FilterSet
 from drf_spectacular.utils import extend_schema
 from rest_framework.decorators import action
@@ -14,7 +13,7 @@ from rest_framework.viewsets import GenericViewSet
 from authentik.core.api.utils import ModelSerializer
 from authentik.enterprise.api import EnterpriseRequiredMixin
 from authentik.enterprise.lifecycle.api.reviews import ReviewSerializer
-from authentik.enterprise.lifecycle.models import LifecycleIteration, ReviewState
+from authentik.enterprise.lifecycle.models import LifecycleIteration, LifecycleRule, ReviewState
 from authentik.enterprise.lifecycle.utils import (
    ContentTypeField,
    ReviewerGroupSerializer,
@@ -26,20 +25,25 @@ from authentik.enterprise.lifecycle.utils import (
 from authentik.lib.utils.time import timedelta_from_string


+class RelatedRuleSerializer(EnterpriseRequiredMixin, ModelSerializer):
+    reviewer_groups = ReviewerGroupSerializer(many=True, read_only=True)
+    min_reviewers = IntegerField(read_only=True)
+    reviewers = ReviewerUserSerializer(many=True, read_only=True)
+
+    class Meta:
+        model = LifecycleRule
+        fields = ["id", "name", "reviewer_groups", "min_reviewers", "reviewers"]
+
+
 class LifecycleIterationSerializer(EnterpriseRequiredMixin, ModelSerializer):
    content_type = ContentTypeField()
    object_verbose = SerializerMethodField()
+    rule = RelatedRuleSerializer(read_only=True)
    object_admin_url = SerializerMethodField(read_only=True)
    grace_period_end = SerializerMethodField(read_only=True)
    reviews = ReviewSerializer(many=True, read_only=True, source="review_set.all")
    user_can_review = SerializerMethodField(read_only=True)

-    reviewer_groups = ReviewerGroupSerializer(
-        many=True, read_only=True, source="rule.reviewer_groups"
-    )
-    min_reviewers = IntegerField(read_only=True, source="rule.min_reviewers")
-    reviewers = ReviewerUserSerializer(many=True, read_only=True, source="rule.reviewers")
-
    next_review_date = SerializerMethodField(read_only=True)

    class Meta:
@@ -55,10 +59,8 @@ class LifecycleIterationSerializer(EnterpriseRequiredMixin, ModelSerializer):
            "grace_period_end",
            "next_review_date",
            "reviews",
+            "rule",
            "user_can_review",
-            "reviewer_groups",
-            "min_reviewers",
-            "reviewers",
        ]
        read_only_fields = fields

@@ -88,43 +90,55 @@ class IterationViewSet(EnterpriseRequiredMixin, CreateModelMixin, GenericViewSet
    queryset = LifecycleIteration.objects.all()
    serializer_class = LifecycleIterationSerializer
    ordering = ["-opened_on"]
-    ordering_fields = ["state", "content_type__model", "opened_on", "grace_period_end"]
+    ordering_fields = [
+        "state",
+        "content_type__model",
+        "rule__name",
+        "opened_on",
+        "grace_period_end",
+    ]
    filterset_class = LifecycleIterationFilterSet

    def get_queryset(self):
        user = self.request.user
        return self.queryset.annotate(
-            user_is_reviewer=Case(
-                When(
-                    Q(rule__reviewers=user)
-                    | Q(rule__reviewer_groups__in=user.groups.all().with_ancestors()),
-                    then=Value(True),
-                ),
-                default=Value(False),
-                output_field=ModelBooleanField(),
+            user_is_reviewer=Exists(
+                LifecycleRule.objects.filter(
+                    pk=OuterRef("rule_id"),
+                ).filter(
+                    Q(reviewers=user) | Q(reviewer_groups__in=user.groups.all().with_ancestors())
+                )
            )
-        ).distinct()
+        )

+    @extend_schema(
+        operation_id="lifecycle_iterations_list_latest",
+        responses={200: LifecycleIterationSerializer(many=True)},
+    )
    @action(
        detail=False,
+        pagination_class=None,
        methods=["get"],
        url_path=r"latest/(?P<content_type>[^/]+)/(?P<object_id>[^/]+)",
    )
-    def latest_iteration(self, request: Request, content_type: str, object_id: str) -> Response:
+    def latest_iterations(self, request: Request, content_type: str, object_id: str) -> Response:
        ct = parse_content_type(content_type)
-        try:
-            obj = (
-                self.get_queryset()
-                .filter(
-                    content_type__app_label=ct["app_label"],
-                    content_type__model=ct["model"],
-                    object_id=object_id,
-                )
-                .latest("opened_on")
+        latest_ids_subquery = (
+            LifecycleIteration.objects.filter(
+                rule=OuterRef("rule"),
+                content_type__app_label=ct["app_label"],
+                content_type__model=ct["model"],
+                object_id=object_id,
            )
-        except LifecycleIteration.DoesNotExist:
-            return Response(status=404)
-        serializer = self.get_serializer(obj)
+            .order_by("-opened_on")
+            .values("id")[:1]
+        )
+        latest_per_rule = LifecycleIteration.objects.filter(
+            content_type__app_label=ct["app_label"],
+            content_type__model=ct["model"],
+            object_id=object_id,
+        ).filter(id=Subquery(latest_ids_subquery))
+        serializer = self.get_serializer(latest_per_rule, many=True)
        return Response(serializer.data)

    @extend_schema(
--- a/authentik/enterprise/lifecycle/api/rules.py
+++ b/authentik/enterprise/lifecycle/api/rules.py
@@ -84,23 +84,6 @@ class LifecycleRuleSerializer(EnterpriseRequiredMixin, ModelSerializer):
                raise ValidationError(
                    {"grace_period": _("Grace period must be shorter than the interval.")}
                )
-        if "content_type" in attrs or "object_id" in attrs:
-            content_type = attrs.get("content_type", getattr(self.instance, "content_type", None))
-            object_id = attrs.get("object_id", getattr(self.instance, "object_id", None))
-            if content_type is not None and object_id is None:
-                existing = LifecycleRule.objects.filter(
-                    content_type=content_type, object_id__isnull=True
-                )
-                if self.instance:
-                    existing = existing.exclude(pk=self.instance.pk)
-                if existing.exists():
-                    raise ValidationError(
-                        {
-                            "content_type": _(
-                                "Only one type-wide rule for each object type is allowed."
-                            )
-                        }
-                    )
        return attrs


--- a/authentik/enterprise/lifecycle/migrations/0003_remove_lifecyclerule_uniq_lifecycle_rule_ct_null_object_and_more.py
+++ b/authentik/enterprise/lifecycle/migrations/0003_remove_lifecyclerule_uniq_lifecycle_rule_ct_null_object_and_more.py
@@ -0,0 +1,21 @@
+# Generated by Django 5.2.11 on 2026-03-05 11:27
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("authentik_lifecycle", "0002_alter_lifecycleiteration_opened_on"),
+    ]
+
+    operations = [
+        migrations.RemoveConstraint(
+            model_name="lifecyclerule",
+            name="uniq_lifecycle_rule_ct_null_object",
+        ),
+        migrations.AlterUniqueTogether(
+            name="lifecyclerule",
+            unique_together=set(),
+        ),
+    ]
--- a/authentik/enterprise/lifecycle/models.py
+++ b/authentik/enterprise/lifecycle/models.py
@@ -56,14 +56,6 @@ class LifecycleRule(SerializerModel):

    class Meta:
        indexes = [models.Index(fields=["content_type"])]
-        unique_together = [["content_type", "object_id"]]
-        constraints = [
-            models.UniqueConstraint(
-                fields=["content_type"],
-                condition=Q(object_id__isnull=True),
-                name="uniq_lifecycle_rule_ct_null_object",
-            )
-        ]

    @property
    def serializer(self) -> type[BaseSerializer]:
@@ -82,12 +74,6 @@ class LifecycleRule(SerializerModel):
        qs = self.content_type.get_all_objects_for_this_type()
        if self.object_id:
            qs = qs.filter(pk=self.object_id)
-        else:
-            qs = qs.exclude(
-                pk__in=LifecycleRule.objects.filter(
-                    content_type=self.content_type, object_id__isnull=False
-                ).values_list(Cast("object_id", output_field=self._get_pk_field()), flat=True)
-            )
        return qs

    def _get_stale_iterations(self) -> QuerySet[LifecycleIteration]:
@@ -107,8 +93,7 @@ class LifecycleRule(SerializerModel):

    def _get_newly_due_objects(self) -> QuerySet:
        recent_iteration_ids = LifecycleIteration.objects.filter(
-            content_type=self.content_type,
-            object_id__isnull=False,
+            rule=self,
            opened_on__gte=start_of_day(
                timezone.now() + timedelta(days=1) - timedelta_from_string(self.interval)
            ),
@@ -214,9 +199,15 @@ class LifecycleIteration(SerializerModel, ManagedModel):
        }

    def initialize(self):
+        if (self.content_type.app_label, self.content_type.model) == ("authentik_core", "group"):
+            object_label = self.object.name
+        elif (self.content_type.app_label, self.content_type.model) == ("authentik_rbac", "role"):
+            object_label = self.object.name
+        else:
+            object_label = str(self.object)
        event = Event.new(
            EventAction.REVIEW_INITIATED,
-            message=_(f"Access review is due for {self.content_type.name} {str(self.object)}"),
+            message=_(f"Access review is due for {self.content_type.name.lower()} {object_label}"),
            **self._get_event_args(),
        )
        event.save()
--- a/authentik/enterprise/lifecycle/signals.py
+++ b/authentik/enterprise/lifecycle/signals.py
@@ -3,6 +3,7 @@ from django.db.models.signals import post_save, pre_delete
 from django.dispatch import receiver

 from authentik.enterprise.lifecycle.models import LifecycleRule, ReviewState
+from authentik.tasks.schedules.models import Schedule


@receiver(post_save, sender=LifecycleRule)
@@ -11,7 +12,9 @@ def post_rule_save(sender, instance: LifecycleRule, created: bool, **_):

    apply_lifecycle_rule.send_with_options(
        args=(instance.id,),
-        rel_obj=instance,
+        rel_obj=Schedule.objects.get(
+            actor_name="authentik.enterprise.lifecycle.tasks.apply_lifecycle_rules"
+        ),
    )


--- a/authentik/enterprise/lifecycle/tasks.py
+++ b/authentik/enterprise/lifecycle/tasks.py
@@ -4,14 +4,17 @@ from dramatiq import actor
 from authentik.core.models import User
 from authentik.enterprise.lifecycle.models import LifecycleRule
 from authentik.events.models import Event, Notification, NotificationTransport
+from authentik.tasks.schedules.models import Schedule


-@actor(description=_("Dispatch tasks to validate lifecycle rules."))
+@actor(description=_("Dispatch tasks to apply lifecycle rules."))
 def apply_lifecycle_rules():
    for rule in LifecycleRule.objects.all():
        apply_lifecycle_rule.send_with_options(
            args=(rule.id,),
-            rel_obj=rule,
+            rel_obj=Schedule.objects.get(
+                actor_name="authentik.enterprise.lifecycle.tasks.apply_lifecycle_rules"
+            ),
        )


--- a/authentik/enterprise/lifecycle/tests/test_api.py
+++ b/authentik/enterprise/lifecycle/tests/test_api.py
@@ -1,3 +1,4 @@
+from django.apps import apps
 from django.contrib.contenttypes.models import ContentType
 from django.urls import reverse
 from rest_framework.test import APITestCase
@@ -19,6 +20,11 @@ class TestLifecycleRuleAPI(APITestCase):
        self.content_type = ContentType.objects.get_for_model(Application)
        self.reviewer_group = Group.objects.create(name=generate_id())

+    @classmethod
+    def setUpTestData(cls):
+        config = apps.get_app_config("authentik_tasks_schedules")
+        config._on_startup_callback(None)
+
    def test_list_rules(self):
        rule = LifecycleRule.objects.create(
            name=generate_id(),
@@ -190,6 +196,11 @@ class TestIterationAPI(APITestCase):
        self.reviewer_group = Group.objects.create(name=generate_id())
        self.reviewer_group.users.add(self.user)

+    @classmethod
+    def setUpTestData(cls):
+        config = apps.get_app_config("authentik_tasks_schedules")
+        config._on_startup_callback(None)
+
    def test_open_iterations(self):
        rule = LifecycleRule.objects.create(
            name=generate_id(),
@@ -231,7 +242,7 @@ class TestIterationAPI(APITestCase):

        response = self.client.get(
            reverse(
-                "authentik_api:lifecycleiteration-latest-iteration",
+                "authentik_api:lifecycleiteration-latest-iterations",
                kwargs={
                    "content_type": f"{self.content_type.app_label}.{self.content_type.model}",
                    "object_id": str(self.app.pk),
@@ -239,19 +250,20 @@ class TestIterationAPI(APITestCase):
            )
        )
        self.assertEqual(response.status_code, 200)
-        self.assertEqual(response.data["object_id"], str(self.app.pk))
+        self.assertEqual(len(response.data), 1)
+        self.assertEqual(response.data[0]["object_id"], str(self.app.pk))

    def test_latest_iteration_not_found(self):
        response = self.client.get(
            reverse(
-                "authentik_api:lifecycleiteration-latest-iteration",
+                "authentik_api:lifecycleiteration-latest-iterations",
                kwargs={
                    "content_type": f"{self.content_type.app_label}.{self.content_type.model}",
                    "object_id": "00000000-0000-0000-0000-000000000000",
                },
            )
        )
-        self.assertEqual(response.status_code, 404)
+        self.assertEqual(response.data, [])

    def test_iteration_includes_user_can_review(self):
        rule = LifecycleRule.objects.create(
@@ -279,6 +291,11 @@ class TestReviewAPI(APITestCase):
        self.reviewer_group = Group.objects.create(name=generate_id())
        self.reviewer_group.users.add(self.user)

+    @classmethod
+    def setUpTestData(cls):
+        config = apps.get_app_config("authentik_tasks_schedules")
+        config._on_startup_callback(None)
+
    def test_create_review(self):
        rule = LifecycleRule.objects.create(
            name=generate_id(),
--- a/authentik/enterprise/lifecycle/tests/test_models.py
+++ b/authentik/enterprise/lifecycle/tests/test_models.py
@@ -2,6 +2,7 @@ import datetime as dt
 from datetime import timedelta
 from unittest.mock import patch

+from django.apps import apps
 from django.contrib.contenttypes.models import ContentType
 from django.test import RequestFactory, TestCase
 from django.utils import timezone
@@ -29,6 +30,11 @@ class TestLifecycleModels(TestCase):
    def setUp(self):
        self.factory = RequestFactory()

+    @classmethod
+    def setUpTestData(cls):
+        config = apps.get_app_config("authentik_tasks_schedules")
+        config._on_startup_callback(None)
+
    def _get_request(self):
        return self.factory.get("/")

@@ -438,31 +444,6 @@ class TestLifecycleModels(TestCase):
        self.assertIn(app_one, objects)
        self.assertIn(app_two, objects)

-    def test_rule_type_excludes_objects_with_specific_rules(self):
-        app_with_rule = Application.objects.create(name=generate_id(), slug=generate_id())
-        app_without_rule = Application.objects.create(name=generate_id(), slug=generate_id())
-        content_type = ContentType.objects.get_for_model(Application)
-
-        # Create a specific rule for app_with_rule
-        LifecycleRule.objects.create(
-            name=generate_id(),
-            content_type=content_type,
-            object_id=str(app_with_rule.pk),
-            interval="days=30",
-        )
-
-        # Create a type-level rule
-        type_rule = LifecycleRule.objects.create(
-            name=generate_id(),
-            content_type=content_type,
-            object_id=None,
-            interval="days=60",
-        )
-
-        objects = list(type_rule.get_objects())
-        self.assertNotIn(app_with_rule, objects)
-        self.assertIn(app_without_rule, objects)
-
    def test_rule_type_apply_creates_iterations_for_all_objects(self):
        app_one = Application.objects.create(name=generate_id(), slug=generate_id())
        app_two = Application.objects.create(name=generate_id(), slug=generate_id())
@@ -669,6 +650,73 @@ class TestLifecycleModels(TestCase):
        self.assertIn(explicit_reviewer, reviewers)
        self.assertIn(group_member, reviewers)

+    def test_multiple_rules_same_object_create_separate_iterations(self):
+        """Two rules targeting the same object each create their own iteration."""
+        obj = Application.objects.create(name=generate_id(), slug=generate_id())
+        content_type = ContentType.objects.get_for_model(obj)
+
+        rule_one = self._create_rule_for_object(obj, interval="days=30", grace_period="days=10")
+        rule_two = self._create_rule_for_object(obj, interval="days=60", grace_period="days=20")
+
+        iterations = LifecycleIteration.objects.filter(
+            content_type=content_type, object_id=str(obj.pk)
+        )
+        self.assertEqual(iterations.count(), 2)
+
+        iter_one = iterations.get(rule=rule_one)
+        iter_two = iterations.get(rule=rule_two)
+        self.assertEqual(iter_one.state, ReviewState.PENDING)
+        self.assertEqual(iter_two.state, ReviewState.PENDING)
+        self.assertNotEqual(iter_one.pk, iter_two.pk)
+
+    def test_multiple_rules_same_object_reviewed_independently(self):
+        """Reviewing one rule's iteration does not affect the other rule's iteration."""
+        obj = Application.objects.create(name=generate_id(), slug=generate_id())
+        content_type = ContentType.objects.get_for_model(obj)
+
+        reviewer = create_test_user()
+
+        rule_one = self._create_rule_for_object(obj, min_reviewers=1)
+        rule_two = self._create_rule_for_object(obj, min_reviewers=1)
+
+        group = Group.objects.create(name=generate_id())
+        group.users.add(reviewer)
+        rule_one.reviewer_groups.add(group)
+        rule_two.reviewer_groups.add(group)
+
+        iter_one = LifecycleIteration.objects.get(
+            content_type=content_type, object_id=str(obj.pk), rule=rule_one
+        )
+        iter_two = LifecycleIteration.objects.get(
+            content_type=content_type, object_id=str(obj.pk), rule=rule_two
+        )
+
+        request = self._get_request()
+
+        # Review only rule_one's iteration
+        Review.objects.create(iteration=iter_one, reviewer=reviewer)
+        iter_one.on_review(request)
+
+        iter_one.refresh_from_db()
+        iter_two.refresh_from_db()
+        self.assertEqual(iter_one.state, ReviewState.REVIEWED)
+        self.assertEqual(iter_two.state, ReviewState.PENDING)
+
+    def test_type_rule_and_object_rule_both_create_iterations(self):
+        """A type-level rule and an object-level rule both create iterations for the same object."""
+        obj = Application.objects.create(name=generate_id(), slug=generate_id())
+        content_type = ContentType.objects.get_for_model(obj)
+
+        object_rule = self._create_rule_for_object(obj, interval="days=30")
+        type_rule = self._create_rule_for_type(Application, interval="days=60")
+
+        iterations = LifecycleIteration.objects.filter(
+            content_type=content_type, object_id=str(obj.pk)
+        )
+        self.assertEqual(iterations.count(), 2)
+        self.assertTrue(iterations.filter(rule=object_rule).exists())
+        self.assertTrue(iterations.filter(rule=type_rule).exists())
+

 class TestLifecycleDateBoundaries(TestCase):
    """Verify that start_of_day normalization ensures correct overdue/due
@@ -679,6 +727,11 @@ class TestLifecycleDateBoundaries(TestCase):
    ensures that the boundary is always at midnight, so millisecond variations
    in task execution time do not affect results."""

+    @classmethod
+    def setUpTestData(cls):
+        config = apps.get_app_config("authentik_tasks_schedules")
+        config._on_startup_callback(None)
+
    def _create_rule_and_iteration(self, grace_period="days=1", interval="days=365"):
        app = Application.objects.create(name=generate_id(), slug=generate_id())
        content_type = ContentType.objects.get_for_model(Application)
--- 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/settings.py
+++ b/authentik/enterprise/settings.py
@@ -14,6 +14,7 @@ TENANT_APPS = [
    "authentik.enterprise.providers.ssf",
    "authentik.enterprise.providers.ws_federation",
    "authentik.enterprise.reports",
+    "authentik.enterprise.stages.account_lockdown",
    "authentik.enterprise.stages.authenticator_endpoint_gdtc",
    "authentik.enterprise.stages.mtls",
    "authentik.enterprise.stages.source",
--- a/packages/django-dramatiq-postgres/django_dramatiq_postgres/management/commands/init.py
+++ b/packages/django-dramatiq-postgres/django_dramatiq_postgres/management/commands/init.py
--- a/authentik/enterprise/stages/account_lockdown/api.py
+++ b/authentik/enterprise/stages/account_lockdown/api.py
@@ -0,0 +1,141 @@
+"""Account Lockdown Stage API Views"""
+
+from django.utils.translation import gettext as _
+from drf_spectacular.utils import OpenApiExample, OpenApiResponse, extend_schema
+from rest_framework.decorators import action
+from rest_framework.exceptions import ValidationError
+from rest_framework.permissions import IsAuthenticated
+from rest_framework.request import Request
+from rest_framework.response import Response
+from rest_framework.serializers import PrimaryKeyRelatedField
+from rest_framework.viewsets import ModelViewSet
+from structlog.stdlib import get_logger
+
+from authentik.api.validation import validate
+from authentik.core.api.used_by import UsedByMixin
+from authentik.core.api.utils import LinkSerializer, PassiveSerializer
+from authentik.core.models import (
+    User,
+)
+from authentik.enterprise.api import EnterpriseRequiredMixin, enterprise_action
+from authentik.enterprise.stages.account_lockdown.models import AccountLockdownStage
+from authentik.enterprise.stages.account_lockdown.stage import (
+    can_lock_user,
+    get_lockdown_target_users,
+)
+from authentik.flows.api.stages import StageSerializer
+from authentik.flows.exceptions import EmptyFlowException, FlowNonApplicableException
+from authentik.flows.planner import PLAN_CONTEXT_PENDING_USER, FlowPlanner
+
+LOGGER = get_logger()
+
+
+class AccountLockdownStageSerializer(EnterpriseRequiredMixin, StageSerializer):
+    """AccountLockdownStage Serializer"""
+
+    class Meta:
+        model = AccountLockdownStage
+        fields = StageSerializer.Meta.fields + [
+            "deactivate_user",
+            "set_unusable_password",
+            "delete_sessions",
+            "revoke_tokens",
+            "self_service_completion_flow",
+        ]
+
+
+class AccountLockdownStageViewSet(UsedByMixin, ModelViewSet):
+    """AccountLockdownStage Viewset"""
+
+    queryset = AccountLockdownStage.objects.all()
+    serializer_class = AccountLockdownStageSerializer
+    filterset_fields = "__all__"
+    ordering = ["name"]
+    search_fields = ["name"]
+
+
+class UserAccountLockdownSerializer(PassiveSerializer):
+    """Choose the target account before starting the lockdown flow."""
+
+    user = PrimaryKeyRelatedField(
+        queryset=get_lockdown_target_users(),
+        required=False,
+        allow_null=True,
+        help_text=_("User to lock. If omitted, locks the current user (self-service)."),
+    )
+
+
+class UserAccountLockdownMixin:
+    """Enterprise account-lockdown API actions for UserViewSet."""
+
+    def _create_lockdown_flow_url(self, request: Request, user: User) -> str:
+        """Create a flow URL for account lockdown.
+
+        The request body selects the target before the flow starts. The API
+        pre-plans the lockdown flow with the target as the pending user, so the
+        account lockdown stage can use the normal flow context.
+        """
+        flow = request._request.brand.flow_lockdown
+        if flow is None:
+            raise ValidationError({"non_field_errors": [_("No lockdown flow configured.")]})
+        planner = FlowPlanner(flow)
+        planner.use_cache = False
+        try:
+            plan = planner.plan(request._request, {PLAN_CONTEXT_PENDING_USER: user})
+        except EmptyFlowException, FlowNonApplicableException:
+            raise ValidationError(
+                {"non_field_errors": [_("Lockdown flow is not applicable.")]}
+            ) from None
+        return plan.to_redirect(request._request, flow).url
+
+    @extend_schema(
+        description=_("Choose the target account, then return a flow link."),
+        request=UserAccountLockdownSerializer,
+        responses={
+            "200": OpenApiResponse(
+                response=LinkSerializer,
+                examples=[
+                    OpenApiExample(
+                        "Lockdown flow URL",
+                        value={
+                            "link": "https://example.invalid/if/flow/default-account-lockdown/",
+                        },
+                        response_only=True,
+                        status_codes=["200"],
+                    )
+                ],
+            ),
+            "400": OpenApiResponse(
+                description=_("No lockdown flow configured or the flow is not applicable")
+            ),
+            "403": OpenApiResponse(
+                description=_("Permission denied (when targeting another user)")
+            ),
+        },
+    )
+    @action(
+        detail=False,
+        methods=["POST"],
+        permission_classes=[IsAuthenticated],
+        url_path="account_lockdown",
+    )
+    @validate(UserAccountLockdownSerializer)
+    @enterprise_action
+    def account_lockdown(self, request: Request, body: UserAccountLockdownSerializer) -> Response:
+        """Trigger account lockdown for a user.
+
+        If no user is specified, locks the current user (self-service).
+        When targeting another user, admin permissions are required.
+
+        Returns a flow link for the frontend to follow. The flow is pre-planned
+        with the target user as pending user for the lockdown stage.
+        """
+        user = body.validated_data.get("user") or request.user
+
+        if not can_lock_user(request.user, user):
+            LOGGER.debug("Permission denied for account lockdown", user=request.user)
+            self.permission_denied(request)
+
+        flow_url = self._create_lockdown_flow_url(request, user)
+        LOGGER.debug("Returning lockdown flow URL", flow_url=flow_url, user=user.username)
+        return Response({"link": flow_url})
--- a/authentik/enterprise/stages/account_lockdown/apps.py
+++ b/authentik/enterprise/stages/account_lockdown/apps.py
@@ -0,0 +1,12 @@
+"""authentik account lockdown stage app config"""
+
+from authentik.enterprise.apps import EnterpriseConfig
+
+
+class AuthentikEnterpriseStageAccountLockdownConfig(EnterpriseConfig):
+    """authentik account lockdown stage config"""
+
+    name = "authentik.enterprise.stages.account_lockdown"
+    label = "authentik_stages_account_lockdown"
+    verbose_name = "authentik Enterprise.Stages.Account Lockdown"
+    default = True
--- a/authentik/enterprise/stages/account_lockdown/migrations/0001_initial.py
+++ b/authentik/enterprise/stages/account_lockdown/migrations/0001_initial.py
@@ -0,0 +1,74 @@
+# Generated by Django 5.2.13 on 2026-04-19 21:56
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+        ("authentik_flows", "0031_alter_flow_layout"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="AccountLockdownStage",
+            fields=[
+                (
+                    "stage_ptr",
+                    models.OneToOneField(
+                        auto_created=True,
+                        on_delete=django.db.models.deletion.CASCADE,
+                        parent_link=True,
+                        primary_key=True,
+                        serialize=False,
+                        to="authentik_flows.stage",
+                    ),
+                ),
+                (
+                    "deactivate_user",
+                    models.BooleanField(
+                        default=True,
+                        help_text="Deactivate the user account (set is_active to False)",
+                    ),
+                ),
+                (
+                    "set_unusable_password",
+                    models.BooleanField(
+                        default=True, help_text="Set an unusable password for the user"
+                    ),
+                ),
+                (
+                    "delete_sessions",
+                    models.BooleanField(
+                        default=True, help_text="Delete all active sessions for the user"
+                    ),
+                ),
+                (
+                    "revoke_tokens",
+                    models.BooleanField(
+                        default=True,
+                        help_text="Revoke all tokens for the user (API, app password, recovery, verification, OAuth)",
+                    ),
+                ),
+                (
+                    "self_service_completion_flow",
+                    models.ForeignKey(
+                        blank=True,
+                        help_text="Flow to redirect users to after self-service lockdown. This flow should not require authentication since the user's session is deleted.",
+                        null=True,
+                        on_delete=django.db.models.deletion.SET_NULL,
+                        related_name="account_lockdown_stages",
+                        to="authentik_flows.flow",
+                    ),
+                ),
+            ],
+            options={
+                "verbose_name": "Account Lockdown Stage",
+                "verbose_name_plural": "Account Lockdown Stages",
+            },
+            bases=("authentik_flows.stage",),
+        ),
+    ]
--- a/authentik/enterprise/stages/account_lockdown/migrations/init.py
+++ b/authentik/enterprise/stages/account_lockdown/migrations/init.py
--- a/authentik/enterprise/stages/account_lockdown/models.py
+++ b/authentik/enterprise/stages/account_lockdown/models.py
@@ -0,0 +1,62 @@
+"""Account lockdown stage models"""
+
+from django.db import models
+from django.utils.translation import gettext_lazy as _
+from django.views import View
+from rest_framework.serializers import BaseSerializer
+
+from authentik.flows.models import Stage
+
+
+class AccountLockdownStage(Stage):
+    """Lock down a target user account."""
+
+    deactivate_user = models.BooleanField(
+        default=True,
+        help_text=_("Deactivate the user account (set is_active to False)"),
+    )
+    set_unusable_password = models.BooleanField(
+        default=True,
+        help_text=_("Set an unusable password for the user"),
+    )
+    delete_sessions = models.BooleanField(
+        default=True,
+        help_text=_("Delete all active sessions for the user"),
+    )
+    revoke_tokens = models.BooleanField(
+        default=True,
+        help_text=_(
+            "Revoke all tokens for the user (API, app password, recovery, verification, OAuth)"
+        ),
+    )
+    self_service_completion_flow = models.ForeignKey(
+        "authentik_flows.Flow",
+        on_delete=models.SET_NULL,
+        null=True,
+        blank=True,
+        related_name="account_lockdown_stages",
+        help_text=_(
+            "Flow to redirect users to after self-service lockdown. "
+            "This flow should not require authentication since the user's session is deleted."
+        ),
+    )
+
+    @property
+    def serializer(self) -> type[BaseSerializer]:
+        from authentik.enterprise.stages.account_lockdown.api import AccountLockdownStageSerializer
+
+        return AccountLockdownStageSerializer
+
+    @property
+    def view(self) -> type[View]:
+        from authentik.enterprise.stages.account_lockdown.stage import AccountLockdownStageView
+
+        return AccountLockdownStageView
+
+    @property
+    def component(self) -> str:
+        return "ak-stage-account-lockdown-form"
+
+    class Meta:
+        verbose_name = _("Account Lockdown Stage")
+        verbose_name_plural = _("Account Lockdown Stages")
--- a/authentik/enterprise/stages/account_lockdown/stage.py
+++ b/authentik/enterprise/stages/account_lockdown/stage.py
@@ -0,0 +1,345 @@
+"""Account lockdown stage logic"""
+
+from django.apps import apps
+from django.core.exceptions import FieldDoesNotExist
+from django.db.models import Model, QuerySet
+from django.db.models.query_utils import Q
+from django.db.transaction import atomic
+from django.http import HttpRequest, HttpResponse, HttpResponseRedirect
+from django.urls import reverse
+from django.utils.translation import gettext_lazy as _
+from dramatiq.actor import Actor
+from dramatiq.composition import group
+from dramatiq.results.errors import ResultTimeout
+
+from authentik.core.models import (
+    AuthenticatedSession,
+    ExpiringModel,
+    Session,
+    Token,
+    User,
+    UserTypes,
+)
+from authentik.enterprise.stages.account_lockdown.models import AccountLockdownStage
+from authentik.events.models import Event, EventAction
+from authentik.flows.stage import StageView
+from authentik.lib.sync.outgoing.models import OutgoingSyncProvider
+from authentik.lib.sync.outgoing.signals import sync_outgoing_inhibit_dispatch
+from authentik.lib.utils.reflection import class_to_path
+from authentik.lib.utils.time import timedelta_from_string
+from authentik.stages.prompt.stage import PLAN_CONTEXT_PROMPT
+
+PLAN_CONTEXT_LOCKDOWN_REASON = "lockdown_reason"
+LOCKDOWN_EVENT_ACTION_ID = "account_lockdown"
+
+TARGET_REQUIRED_MESSAGE = _("No target user specified for account lockdown")
+PERMISSION_DENIED_MESSAGE = _("You do not have permission to lock down this account.")
+ACCOUNT_LOCKDOWN_FAILED_MESSAGE = _("Account lockdown failed for this account.")
+SELF_SERVICE_COMPLETION_FLOW_REQUIRED_MESSAGE = _(
+    "Self-service account lockdown requires a completion flow."
+)
+
+
+def get_lockdown_target_users() -> QuerySet[User]:
+    """Return users that can be targeted by account lockdown."""
+    return User.objects.exclude_anonymous().exclude(type=UserTypes.INTERNAL_SERVICE_ACCOUNT)
+
+
+def _get_model_field(model: type[Model], field_name: str):
+    """Get a model field by name, if present."""
+    try:
+        return model._meta.get_field(field_name)
+    except FieldDoesNotExist:
+        return None
+
+
+def _has_user_field(model: type[Model]) -> bool:
+    """Check if a model has a direct user foreign key."""
+    field = _get_model_field(model, "user")
+    return bool(field and getattr(field, "remote_field", None) and field.remote_field.model is User)
+
+
+def _has_authenticated_session_field(model: type[Model]) -> bool:
+    """Check if a model is linked to an authenticated session."""
+    field = _get_model_field(model, "session")
+    return bool(
+        field
+        and getattr(field, "remote_field", None)
+        and field.remote_field.model is AuthenticatedSession
+    )
+
+
+def _has_provider_field(model: type[Model]) -> bool:
+    """Check if a model is linked to a provider."""
+    return _get_model_field(model, "provider") is not None
+
+
+def get_lockdown_token_models() -> tuple[type[Model], ...]:
+    """Return token, grant, and provider session models removed by account lockdown."""
+    token_models: list[type[Model]] = []
+    for model in apps.get_models():
+        if model._meta.abstract or not issubclass(model, ExpiringModel):
+            continue
+        if model is Token:
+            token_models.append(model)
+        elif _has_user_field(model) and (
+            _has_provider_field(model) or _has_authenticated_session_field(model)
+        ):
+            token_models.append(model)
+        elif _has_authenticated_session_field(model):
+            token_models.append(model)
+    return tuple(token_models)
+
+
+def get_lockdown_token_queryset(model: type[Model], user: User) -> QuerySet:
+    """Return account lockdown artifacts for a model and user."""
+    manager = model.objects.including_expired()
+    if _has_user_field(model):
+        return manager.filter(user=user)
+    return manager.filter(session__user=user)
+
+
+def can_lock_user(actor, user: User) -> bool:
+    """Check whether the actor may lock the target user."""
+    if not actor.is_authenticated:
+        return False
+    if user.pk == actor.pk:
+        return True
+    return actor.has_perm("authentik_core.change_user", user)
+
+
+def get_outgoing_sync_tasks() -> tuple[tuple[type[OutgoingSyncProvider], Actor], ...]:
+    """Return outgoing sync provider types and their direct sync tasks."""
+    from authentik.enterprise.providers.google_workspace.models import GoogleWorkspaceProvider
+    from authentik.enterprise.providers.google_workspace.tasks import google_workspace_sync_direct
+    from authentik.enterprise.providers.microsoft_entra.models import MicrosoftEntraProvider
+    from authentik.enterprise.providers.microsoft_entra.tasks import microsoft_entra_sync_direct
+    from authentik.providers.scim.models import SCIMProvider
+    from authentik.providers.scim.tasks import scim_sync_direct
+
+    return (
+        (SCIMProvider, scim_sync_direct),
+        (GoogleWorkspaceProvider, google_workspace_sync_direct),
+        (MicrosoftEntraProvider, microsoft_entra_sync_direct),
+    )
+
+
+class AccountLockdownStageView(StageView):
+    """Execute account lockdown actions on the target user."""
+
+    def is_self_service(self, request: HttpRequest, user: User) -> bool:
+        """Check whether the currently authenticated user is locking their own account."""
+        return request.user.is_authenticated and user.pk == request.user.pk
+
+    def get_reason(self) -> str:
+        """Get the lockdown reason from the plan context.
+
+        Priority:
+        1. prompt_data[PLAN_CONTEXT_LOCKDOWN_REASON]
+        2. PLAN_CONTEXT_LOCKDOWN_REASON (explicitly set)
+        3. Empty string as fallback
+        """
+        prompt_data = self.executor.plan.context.get(PLAN_CONTEXT_PROMPT, {})
+        if PLAN_CONTEXT_LOCKDOWN_REASON in prompt_data:
+            return prompt_data[PLAN_CONTEXT_LOCKDOWN_REASON]
+        return self.executor.plan.context.get(PLAN_CONTEXT_LOCKDOWN_REASON, "")
+
+    def _apply_lockdown_actions(self, stage: AccountLockdownStage, user: User) -> None:
+        """Apply the configured account changes to the target user."""
+        if stage.deactivate_user:
+            user.is_active = False
+        if stage.set_unusable_password:
+            user.set_unusable_password()
+        if stage.deactivate_user:
+            with sync_outgoing_inhibit_dispatch():
+                user.save()
+            return
+        user.save()
+
+    def _sync_deactivated_user_to_outgoing_providers(self, user: User) -> None:
+        """Synchronize a deactivated user to outgoing sync providers."""
+        messages = []
+        wait_timeout = 0
+        model = class_to_path(User)
+        provider_filter = Q(backchannel_application__isnull=False) | Q(application__isnull=False)
+
+        for provider_model, task_sync_direct in get_outgoing_sync_tasks():
+            for provider in provider_model.objects.filter(provider_filter):
+                time_limit = int(
+                    timedelta_from_string(provider.sync_page_timeout).total_seconds() * 1000
+                )
+                messages.append(
+                    task_sync_direct.message_with_options(
+                        args=(model, user.pk, provider.pk),
+                        rel_obj=provider,
+                        time_limit=time_limit,
+                        uid=f"{provider.name}:user:{user.pk}:direct",
+                    )
+                )
+                wait_timeout += time_limit
+
+        if not messages:
+            return
+        try:
+            group(messages).run().wait(timeout=wait_timeout)
+        except ResultTimeout:
+            self.logger.warning(
+                "Timed out waiting for outgoing sync tasks; tasks remain queued",
+                user=user.username,
+                timeout=wait_timeout,
+            )
+
+    def _get_lockdown_artifact_querysets(
+        self, stage: AccountLockdownStage, user: User
+    ) -> tuple[QuerySet, ...]:
+        """Return the configured sessions and tokens targeted by lockdown."""
+        querysets: list[QuerySet] = []
+        if stage.delete_sessions:
+            querysets.append(Session.objects.filter(authenticatedsession__user=user))
+        if stage.revoke_tokens:
+            querysets.extend(
+                get_lockdown_token_queryset(model, user) for model in get_lockdown_token_models()
+            )
+        return tuple(querysets)
+
+    def _delete_lockdown_artifacts(self, stage: AccountLockdownStage, user: User) -> None:
+        """Delete sessions and tokens selected by the lockdown configuration."""
+        for queryset in self._get_lockdown_artifact_querysets(stage, user):
+            queryset.delete()
+
+    def _has_lockdown_artifacts(self, stage: AccountLockdownStage, user: User) -> bool:
+        """Check whether there are still sessions or tokens to remove."""
+        return any(
+            queryset.exists() for queryset in self._get_lockdown_artifact_querysets(stage, user)
+        )
+
+    def _emit_lockdown_event(self, request: HttpRequest, user: User, reason: str) -> None:
+        """Emit the audit event for a completed lockdown."""
+        # Emit the audit event after the transaction commits. If event creation
+        # fails here, dispatch() would otherwise treat the whole lockdown as
+        # failed even though the account changes have already been committed.
+        try:
+            Event.new(
+                EventAction.USER_WRITE,
+                action_id=LOCKDOWN_EVENT_ACTION_ID,
+                reason=reason,
+                affected_user=user.username,
+            ).from_http(request)
+        except Exception as exc:  # noqa: BLE001
+            # Event emission should not make the lockdown itself fail.
+            self.logger.warning(
+                "Failed to emit account lockdown event",
+                user=user.username,
+                exc=exc,
+            )
+
+    def _lockdown_user(
+        self,
+        request: HttpRequest,
+        stage: AccountLockdownStage,
+        user: User,
+        reason: str,
+    ) -> None:
+        """Execute lockdown actions on a single user."""
+        with atomic():
+            user = User.objects.get(pk=user.pk)
+            self._apply_lockdown_actions(stage, user)
+            self._delete_lockdown_artifacts(stage, user)
+
+        # These additional checks/deletes are done to prevent a timing attack that creates tokens
+        # with a compromised token that is simultaneously being deleted.
+        while self._has_lockdown_artifacts(stage, user):
+            with atomic():
+                self._delete_lockdown_artifacts(stage, user)
+
+        if stage.deactivate_user:
+            try:
+                self._sync_deactivated_user_to_outgoing_providers(user)
+            except Exception as exc:  # noqa: BLE001
+                # Local lockdown has already committed. Provider sync failures
+                # must not reopen access or mark the lockdown itself as failed.
+                self.logger.warning(
+                    "Failed to sync account lockdown deactivation to outgoing providers",
+                    user=user.username,
+                    exc=exc,
+                )
+        self._emit_lockdown_event(request, user, reason)
+
+    def dispatch(self, request: HttpRequest) -> HttpResponse:
+        """Execute account lockdown actions."""
+        self.request = request
+        stage: AccountLockdownStage = self.executor.current_stage
+
+        pending_user = self.get_pending_user()
+        if not pending_user.is_authenticated:
+            self.logger.warning("No target user found for account lockdown")
+            return self.executor.stage_invalid(TARGET_REQUIRED_MESSAGE)
+        user = get_lockdown_target_users().filter(pk=pending_user.pk).first()
+        if user is None:
+            self.logger.warning("Target user is not eligible for account lockdown")
+            return self.executor.stage_invalid(TARGET_REQUIRED_MESSAGE)
+        if not can_lock_user(request.user, user):
+            self.logger.warning(
+                "Permission denied for account lockdown",
+                actor=getattr(request.user, "username", None),
+                target=user.username,
+            )
+            return self.executor.stage_invalid(PERMISSION_DENIED_MESSAGE)
+
+        reason = self.get_reason()
+        self_service = self.is_self_service(request, user)
+        if self_service and stage.delete_sessions and not stage.self_service_completion_flow:
+            self.logger.warning("No completion flow configured for self-service account lockdown")
+            return self.executor.stage_invalid(SELF_SERVICE_COMPLETION_FLOW_REQUIRED_MESSAGE)
+
+        self.logger.info(
+            "Executing account lockdown",
+            user=user.username,
+            reason=reason,
+            self_service=self_service,
+            deactivate_user=stage.deactivate_user,
+            set_unusable_password=stage.set_unusable_password,
+            delete_sessions=stage.delete_sessions,
+            revoke_tokens=stage.revoke_tokens,
+        )
+
+        try:
+            self._lockdown_user(request, stage, user, reason)
+            self.logger.info("Account lockdown completed", user=user.username)
+        except Exception as exc:  # noqa: BLE001
+            # Convert unexpected lockdown errors to a flow-stage failure instead
+            # of leaking an exception through the flow executor.
+            self.logger.warning("Account lockdown failed", user=user.username, exc=exc)
+            return self.executor.stage_invalid(ACCOUNT_LOCKDOWN_FAILED_MESSAGE)
+
+        if self_service:
+            if stage.delete_sessions:
+                return self._self_service_completion_response(request)
+            return self.executor.stage_ok()
+
+        return self.executor.stage_ok()
+
+    def _self_service_completion_response(self, request: HttpRequest) -> HttpResponse:
+        """Redirect to completion flow after self-service lockdown.
+
+        Since all sessions are deleted, the user cannot continue in the flow.
+        Redirect them to an unauthenticated completion flow that shows the
+        lockdown message.
+
+        We use a direct HTTP redirect instead of a challenge because the
+        flow executor's challenge handling may try to access the session
+        which we just deleted.
+        """
+        stage: AccountLockdownStage = self.executor.current_stage
+        completion_flow = stage.self_service_completion_flow
+        if completion_flow:
+            # Flush the current request's session to prevent Django's session
+            # middleware from trying to save a deleted session
+            if hasattr(request, "session"):
+                request.session.flush()
+            redirect_to = reverse(
+                "authentik_core:if-flow",
+                kwargs={"flow_slug": completion_flow.slug},
+            )
+            return HttpResponseRedirect(redirect_to)
+        return self.executor.stage_invalid(SELF_SERVICE_COMPLETION_FLOW_REQUIRED_MESSAGE)
--- a/authentik/enterprise/stages/account_lockdown/tests/init.py
+++ b/authentik/enterprise/stages/account_lockdown/tests/init.py
--- a/authentik/enterprise/stages/account_lockdown/tests/test_api.py
+++ b/authentik/enterprise/stages/account_lockdown/tests/test_api.py
@@ -0,0 +1,148 @@
+"""Test Users Account Lockdown API"""
+
+from json import loads
+from unittest.mock import MagicMock, patch
+from urllib.parse import urlparse
+
+from django.urls import reverse
+from rest_framework.test import APITestCase
+
+from authentik.core.tests.utils import (
+    create_test_brand,
+    create_test_flow,
+    create_test_user,
+)
+from authentik.enterprise.stages.account_lockdown.models import AccountLockdownStage
+from authentik.flows.models import FlowDesignation, FlowStageBinding
+from authentik.flows.planner import PLAN_CONTEXT_PENDING_USER
+from authentik.flows.views.executor import SESSION_KEY_PLAN
+from authentik.lib.generators import generate_id
+
+# Patch for enterprise license check
+patch_license = patch(
+    "authentik.enterprise.models.LicenseUsageStatus.is_valid",
+    MagicMock(return_value=True),
+)
+
+
+@patch_license
+class AccountLockdownAPITestCase(APITestCase):
+    """Shared helpers for account lockdown API tests."""
+
+    def setUp(self) -> None:
+        self.lockdown_flow = create_test_flow(FlowDesignation.STAGE_CONFIGURATION)
+        self.lockdown_stage = AccountLockdownStage.objects.create(name=generate_id())
+        FlowStageBinding.objects.create(
+            target=self.lockdown_flow,
+            stage=self.lockdown_stage,
+            order=0,
+        )
+        self.brand = create_test_brand()
+        self.brand.flow_lockdown = self.lockdown_flow
+        self.brand.save()
+
+    def create_user_with_email(self):
+        """Create a regular user with a unique email address."""
+        user = create_test_user()
+        user.email = f"{generate_id()}@test.com"
+        user.save()
+        return user
+
+    def assert_redirect_targets(self, response, user):
+        """Assert that a response contains a pre-planned lockdown flow link for a user."""
+        self.assertEqual(response.status_code, 200)
+        body = loads(response.content)
+        self.assertIn(self.lockdown_flow.slug, body["link"])
+        self.assertEqual(urlparse(body["link"]).query, "")
+        plan = self.client.session[SESSION_KEY_PLAN]
+        self.assertEqual(plan.context[PLAN_CONTEXT_PENDING_USER].pk, user.pk)
+
+    def assert_no_flow_configured(self, response):
+        """Assert that the API reports a missing lockdown flow."""
+        self.assertEqual(response.status_code, 400)
+        body = loads(response.content)
+        self.assertIn("No lockdown flow configured", body["non_field_errors"][0])
+
+
+@patch_license
+class TestUsersAccountLockdownAPI(AccountLockdownAPITestCase):
+    """Test Users Account Lockdown API"""
+
+    def setUp(self) -> None:
+        super().setUp()
+        self.actor = create_test_user()
+        self.user = self.create_user_with_email()
+
+    def test_account_lockdown_with_change_user_returns_redirect(self):
+        """Test that account lockdown allows users with change_user permission."""
+        self.actor.assign_perms_to_managed_role("authentik_core.change_user", self.user)
+        self.client.force_login(self.actor)
+
+        response = self.client.post(
+            reverse("authentik_api:user-account-lockdown"),
+            data={"user": self.user.pk},
+            format="json",
+        )
+
+        self.assert_redirect_targets(response, self.user)
+
+    def test_account_lockdown_no_flow_configured(self):
+        """Test account lockdown when no flow is configured"""
+        self.brand.flow_lockdown = None
+        self.brand.save()
+        self.actor.assign_perms_to_managed_role("authentik_core.change_user", self.user)
+        self.client.force_login(self.actor)
+
+        response = self.client.post(
+            reverse("authentik_api:user-account-lockdown"),
+            data={"user": self.user.pk},
+            format="json",
+        )
+
+        self.assert_no_flow_configured(response)
+
+    def test_account_lockdown_unauthenticated(self):
+        """Test account lockdown requires authentication"""
+        response = self.client.post(
+            reverse("authentik_api:user-account-lockdown"),
+            data={"user": self.user.pk},
+            format="json",
+        )
+
+        self.assertEqual(response.status_code, 403)
+
+    def test_account_lockdown_without_change_user_denied(self):
+        """Test account lockdown denies users without change_user permission."""
+        self.client.force_login(self.actor)
+
+        response = self.client.post(
+            reverse("authentik_api:user-account-lockdown"),
+            data={"user": self.user.pk},
+            format="json",
+        )
+
+        self.assertEqual(response.status_code, 403)
+
+    def test_account_lockdown_self_returns_redirect(self):
+        """Test successful self-service account lockdown returns a direct redirect."""
+        self.client.force_login(self.user)
+
+        response = self.client.post(
+            reverse("authentik_api:user-account-lockdown"),
+            data={},
+            format="json",
+        )
+
+        self.assert_redirect_targets(response, self.user)
+
+    def test_account_lockdown_self_target_without_change_user_returns_redirect(self):
+        """Test self-service does not require change_user permission."""
+        self.client.force_login(self.user)
+
+        response = self.client.post(
+            reverse("authentik_api:user-account-lockdown"),
+            data={"user": self.user.pk},
+            format="json",
+        )
+
+        self.assert_redirect_targets(response, self.user)
--- a/authentik/enterprise/stages/account_lockdown/tests/test_blueprint.py
+++ b/authentik/enterprise/stages/account_lockdown/tests/test_blueprint.py
@@ -0,0 +1,46 @@
+"""Tests for the packaged account-lockdown blueprint."""
+
+from unittest.mock import patch
+
+from django.test import TransactionTestCase
+
+from authentik.blueprints.models import BlueprintInstance
+from authentik.blueprints.v1.importer import Importer
+from authentik.blueprints.v1.tasks import blueprints_find, check_blueprint_v1_file
+from authentik.enterprise.license import LicenseKey
+from authentik.flows.models import Flow
+
+BLUEPRINT_PATH = "example/flow-default-account-lockdown.yaml"
+
+
+class TestAccountLockdownBlueprint(TransactionTestCase):
+    """Test the packaged account-lockdown blueprint behavior."""
+
+    def test_blueprint_is_not_auto_instantiated(self):
+        """Test the packaged blueprint is opt-in and skipped by discovery."""
+        BlueprintInstance.objects.filter(path=BLUEPRINT_PATH).delete()
+        blueprint = next(item for item in blueprints_find() if item.path == BLUEPRINT_PATH)
+
+        check_blueprint_v1_file(blueprint)
+
+        self.assertFalse(BlueprintInstance.objects.filter(path=BLUEPRINT_PATH).exists())
+
+    def test_blueprint_requires_licensed_context(self):
+        """Test manual import only creates flows when enterprise is licensed."""
+        content = BlueprintInstance(path=BLUEPRINT_PATH).retrieve()
+        license_key = LicenseKey("test", 253402300799, "Test license", 1000, 1000)
+
+        with patch("authentik.enterprise.license.LicenseKey.get_total", return_value=license_key):
+            importer = Importer.from_string(content, {"goauthentik.io/enterprise/licensed": False})
+            valid, logs = importer.validate()
+            self.assertTrue(valid, logs)
+            self.assertTrue(importer.apply())
+            self.assertFalse(Flow.objects.filter(slug="default-account-lockdown").exists())
+            self.assertFalse(Flow.objects.filter(slug="default-account-lockdown-complete").exists())
+
+            importer = Importer.from_string(content, {"goauthentik.io/enterprise/licensed": True})
+            valid, logs = importer.validate()
+            self.assertTrue(valid, logs)
+            self.assertTrue(importer.apply())
+            self.assertTrue(Flow.objects.filter(slug="default-account-lockdown").exists())
+            self.assertTrue(Flow.objects.filter(slug="default-account-lockdown-complete").exists())
--- a/authentik/enterprise/stages/account_lockdown/tests/test_stage.py
+++ b/authentik/enterprise/stages/account_lockdown/tests/test_stage.py
@@ -0,0 +1,627 @@
+"""Account lockdown stage tests"""
+
+import json
+from dataclasses import asdict
+from threading import Event as ThreadEvent
+from threading import Thread
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from django.db import connection
+from django.http import HttpResponse
+from django.test import TransactionTestCase
+from django.urls import reverse
+from django.utils import timezone
+from dramatiq.results.errors import ResultTimeout
+
+from authentik.core.models import AuthenticatedSession, Session, Token, TokenIntents
+from authentik.core.tests.utils import (
+    RequestFactory,
+    create_test_admin_user,
+    create_test_cert,
+    create_test_flow,
+    create_test_user,
+)
+from authentik.enterprise.stages.account_lockdown.models import AccountLockdownStage
+from authentik.enterprise.stages.account_lockdown.stage import (
+    LOCKDOWN_EVENT_ACTION_ID,
+    PLAN_CONTEXT_LOCKDOWN_REASON,
+    AccountLockdownStageView,
+    can_lock_user,
+)
+from authentik.events.models import Event, EventAction
+from authentik.flows.markers import StageMarker
+from authentik.flows.models import FlowDesignation, FlowStageBinding
+from authentik.flows.planner import PLAN_CONTEXT_PENDING_USER, FlowPlan
+from authentik.flows.tests import FlowTestCase
+from authentik.lib.generators import generate_id
+from authentik.lib.utils.reflection import class_to_path
+from authentik.providers.oauth2.id_token import IDToken
+from authentik.providers.oauth2.models import (
+    AccessToken,
+    AuthorizationCode,
+    DeviceToken,
+    OAuth2Provider,
+    RedirectURI,
+    RedirectURIMatchingMode,
+    RefreshToken,
+)
+from authentik.providers.saml.models import SAMLProvider, SAMLSession
+from authentik.stages.prompt.stage import PLAN_CONTEXT_PROMPT
+
+patch_enterprise_enabled = patch(
+    "authentik.enterprise.apps.AuthentikEnterpriseConfig.check_enabled",
+    return_value=True,
+)
+
+
+class AccountLockdownStageTestMixin:
+    """Shared setup helpers for account lockdown stage tests."""
+
+    @classmethod
+    def setUpClass(cls):
+        cls.patch_enterprise_enabled = patch_enterprise_enabled.start()
+        cls.patch_event_dispatch = patch("authentik.events.tasks.event_trigger_dispatch.send")
+        cls.patch_event_dispatch.start()
+        super().setUpClass()
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.patch_event_dispatch.stop()
+        patch_enterprise_enabled.stop()
+        super().tearDownClass()
+
+    def setUp(self):
+        super().setUp()
+        self.user = create_test_admin_user()
+        self.target_user = create_test_admin_user()
+        self.flow = create_test_flow(FlowDesignation.STAGE_CONFIGURATION)
+        self.stage = AccountLockdownStage.objects.create(
+            name="lockdown",
+        )
+        self.binding = FlowStageBinding.objects.create(target=self.flow, stage=self.stage, order=0)
+        self.request_factory = RequestFactory()
+
+    def make_stage_view(self, plan: FlowPlan):
+        def _stage_ok():
+            return HttpResponse(status=204)
+
+        def _stage_invalid(_error_message=None):
+            return HttpResponse(status=400)
+
+        return AccountLockdownStageView(
+            SimpleNamespace(
+                plan=plan,
+                current_stage=self.stage,
+                current_binding=self.binding,
+                flow=self.flow,
+                stage_ok=_stage_ok,
+                stage_invalid=_stage_invalid,
+            )
+        )
+
+    def make_request(self, *, user=None, query=None):
+        return self.request_factory.post(
+            reverse("authentik_api:flow-executor", kwargs={"flow_slug": self.flow.slug}),
+            query_params=query or {},
+            user=user,
+        )
+
+    def get_lockdown_event(self):
+        """Return the account-lockdown user-write event."""
+        return Event.objects.filter(
+            action=EventAction.USER_WRITE,
+            context__action_id=LOCKDOWN_EVENT_ACTION_ID,
+        ).first()
+
+
+class TestAccountLockdownStage(AccountLockdownStageTestMixin, FlowTestCase):
+    """Account lockdown stage tests"""
+
+    def test_lockdown_no_target(self):
+        """Test lockdown stage with no pending user fails"""
+        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
+        view = self.make_stage_view(plan)
+
+        response = view.dispatch(self.make_request())
+
+        self.assertEqual(response.status_code, 400)
+
+    def test_lockdown_with_pending_user(self):
+        """Test lockdown stage with a pending target user."""
+        self.target_user.is_active = True
+        self.target_user.save()
+
+        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
+        plan.context[PLAN_CONTEXT_LOCKDOWN_REASON] = "Security incident"
+        plan.context[PLAN_CONTEXT_PENDING_USER] = self.target_user
+        view = self.make_stage_view(plan)
+        request = self.make_request(user=self.user)
+
+        self.assertTrue(can_lock_user(request.user, self.target_user))
+        response = view.dispatch(request)
+
+        self.target_user.refresh_from_db()
+        self.assertFalse(self.target_user.is_active)
+        self.assertFalse(self.target_user.has_usable_password())
+        self.assertEqual(response.status_code, 204)
+
+        # Check event was created
+        event = self.get_lockdown_event()
+        self.assertIsNotNone(event)
+        self.assertEqual(event.context["action_id"], LOCKDOWN_EVENT_ACTION_ID)
+        self.assertEqual(event.context["reason"], "Security incident")
+        self.assertEqual(event.context["affected_user"], self.target_user.username)
+
+    def test_lockdown_with_pending_user_reason(self):
+        """Test lockdown stage with a pending target and explicit reason."""
+        self.target_user.is_active = True
+        self.target_user.save()
+
+        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
+        plan.context[PLAN_CONTEXT_LOCKDOWN_REASON] = "Compromised account"
+        plan.context[PLAN_CONTEXT_PENDING_USER] = self.target_user
+        view = self.make_stage_view(plan)
+        request = self.make_request(user=self.user)
+
+        self.assertTrue(can_lock_user(request.user, self.target_user))
+        response = view.dispatch(request)
+
+        self.target_user.refresh_from_db()
+        self.assertFalse(self.target_user.is_active)
+        self.assertEqual(response.status_code, 204)
+
+    def test_lockdown_reason_from_prompt(self):
+        """Test lockdown stage reads the reason from prompt data."""
+        self.target_user.is_active = True
+        self.target_user.save()
+
+        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
+        plan.context[PLAN_CONTEXT_PROMPT] = {
+            PLAN_CONTEXT_LOCKDOWN_REASON: "User requested lockdown",
+        }
+        view = self.make_stage_view(plan)
+        request = self.make_request(user=self.user)
+        view._lockdown_user(request, self.stage, self.target_user, view.get_reason())
+
+        event = self.get_lockdown_event()
+        self.assertIsNotNone(event)
+        self.assertEqual(event.context["reason"], "User requested lockdown")
+
+    def test_lockdown_event_failure_does_not_fail_self_service(self):
+        """Test lockdown still succeeds when event emission fails."""
+        self.stage.delete_sessions = False
+        self.stage.save()
+
+        self.target_user.is_active = True
+        self.target_user.save()
+
+        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
+        plan.context[PLAN_CONTEXT_PENDING_USER] = self.target_user
+        view = self.make_stage_view(plan)
+        request = self.make_request(user=self.target_user)
+
+        original_event_new = Event.new
+
+        def _event_new_side_effect(action, *args, **kwargs):
+            if (
+                action == EventAction.USER_WRITE
+                and kwargs.get("action_id") == LOCKDOWN_EVENT_ACTION_ID
+            ):
+                raise RuntimeError("simulated event failure")
+            return original_event_new(action, *args, **kwargs)
+
+        with patch(
+            "authentik.enterprise.stages.account_lockdown.stage.Event.new",
+            side_effect=_event_new_side_effect,
+        ):
+            view._lockdown_user(request, self.stage, self.target_user, view.get_reason())
+
+        self.target_user.refresh_from_db()
+        self.assertFalse(self.target_user.is_active)
+
+    def test_dispatch_records_success_when_event_emission_fails(self):
+        """Test dispatch still completes if event emission fails."""
+        self.stage.delete_sessions = False
+        self.stage.save()
+
+        self.target_user.is_active = True
+        self.target_user.save()
+
+        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
+        plan.context[PLAN_CONTEXT_PENDING_USER] = self.target_user
+        view = self.make_stage_view(plan)
+        request = self.make_request(
+            user=self.target_user,
+        )
+
+        original_event_new = Event.new
+
+        def _event_new_side_effect(action, *args, **kwargs):
+            if (
+                action == EventAction.USER_WRITE
+                and kwargs.get("action_id") == LOCKDOWN_EVENT_ACTION_ID
+            ):
+                raise RuntimeError("simulated event failure")
+            return original_event_new(action, *args, **kwargs)
+
+        with patch(
+            "authentik.enterprise.stages.account_lockdown.stage.Event.new",
+            side_effect=_event_new_side_effect,
+        ):
+            response = view.dispatch(request)
+
+        self.target_user.refresh_from_db()
+        self.assertFalse(self.target_user.is_active)
+        self.assertEqual(response.status_code, 204)
+
+    def test_lockdown_self_service_redirects_to_completion_flow(self):
+        """Test self-service lockdown redirects to completion flow when sessions are deleted."""
+        completion_flow = create_test_flow(FlowDesignation.STAGE_CONFIGURATION)
+        self.stage.self_service_completion_flow = completion_flow
+        self.stage.save()
+
+        self.target_user.is_active = True
+        self.target_user.save()
+
+        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
+        view = self.make_stage_view(plan)
+        request = self.make_request(user=self.target_user)
+        view._lockdown_user(request, self.stage, self.target_user, view.get_reason())
+        response = view._self_service_completion_response(request)
+
+        self.assertEqual(response.status_code, 302)
+        self.assertEqual(
+            response.url,
+            reverse("authentik_core:if-flow", kwargs={"flow_slug": completion_flow.slug}),
+        )
+
+    def test_lockdown_self_service_requires_completion_flow(self):
+        """Test self-service lockdown fails before deleting sessions without a completion flow."""
+        self.stage.self_service_completion_flow = None
+        self.stage.save()
+
+        self.target_user.is_active = True
+        self.target_user.save()
+
+        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
+        plan.context[PLAN_CONTEXT_PENDING_USER] = self.target_user
+        view = self.make_stage_view(plan)
+        request = self.make_request(user=self.target_user)
+
+        response = view.dispatch(request)
+
+        self.assertEqual(response.status_code, 400)
+        self.target_user.refresh_from_db()
+        self.assertTrue(self.target_user.is_active)
+
+    def test_lockdown_denies_other_user_without_permission(self):
+        """Test lockdown stage rejects non-self requests without change_user permission."""
+        actor = create_test_user()
+
+        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
+        plan.context[PLAN_CONTEXT_PENDING_USER] = self.target_user
+        view = self.make_stage_view(plan)
+        request = self.make_request(user=actor)
+
+        self.assertFalse(can_lock_user(request.user, self.target_user))
+        response = view.dispatch(request)
+        self.assertEqual(response.status_code, 400)
+
+    def test_lockdown_revokes_tokens(self):
+        """Test lockdown stage revokes tokens"""
+        Token.objects.create(
+            user=self.target_user,
+            identifier="test-token",
+            intent=TokenIntents.INTENT_API,
+            key=generate_id(),
+            expiring=False,
+        )
+        self.assertEqual(Token.objects.filter(user=self.target_user).count(), 1)
+
+        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
+        view = self.make_stage_view(plan)
+        view._lockdown_user(self.make_request(user=self.user), self.stage, self.target_user, "")
+
+        self.assertEqual(Token.objects.filter(user=self.target_user).count(), 0)
+
+    def test_lockdown_revokes_provider_tokens(self):
+        """Test lockdown stage revokes provider tokens and sessions."""
+        oauth_provider = OAuth2Provider.objects.create(
+            name=generate_id(),
+            authorization_flow=create_test_flow(),
+            redirect_uris=[
+                RedirectURI(RedirectURIMatchingMode.STRICT, "http://testserver/callback")
+            ],
+            signing_key=create_test_cert(),
+        )
+        saml_provider = SAMLProvider.objects.create(
+            name=generate_id(),
+            authorization_flow=create_test_flow(),
+            acs_url="https://sp.example.com/acs",
+            issuer_override="https://idp.example.com",
+        )
+        session = Session.objects.create(
+            session_key=generate_id(),
+            expires=timezone.now() + timezone.timedelta(hours=1),
+            last_ip="127.0.0.1",
+        )
+        auth_session = AuthenticatedSession.objects.create(
+            session=session,
+            user=self.target_user,
+        )
+        grant_kwargs = {
+            "provider": oauth_provider,
+            "user": self.target_user,
+            "auth_time": timezone.now(),
+            "_scope": "openid profile",
+            "expiring": False,
+        }
+        token_kwargs = grant_kwargs | {"_id_token": json.dumps(asdict(IDToken("foo", "bar")))}
+        AuthorizationCode.objects.create(
+            code=generate_id(),
+            session=auth_session,
+            **grant_kwargs,
+        )
+        AccessToken.objects.create(
+            token=generate_id(),
+            session=auth_session,
+            **token_kwargs,
+        )
+        RefreshToken.objects.create(
+            token=generate_id(),
+            session=auth_session,
+            **token_kwargs,
+        )
+        DeviceToken.objects.create(
+            provider=oauth_provider,
+            user=self.target_user,
+            session=auth_session,
+            _scope="openid profile",
+            expiring=False,
+        )
+        SAMLSession.objects.create(
+            provider=saml_provider,
+            user=self.target_user,
+            session=auth_session,
+            session_index=generate_id(),
+            name_id=self.target_user.email,
+            expires=timezone.now() + timezone.timedelta(hours=1),
+            expiring=True,
+        )
+
+        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
+        view = self.make_stage_view(plan)
+        view._lockdown_user(self.make_request(user=self.user), self.stage, self.target_user, "")
+
+        self.assertEqual(AuthorizationCode.objects.filter(user=self.target_user).count(), 0)
+        self.assertEqual(AccessToken.objects.filter(user=self.target_user).count(), 0)
+        self.assertEqual(RefreshToken.objects.filter(user=self.target_user).count(), 0)
+        self.assertEqual(DeviceToken.objects.filter(user=self.target_user).count(), 0)
+        self.assertEqual(SAMLSession.objects.filter(user=self.target_user).count(), 0)
+
+    def test_lockdown_selective_actions(self):
+        """Test lockdown stage with selective actions"""
+        self.stage.deactivate_user = True
+        self.stage.set_unusable_password = False
+        self.stage.delete_sessions = False
+        self.stage.revoke_tokens = False
+        self.stage.save()
+
+        self.target_user.is_active = True
+        self.target_user.set_password("testpassword")
+        self.target_user.save()
+
+        Token.objects.create(
+            user=self.target_user,
+            identifier="test-token",
+            intent=TokenIntents.INTENT_API,
+            key=generate_id(),
+            expiring=False,
+        )
+
+        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
+        view = self.make_stage_view(plan)
+        view._lockdown_user(self.make_request(user=self.user), self.stage, self.target_user, "")
+
+        self.target_user.refresh_from_db()
+        # User should be deactivated
+        self.assertFalse(self.target_user.is_active)
+        # Password should still be usable
+        self.assertTrue(self.target_user.has_usable_password())
+        # Token should still exist
+        self.assertEqual(Token.objects.filter(user=self.target_user).count(), 1)
+
+    def test_lockdown_no_actions(self):
+        """Test lockdown stage with all actions disabled"""
+        self.stage.deactivate_user = False
+        self.stage.set_unusable_password = False
+        self.stage.delete_sessions = False
+        self.stage.revoke_tokens = False
+        self.stage.save()
+
+        self.target_user.is_active = True
+        self.target_user.set_password("testpassword")
+        self.target_user.save()
+
+        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
+        view = self.make_stage_view(plan)
+        view._lockdown_user(self.make_request(user=self.user), self.stage, self.target_user, "")
+
+        self.target_user.refresh_from_db()
+        # User should still be active
+        self.assertTrue(self.target_user.is_active)
+        # Password should still be usable
+        self.assertTrue(self.target_user.has_usable_password())
+        # Event should still be created
+        event = self.get_lockdown_event()
+        self.assertIsNotNone(event)
+
+    def test_lockdown_deactivation_inhibits_signal_dispatch_until_after_commit(self):
+        """Test lockdown queues explicit outgoing syncs after the deactivation transaction."""
+        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
+        view = self.make_stage_view(plan)
+
+        with (
+            patch(
+                "authentik.enterprise.stages.account_lockdown.stage.sync_outgoing_inhibit_dispatch"
+            ) as inhibit,
+            patch.object(view, "_sync_deactivated_user_to_outgoing_providers") as sync_outgoing,
+        ):
+            view._lockdown_user(self.make_request(user=self.user), self.stage, self.target_user, "")
+
+        inhibit.assert_called_once()
+        sync_outgoing.assert_called_once()
+        synced_user = sync_outgoing.call_args.args[0]
+        self.assertEqual(synced_user.pk, self.target_user.pk)
+        self.assertFalse(synced_user.is_active)
+
+    def test_lockdown_waits_for_direct_outgoing_provider_syncs(self):
+        """Test direct outgoing sync tasks are enqueued and waited on."""
+        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
+        view = self.make_stage_view(plan)
+        provider = SimpleNamespace(name="outgoing", pk=1, sync_page_timeout="seconds=5")
+        task_sync_direct = MagicMock()
+        task_sync_direct.message_with_options.return_value = "direct-message"
+        provider_model = SimpleNamespace(
+            objects=SimpleNamespace(filter=MagicMock(return_value=[provider]))
+        )
+        task_group = MagicMock()
+
+        with (
+            patch(
+                "authentik.enterprise.stages.account_lockdown.stage.get_outgoing_sync_tasks",
+                return_value=((provider_model, task_sync_direct),),
+            ),
+            patch(
+                "authentik.enterprise.stages.account_lockdown.stage.group",
+                return_value=task_group,
+            ) as task_group_cls,
+        ):
+            view._sync_deactivated_user_to_outgoing_providers(self.target_user)
+
+        task_sync_direct.message_with_options.assert_called_once_with(
+            args=(class_to_path(type(self.target_user)), self.target_user.pk, provider.pk),
+            rel_obj=provider,
+            time_limit=5000,
+            uid=f"{provider.name}:user:{self.target_user.pk}:direct",
+        )
+        task_group_cls.assert_called_once_with(["direct-message"])
+        task_group.run.return_value.wait.assert_called_once_with(timeout=5000)
+
+    def test_lockdown_outgoing_provider_sync_timeout_leaves_tasks_running(self):
+        """Test timeout while waiting for direct outgoing syncs does not fail lockdown."""
+        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
+        view = self.make_stage_view(plan)
+        provider = SimpleNamespace(name="outgoing", pk=1, sync_page_timeout="seconds=5")
+        task_sync_direct = MagicMock()
+        task_sync_direct.message_with_options.return_value = "direct-message"
+        provider_model = SimpleNamespace(
+            objects=SimpleNamespace(filter=MagicMock(return_value=[provider]))
+        )
+        task_group = MagicMock()
+        task_group.run.return_value.wait.side_effect = ResultTimeout("timed out")
+
+        with (
+            patch(
+                "authentik.enterprise.stages.account_lockdown.stage.get_outgoing_sync_tasks",
+                return_value=((provider_model, task_sync_direct),),
+            ),
+            patch(
+                "authentik.enterprise.stages.account_lockdown.stage.group",
+                return_value=task_group,
+            ),
+        ):
+            view._sync_deactivated_user_to_outgoing_providers(self.target_user)
+
+        task_group.run.assert_called_once_with()
+        task_group.run.return_value.wait.assert_called_once_with(timeout=5000)
+
+    def test_lockdown_outgoing_provider_sync_failure_does_not_fail_lockdown(self):
+        """Test completed local lockdown still emits an event if outgoing sync fails."""
+        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
+        view = self.make_stage_view(plan)
+
+        with patch.object(
+            view,
+            "_sync_deactivated_user_to_outgoing_providers",
+            side_effect=ValueError("sync failed"),
+        ):
+            view._lockdown_user(self.make_request(user=self.user), self.stage, self.target_user, "")
+
+        self.target_user.refresh_from_db()
+        self.assertFalse(self.target_user.is_active)
+        event = self.get_lockdown_event()
+        self.assertIsNotNone(event)
+
+
+class TestAccountLockdownStageConcurrency(AccountLockdownStageTestMixin, TransactionTestCase):
+    """Account lockdown concurrency tests."""
+
+    def test_lockdown_retries_when_another_transaction_recreates_a_token(self):
+        """Lockdown should remove a token recreated before the retry check runs."""
+        Token.objects.create(
+            user=self.target_user,
+            identifier=f"initial-token-{generate_id()}",
+            intent=TokenIntents.INTENT_API,
+            key=generate_id(),
+            expiring=False,
+        )
+
+        plan = FlowPlan(flow_pk=self.flow.pk.hex, bindings=[self.binding], markers=[StageMarker()])
+        view = self.make_stage_view(plan)
+        original_has_artifacts = view._has_lockdown_artifacts
+        target_user = self.target_user
+        thread_ready = ThreadEvent()
+        start_create = ThreadEvent()
+        thread_done = ThreadEvent()
+        thread_errors = []
+
+        class TokenCreatorThread(Thread):
+            __test__ = False
+
+            def run(self):
+                try:
+                    thread_ready.set()
+                    if not start_create.wait(timeout=5):
+                        thread_errors.append("timed out waiting to recreate token")
+                        return
+                    Token.objects.create(
+                        user=target_user,
+                        identifier=f"concurrent-token-{generate_id()}",
+                        intent=TokenIntents.INTENT_API,
+                        key=generate_id(),
+                        expiring=False,
+                    )
+                except Exception as exc:  # noqa: BLE001
+                    thread_errors.append(exc)
+                finally:
+                    thread_done.set()
+                    connection.close()
+
+        def has_artifacts_after_concurrent_create(stage, user):
+            if not start_create.is_set():
+                start_create.set()
+                self.assertTrue(
+                    thread_done.wait(timeout=30),
+                    (
+                        "Concurrent token creation did not complete "
+                        f"before retry check: {thread_errors}"
+                    ),
+                )
+            return original_has_artifacts(stage, user)
+
+        creator = TokenCreatorThread()
+        with patch.object(
+            view, "_has_lockdown_artifacts", side_effect=has_artifacts_after_concurrent_create
+        ):
+            creator.start()
+            self.assertTrue(
+                thread_ready.wait(timeout=5),
+                "Concurrent token creation thread did not start",
+            )
+            view._lockdown_user(self.make_request(user=self.user), self.stage, self.target_user, "")
+            creator.join()
+
+        self.assertEqual(thread_errors, [])
+        self.assertEqual(Token.objects.filter(user=self.target_user).count(), 0)
--- a/authentik/enterprise/stages/account_lockdown/urls.py
+++ b/authentik/enterprise/stages/account_lockdown/urls.py
@@ -0,0 +1,5 @@
+"""API URLs"""
+
+from authentik.enterprise.stages.account_lockdown.api import AccountLockdownStageViewSet
+
+api_urlpatterns = [("stages/account_lockdown", AccountLockdownStageViewSet)]
--- 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/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_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/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/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(
@@ -188,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(
            "/",
@@ -206,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(
@@ -227,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={
@@ -246,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:
@@ -261,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={
@@ -274,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(
@@ -297,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()
@@ -333,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(
@@ -404,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(
@@ -466,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()
@@ -515,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(
@@ -572,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()
@@ -612,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(
@@ -635,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(
@@ -667,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()
@@ -697,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()
@@ -736,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()
@@ -762,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/authentik/providers/oauth2/tests/test_token_cc_standard.py
+++ b/authentik/providers/oauth2/tests/test_token_cc_standard.py
@@ -21,6 +21,7 @@ from authentik.policies.models import PolicyBinding
 from authentik.providers.oauth2.errors import TokenError
 from authentik.providers.oauth2.models import (
    AccessToken,
+    GrantType,
    OAuth2Provider,
    RedirectURI,
    RedirectURIMatchingMode,
@@ -41,6 +42,7 @@ class TestTokenClientCredentialsStandard(OAuthTestCase):
            authorization_flow=create_test_flow(),
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://testserver")],
            signing_key=create_test_cert(),
+            grant_types=[GrantType.CLIENT_CREDENTIALS, GrantType.PASSWORD],
        )
        self.provider.property_mappings.set(ScopeMapping.objects.all())
        self.app = Application.objects.create(name="test", slug="test", provider=self.provider)
--- a/authentik/providers/oauth2/tests/test_token_cc_standard_compat.py
+++ b/authentik/providers/oauth2/tests/test_token_cc_standard_compat.py
@@ -22,6 +22,7 @@ from authentik.core.tests.utils import create_test_admin_user, create_test_cert,
 from authentik.policies.models import PolicyBinding
 from authentik.providers.oauth2.errors import TokenError
 from authentik.providers.oauth2.models import (
+    GrantType,
    OAuth2Provider,
    RedirectURI,
    RedirectURIMatchingMode,
@@ -42,6 +43,7 @@ class TestTokenClientCredentialsStandardCompat(OAuthTestCase):
            authorization_flow=create_test_flow(),
            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://testserver")],
            signing_key=create_test_cert(),
+            grant_types=[GrantType.CLIENT_CREDENTIALS, GrantType.PASSWORD],
        )
        self.provider.property_mappings.set(ScopeMapping.objects.all())
        self.app = Application.objects.create(name="test", slug="test", provider=self.provider)
--- a/Show More
+++ b/Show More