providers/oauth2: OpenID key binding support (#22590)

* Adds OpenID key binding and tests

* Ensures dpop_jkt only set for key_bound scope, only JKT covered claims written to cnf

* Checks scope and jkt for device flow, makes library use more consistant, fixes RSA key size computation

* Fixes linter issues

* Addresses linter errors

* Fixes scopes name in device flow

* Increase test coverage

* Simplify jti cache mechanism

* Increased the JTI replay window to 3 minutes

* Refactoring, simplifying code

* Fix wrong JWK library (working with openpubkey again)

* Fix test, refactor errors

* fix authorize error calls

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

* remove cnf from introspection

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

* remove duplicate code_sha256

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

* fix double parsing of key

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

* re-migrate

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

---------

Signed-off-by: Ethan Heilman <ethan.r.heilman@gmail.com>
Signed-off-by: Jens Langhammer <jens@goauthentik.io>
Co-authored-by: Jens Langhammer <jens@goauthentik.io>

authentik/common/oauth/constants.py

@@ -39,6 +39,7 @@ SCOPE_OPENID = "openid"
 SCOPE_OPENID_PROFILE = "profile"
 SCOPE_OPENID_EMAIL = "email"
 SCOPE_OFFLINE_ACCESS = "offline_access"
+SCOPE_BOUND_KEY = "bound_key"
 
 UI_LOCALES = "ui_locales"
 
@@ -47,6 +48,7 @@ PKCE_METHOD_PLAIN = "plain"
 PKCE_METHOD_S256 = "S256"
 
 TOKEN_TYPE = "Bearer"  # nosec
+JWT_TYPE_DPOP_ID_TOKEN = "dpop+id_token"
 
 SCOPE_AUTHENTIK_API = "goauthentik.io/api"
 

authentik/providers/oauth2/dpop.py

@@ -0,0 +1,270 @@
+"""DPoP (Demonstrating Proof-of-Possession) utils"""
+
+import hashlib
+import re
+import time
+from hmac import compare_digest
+from urllib.parse import urlparse
+
+from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicKey
+from django.core.cache import cache
+from django.db import transaction
+from jwcrypto.jwk import JWK
+from jwt import PyJWK
+from jwt import decode as jwt_decode
+from jwt import decode_complete as jwt_decode_complete
+from jwt.exceptions import InvalidTokenError, PyJWTError
+from structlog.stdlib import get_logger
+
+LOGGER = get_logger()
+
+
+# DPoP JWT type header value
+DPOP_JWT_TYPE = "dpop+jwt"
+
+# Supported asymmetric key types for DPoP
+DPOP_SUPPORTED_KTYS = {"EC", "RSA"}
+
+DPOP_SUPPORTED_EC_CURVES = {"P-256", "P-384", "P-521"}
+
+# RSA key size limits for DPoP (bits)
+DPOP_RSA_MIN_KEY_SIZE = 2048
+DPOP_RSA_MAX_KEY_SIZE = 8192
+
+DPOP_JKT_RE = re.compile(r"^[A-Za-z0-9_-]{43}$")
+
+# Supported asymmetric signature algorithms for DPoP
+DPOP_SUPPORTED_ALGS = {
+    "ES256",
+    "ES384",
+    "ES512",
+    "RS256",
+    "RS384",
+    "RS512",
+    "PS256",
+    "PS384",
+    "PS512",
+}
+
+# Required JWK members per RFC 7638, by key type. These are exactly the
+# members the thumbprint is computed over, so a JWK rebuilt from only these
+# has the same thumbprint and carries no additional user supplied extra
+# claims (kid, alg, use, key_ops, x5c, x5u, jku, or unknown members).
+JWK_REQUIRED_CLAIMS = {
+    "EC": ("crv", "kty", "x", "y"),
+    "RSA": ("e", "kty", "n"),
+}
+
+# Clock skew tolerance in seconds for `iat` validation
+DPOP_IAT_CLOCK_SKEW = 60
+
+# JTI replay protection window in seconds. Must cover the iat validity window
+# (DPOP_IAT_CLOCK_SKEW) so a proof can never fall out of the replay cache
+# while still being iat fresh. To avoid edge cases we use
+# 3 * DPOP_IAT_CLOCK_SKEW rather than 2 * DPOP_IAT_CLOCK_SKEW.
+DPOP_JTI_REPLAY_WINDOW = 3 * DPOP_IAT_CLOCK_SKEW
+
+# Cache key template for tracked JTIs
+CACHE_KEY_DPOP_JTI = "authentik_providers_oauth2_dpop_jti_%s"
+
+
+def jwk_thumbprint(jwk: dict) -> str:
+    """Compute the SHA-256 JWK Thumbprint per RFC 7638"""
+    return JWK(**jwk).thumbprint()
+
+
+def is_valid_jkt(value: str) -> bool:
+    """True if value is a well-formed base64url SHA-256 JWK thumbprint."""
+    return bool(DPOP_JKT_RE.fullmatch(value))
+
+
+def canonical_public_jwk(jwk: dict) -> dict:
+    """Return a JWK containing only the RFC 7638 required public members."""
+    kty = jwk.get("kty")
+    members = JWK_REQUIRED_CLAIMS.get(kty)
+    if members is None:
+        raise DPoPError(f"Cannot canonicalize JWK of type {kty}")
+    missing = [m for m in members if m not in jwk]
+    if missing:
+        raise DPoPError(f"JWK missing required members: {missing}")
+    return {m: jwk[m] for m in members}
+
+
+class DPoPError(Exception):
+    """Raised when DPoP proof validation fails"""
+
+
+class DPoPValidator:
+    """Validates DPoP proof JWTs per RFC 9449 Section 5"""
+
+    def validate(
+        self,
+        dpop_proof: str,
+        expected_htm: str,
+        expected_htu: str,
+        expected_jkt: str | None = None,
+        expected_c_s256: str | None = None,
+    ) -> dict:
+        """Validate a DPoP proof JWT.
+
+        :param dpop_proof: The DPoP proof JWT string
+        :param expected_htm: Expected HTTP method (e.g., "POST")
+        :param expected_htu: Expected token endpoint URL
+        :param expected_jkt: Expected JWK thumbprint (from auth request)
+        :param expected_c_s256: Expected c_s256 hash (of code or device_code)
+        :return: The validated public key JWK dict
+        :raises DPoPError: If validation fails
+        """
+        header = self._extract_header(dpop_proof)
+        jwk, key = self._get_and_validate_jwk(header)
+        jwk = canonical_public_jwk(jwk)
+        alg = self._get_and_validate_alg(header)
+        payload = self._verify_signature(dpop_proof, jwk, alg, key)
+        jti = self._validate_payload_claims(payload, expected_htm, expected_htu)
+        self._check_jti_replay(jti)
+        self._validate_optional_claims(payload, expected_c_s256, expected_jkt, jwk)
+        return jwk
+
+    def _extract_header(self, dpop_proof: str) -> dict:
+        """Extract and return the unverified JOSE header."""
+        try:
+            unverified = jwt_decode_complete(
+                dpop_proof,
+                options={"verify_signature": False, "verify_exp": False, "verify_iat": False},
+            )
+            header = unverified.get("header", {})
+        except PyJWTError as exc:
+            raise DPoPError("Invalid DPoP proof JWT") from exc
+
+        if header.get("typ") != DPOP_JWT_TYPE:
+            raise DPoPError(f"Invalid DPoP typ header: {header.get('typ')}")
+        return header
+
+    def _get_and_validate_jwk(self, header: dict) -> tuple[dict, PyJWK | None]:
+        """Extract jwk from header and validate it.
+
+        Returns the raw jwk dict together with the PyJWK already constructed
+        while validating an RSA key (or None for EC), so `_verify_signature`
+        can reuse it instead of re-parsing the same key material.
+        """
+        jwk = header.get("jwk")
+        if not isinstance(jwk, dict):
+            raise DPoPError("Missing jwk in DPoP header")
+        key = self._validate_jwk(jwk)
+        return jwk, key
+
+    def _get_and_validate_alg(self, header: dict) -> str:
+        """Extract and validate the alg header."""
+        alg = header.get("alg")
+        if not alg:
+            raise DPoPError("Missing alg in DPoP header")
+        if alg not in DPOP_SUPPORTED_ALGS:
+            raise DPoPError(f"Unsupported DPoP algorithm: {alg}")
+        return alg
+
+    def _verify_signature(
+        self, dpop_proof: str, jwk: dict, alg: str, key: PyJWK | None = None
+    ) -> dict:
+        """Verify the DPoP proof signature and return the payload."""
+        try:
+            key = key or PyJWK.from_dict(jwk)
+            return jwt_decode(dpop_proof, key.key, algorithms=[alg], options={"verify_iat": False})
+        except (PyJWTError, InvalidTokenError, TypeError, ValueError) as exc:
+            raise DPoPError("DPoP proof signature verification failed") from exc
+
+    def _validate_payload_claims(self, payload: dict, expected_htm: str, expected_htu: str) -> str:
+        """Validate htm, htu, iat, jti claims. Return the jti value."""
+        if payload.get("htm") != expected_htm:
+            raise DPoPError(f"DPoP htm mismatch: expected {expected_htm}, got {payload.get('htm')}")
+
+        payload_htu = payload.get("htu")
+        if not payload_htu:
+            raise DPoPError("DPoP proof missing htu claim")
+        if not self._htu_matches(payload_htu, expected_htu):
+            raise DPoPError(f"DPoP htu mismatch: expected {expected_htu}, got {payload_htu}")
+
+        iat = payload.get("iat")
+        if not isinstance(iat, int):
+            raise DPoPError("DPoP proof missing or invalid iat claim")
+        now = int(time.time())
+        if abs(now - iat) > DPOP_IAT_CLOCK_SKEW:
+            raise DPoPError("DPoP proof iat outside acceptable clock skew")
+
+        jti = payload.get("jti")
+        if not jti:
+            raise DPoPError("DPoP proof missing jti claim")
+
+        return jti
+
+    def _check_jti_replay(self, jti: str) -> None:
+        """Check if the jti has been seen before (replay protection)."""
+
+        # Only store the hash of the JTI to prevent memory-exhaustion attacks
+        # Recommended by RFC 9449 11.1
+        jti_hash = hashlib.sha256(jti.encode("utf-8")).hexdigest()
+        cache_key = CACHE_KEY_DPOP_JTI % jti_hash
+        # `cache.add()` is atomic set-if-not-exists. Wrap so when the cache
+        # backend is DB-backed (as in tests), a duplicate-key insert does not
+        # break the surrounding transaction.
+        with transaction.atomic():
+            added = cache.add(cache_key, True, timeout=DPOP_JTI_REPLAY_WINDOW)
+        if not added:
+            raise DPoPError("DPoP proof jti replay detected")
+
+    def _validate_optional_claims(
+        self,
+        payload: dict,
+        expected_c_s256: str | None,
+        expected_jkt: str | None,
+        jwk: dict,
+    ) -> None:
+        """Validate optional c_s256 and jkt claims if expected."""
+        if expected_c_s256 is not None:
+            proof_c_s256 = payload.get("c_s256")
+            if proof_c_s256 is None:
+                raise DPoPError("DPoP proof missing required c_s256 claim")
+            if not compare_digest(proof_c_s256, expected_c_s256):
+                raise DPoPError("DPoP proof c_s256 mismatch")
+
+        if expected_jkt is not None:
+            thumbprint = jwk_thumbprint(jwk)
+            if not compare_digest(thumbprint, expected_jkt):
+                raise DPoPError("DPoP proof JWK thumbprint mismatch")
+
+    def _validate_jwk(self, jwk: dict) -> PyJWK | None:
+        """Ensure the JWK is a public asymmetric key without private material.
+
+        Returns the PyJWK constructed while checking the RSA key size, so
+        `_verify_signature` can reuse it instead of re-parsing the same key.
+        """
+        kty = jwk.get("kty")
+        if kty not in DPOP_SUPPORTED_KTYS:
+            raise DPoPError(f"Unsupported JWK kty for DPoP: {kty}")
+
+        key = None
+        if kty == "RSA":
+            key = PyJWK.from_dict(jwk)
+            if isinstance(key.key, RSAPublicKey) and key.key.key_size < DPOP_RSA_MIN_KEY_SIZE:
+                raise DPoPError("RSA key too small for DPoP (minimum 2048 bits)")
+            if isinstance(key.key, RSAPublicKey) and key.key.key_size > DPOP_RSA_MAX_KEY_SIZE:
+                raise DPoPError("RSA key too large for DPoP")
+        elif kty == "EC":
+            crv = jwk.get("crv")
+            if crv not in DPOP_SUPPORTED_EC_CURVES:
+                raise DPoPError(f"Unsupported EC curve for DPoP: {crv}")
+
+        private_fields = {"d", "p", "q", "dp", "dq", "qi"}
+        if any(field in jwk for field in private_fields):
+            raise DPoPError("DPoP JWK must not contain private key material")
+
+        return key
+
+    def _htu_matches(self, proof_htu: str, expected_htu: str) -> bool:
+        """Compare htu values ignoring query string and fragment"""
+        parsed_proof = urlparse(proof_htu)
+        parsed_expected = urlparse(expected_htu)
+        return (
+            parsed_proof.scheme == parsed_expected.scheme
+            and parsed_proof.netloc == parsed_expected.netloc
+            and parsed_proof.path == parsed_expected.path
+        )

authentik/providers/oauth2/errors.py

@@ -311,6 +311,15 @@ class DeviceCodeError(TokenError):
             "still pending and polling should continue, but the interval MUST"
             "be increased by 5 seconds for this and all subsequent requests."
         ),
+        "invalid_dpop_jkt": (
+            'The "dpop_jkt" parameter is not a valid base64url-encoded SHA-256 JWK thumbprint'
+        ),
+        "dpop_jkt_required": (
+            'The "dpop_jkt" parameter is required when the "bound_key" scope is requested'
+        ),
+        "dpop_jkt_not_allowed": (
+            'The "dpop_jkt" parameter must not be set unless the "bound_key" scope is requested'
+        ),
     }
 
     def __init__(self, error: str):

authentik/providers/oauth2/id_token.py

@@ -70,6 +70,8 @@ class IDToken:
     sid: str | None = None
     # JWT ID, https://www.rfc-editor.org/rfc/rfc7519.html#section-4.1.7
     jti: str | None = None
+    # Confirmation JWK, https://datatracker.ietf.org/doc/html/rfc7800#section-3
+    cnf: dict | None = None
 
     claims: dict[str, Any] = field(default_factory=dict)
 
@@ -153,11 +155,13 @@ class IDToken:
     def to_access_token(self, provider: OAuth2Provider, token: BaseGrantModel) -> str:
         """Encode id_token for use as access token, adding fields"""
         final = self.to_dict()
+        # Access tokens remain bearer tokens not DPoP, should not have key-binding cnf
+        final.pop("cnf", None)
         final["azp"] = provider.client_id
         final["uid"] = generate_id()
         final.setdefault("scope", " ".join(token.scope))
         return provider.encode(final)
 
-    def to_jwt(self, provider: OAuth2Provider) -> str:
+    def to_jwt(self, provider: OAuth2Provider, jwt_type: str | None = None) -> str:
         """Shortcut to encode id_token to jwt, signed by self.provider"""
-        return provider.encode(self.to_dict())
+        return provider.encode(self.to_dict(), jwt_type=jwt_type)

authentik/providers/oauth2/migrations/0034_authorizationcode_dpop_jkt_devicetoken_dpop_jkt_and_more.py

@@ -0,0 +1,34 @@
+# Generated by Django 5.2.15 on 2026-07-13 11:14
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("authentik_providers_oauth2", "0033_alter_oauth2provider_grant_types"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="authorizationcode",
+            name="dpop_jkt",
+            field=models.CharField(
+                default=None, max_length=255, null=True, verbose_name="DPoP JWK Thumbprint"
+            ),
+        ),
+        migrations.AddField(
+            model_name="devicetoken",
+            name="dpop_jkt",
+            field=models.CharField(
+                default=None, max_length=255, null=True, verbose_name="DPoP JWK Thumbprint"
+            ),
+        ),
+        migrations.AddField(
+            model_name="refreshtoken",
+            name="dpop_jkt",
+            field=models.CharField(
+                default=None, max_length=255, null=True, verbose_name="DPoP JWK Thumbprint"
+            ),
+        ),
+    ]

authentik/providers/oauth2/models.py

@@ -530,6 +530,9 @@ class AuthorizationCode(InternallyManagedMixin, SerializerModel, ExpiringModel,
     code_challenge_method = models.CharField(
         max_length=255, null=True, verbose_name=_("Code Challenge Method")
     )
+    dpop_jkt = models.CharField(
+        max_length=255, null=True, default=None, verbose_name=_("DPoP JWK Thumbprint")
+    )
 
     class Meta:
         verbose_name = _("Authorization Code")
@@ -609,6 +612,9 @@ class RefreshToken(InternallyManagedMixin, SerializerModel, ExpiringModel, BaseG
 
     token = models.TextField(default=generate_client_secret)
     _id_token = models.TextField(verbose_name=_("ID Token"))
+    dpop_jkt = models.CharField(
+        max_length=255, null=True, default=None, verbose_name=_("DPoP JWK Thumbprint")
+    )
     # Shadow the `session` field from `BaseGrantModel` as we want refresh tokens to persist even
     # when the session is terminated.
     session = models.ForeignKey(
@@ -654,6 +660,9 @@ class DeviceToken(InternallyManagedMixin, ExpiringModel):
     device_code = models.TextField(default=generate_key)
     user_code = models.TextField(default=generate_code_fixed_length)
     _scope = models.TextField(default="", verbose_name=_("Scopes"))
+    dpop_jkt = models.CharField(
+        max_length=255, null=True, default=None, verbose_name=_("DPoP JWK Thumbprint")
+    )
     session = models.ForeignKey(
         AuthenticatedSession, null=True, on_delete=models.SET_DEFAULT, default=None
     )

authentik/providers/oauth2/tests/test_authorize.py

@@ -9,7 +9,12 @@ from django.utils import translation
 from django.utils.timezone import now
 
 from authentik.blueprints.tests import apply_blueprint
-from authentik.common.oauth.constants import SCOPE_OFFLINE_ACCESS, SCOPE_OPENID, TOKEN_TYPE
+from authentik.common.oauth.constants import (
+    SCOPE_BOUND_KEY,
+    SCOPE_OFFLINE_ACCESS,
+    SCOPE_OPENID,
+    TOKEN_TYPE,
+)
 from authentik.core.models import Application
 from authentik.core.tests.utils import create_test_admin_user, create_test_brand, create_test_flow
 from authentik.events.models import Event, EventAction
@@ -819,3 +824,124 @@ class TestAuthorize(OAuthTestCase):
         self.assertEqual(response.status_code, 302)
         plan = self.client.session.get(SESSION_KEY_PLAN)
         self.assertEqual(plan.context[PLAN_CONTEXT_PENDING_USER_IDENTIFIER], "foo")
+
+    @apply_blueprint("system/providers-oauth2.yaml")
+    def test_dpop_jkt_persisted(self):
+        """Test that dpop_jkt is persisted in the authorization code"""
+        flow = create_test_flow()
+        provider = OAuth2Provider.objects.create(
+            name=generate_id(),
+            client_id="test",
+            authorization_flow=flow,
+            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "foo://localhost")],
+            access_code_validity="seconds=100",
+            grant_types=[GrantType.AUTHORIZATION_CODE],
+        )
+        provider.property_mappings.set(
+            ScopeMapping.objects.filter(
+                managed__in=[
+                    "goauthentik.io/providers/oauth2/scope-openid",
+                    "goauthentik.io/providers/oauth2/scope-bound_key",
+                ]
+            )
+        )
+        Application.objects.create(name="app", slug="app", provider=provider)
+        state = generate_id()
+        user = create_test_admin_user()
+        self.client.force_login(user)
+        dpop_jkt = "n4bQgYhMfWWaL-qgxVrQFaO_TxsrC4Is0V1sFbDwCgg"
+        response = self.client.get(
+            reverse("authentik_providers_oauth2:authorize"),
+            data={
+                "response_type": "code",
+                "client_id": "test",
+                "state": state,
+                "redirect_uri": "foo://localhost",
+                "scope": f"{SCOPE_OPENID} {SCOPE_BOUND_KEY}",
+                "dpop_jkt": dpop_jkt,
+            },
+        )
+        code: AuthorizationCode = AuthorizationCode.objects.filter(user=user).first()
+        self.assertIsNotNone(code)
+        self.assertEqual(code.dpop_jkt, dpop_jkt)
+        self.assertEqual(
+            response.url,
+            f"foo://localhost?code={code.code}&state={state}",
+        )
+
+    def test_dpop_jkt_without_bound_key_rejected(self):
+        """dpop_jkt supplied without bound_key scope must be rejected, not 500"""
+        provider = OAuth2Provider.objects.create(
+            name=generate_id(),
+            client_id="test-dpop-nobk",
+            authorization_flow=create_test_flow(),
+            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://local.invalid/Foo")],
+            grant_types=[GrantType.AUTHORIZATION_CODE],
+        )
+        request = self.factory.get(
+            "/",
+            data={
+                "response_type": "code",
+                "client_id": provider.client_id,
+                "redirect_uri": "http://local.invalid/Foo",
+                "scope": SCOPE_OPENID,
+                "dpop_jkt": "n4bQgYhMfWWaL-qgxVrQFaO_TxsrC4Is0V1sFbDwCgg",
+            },
+        )
+        with self.assertRaises(AuthorizeError) as cm:
+            OAuthAuthorizationParams.from_request(request)
+        self.assertEqual(cm.exception.error, "invalid_request")
+
+    @apply_blueprint("system/providers-oauth2.yaml")
+    def test_bound_key_without_dpop_jkt_rejected(self):
+        """bound_key scope without dpop_jkt must be rejected"""
+        provider = OAuth2Provider.objects.create(
+            name=generate_id(),
+            client_id="test-bk-nojkt",
+            authorization_flow=create_test_flow(),
+            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://local.invalid/Foo")],
+            grant_types=[GrantType.AUTHORIZATION_CODE],
+        )
+        provider.property_mappings.set(
+            ScopeMapping.objects.filter(
+                managed__in=[
+                    "goauthentik.io/providers/oauth2/scope-openid",
+                    "goauthentik.io/providers/oauth2/scope-bound_key",
+                ]
+            )
+        )
+        request = self.factory.get(
+            "/",
+            data={
+                "response_type": "code",
+                "client_id": provider.client_id,
+                "redirect_uri": "http://local.invalid/Foo",
+                "scope": f"{SCOPE_OPENID} {SCOPE_BOUND_KEY}",
+            },
+        )
+        with self.assertRaises(AuthorizeError) as cm:
+            OAuthAuthorizationParams.from_request(request)
+        self.assertEqual(cm.exception.error, "invalid_request")
+
+    def test_malformed_dpop_jkt_rejected(self):
+        """dpop_jkt that isn't a 43-char base64url thumbprint must be rejected"""
+        provider = OAuth2Provider.objects.create(
+            name=generate_id(),
+            client_id="test-bad-jkt",
+            authorization_flow=create_test_flow(),
+            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://local.invalid/Foo")],
+            grant_types=[GrantType.AUTHORIZATION_CODE],
+        )
+        request = self.factory.get(
+            "/",
+            data={
+                "response_type": "code",
+                "client_id": provider.client_id,
+                "redirect_uri": "http://local.invalid/Foo",
+                "scope": f"{SCOPE_OPENID} {SCOPE_BOUND_KEY}",
+                "dpop_jkt": "too-short",
+            },
+        )
+        with self.assertRaises(AuthorizeError) as cm:
+            OAuthAuthorizationParams.from_request(request)
+        self.assertEqual(cm.exception.error, "invalid_request")

authentik/providers/oauth2/tests/test_device_backchannel.py

@@ -7,6 +7,7 @@ from urllib.parse import quote
 from django.urls import reverse
 
 from authentik.blueprints.tests import apply_blueprint
+from authentik.common.oauth.constants import SCOPE_BOUND_KEY, SCOPE_OPENID
 from authentik.core.models import Application
 from authentik.core.tests.utils import create_test_flow
 from authentik.lib.generators import generate_id
@@ -181,3 +182,94 @@ class TesOAuth2DeviceBackchannel(OAuthTestCase):
         self.assertEqual(len(token.scope), 2)
         self.assertIn("openid", token.scope)
         self.assertIn("email", token.scope)
+
+    @apply_blueprint("system/providers-oauth2.yaml")
+    def test_dpop_jkt_persisted_in_device_token(self):
+        """Test that dpop_jkt is persisted in the device token."""
+        self.provider.property_mappings.set(
+            ScopeMapping.objects.filter(
+                managed__in=[
+                    "goauthentik.io/providers/oauth2/scope-openid",
+                    "goauthentik.io/providers/oauth2/scope-bound_key",
+                ]
+            )
+        )
+        dpop_jkt = "n4bQgYhMfWWaL-qgxVrQFaO_TxsrC4Is0V1sFbDwCgg"
+        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": f"{SCOPE_OPENID} {SCOPE_BOUND_KEY}", "dpop_jkt": dpop_jkt},
+        )
+        self.assertEqual(res.status_code, 200)
+        body = loads(res.content.decode())
+        token = DeviceToken.objects.filter(device_code=body["device_code"]).first()
+        self.assertIsNotNone(token)
+        self.assertEqual(token.dpop_jkt, dpop_jkt)
+        self.assertIn(SCOPE_BOUND_KEY, token.scope)
+
+    @apply_blueprint("system/providers-oauth2.yaml")
+    def test_device_dpop_jkt_without_bound_key_rejected(self):
+        """Test dpop_jkt without bound_key scope throws HTTP 400 error"""
+        self.provider.property_mappings.set(
+            ScopeMapping.objects.filter(
+                managed__in=["goauthentik.io/providers/oauth2/scope-openid"]
+            )
+        )
+        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": SCOPE_OPENID,
+                "dpop_jkt": "n4bQgYhMfWWaL-qgxVrQFaO_TxsrC4Is0V1sFbDwCgg",
+            },
+        )
+        self.assertEqual(res.status_code, 400)
+        body = res.json()
+        self.assertEqual(body["error"], "dpop_jkt_not_allowed")
+        self.assertIn("dpop_jkt", body["error_description"])
+
+    @apply_blueprint("system/providers-oauth2.yaml")
+    def test_device_bound_key_without_dpop_jkt_rejected(self):
+        """bound_key scope without dpop_jkt must 400"""
+        self.provider.property_mappings.set(
+            ScopeMapping.objects.filter(
+                managed__in=[
+                    "goauthentik.io/providers/oauth2/scope-openid",
+                    "goauthentik.io/providers/oauth2/scope-bound_key",
+                ]
+            )
+        )
+        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": f"{SCOPE_OPENID} {SCOPE_BOUND_KEY}"},
+        )
+        self.assertEqual(res.status_code, 400)
+        body = res.json()
+        self.assertEqual(body["error"], "dpop_jkt_required")
+        self.assertIn("dpop_jkt", body["error_description"])
+
+    @apply_blueprint("system/providers-oauth2.yaml")
+    def test_device_malformed_dpop_jkt_rejected(self):
+        """Malformed dpop_jkt must 400"""
+        self.provider.property_mappings.set(
+            ScopeMapping.objects.filter(
+                managed__in=[
+                    "goauthentik.io/providers/oauth2/scope-openid",
+                    "goauthentik.io/providers/oauth2/scope-bound_key",
+                ]
+            )
+        )
+        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": f"{SCOPE_OPENID} {SCOPE_BOUND_KEY}", "dpop_jkt": "nope"},
+        )
+        self.assertEqual(res.status_code, 400)
+        body = res.json()
+        self.assertEqual(body["error"], "invalid_dpop_jkt")
+        self.assertIn("dpop_jkt", body["error_description"])

authentik/providers/oauth2/tests/test_dpop.py

@@ -0,0 +1,506 @@
+"""Test DPoP validation utilities"""
+
+import base64
+import hashlib
+import json
+import time
+
+from cryptography.hazmat.primitives.asymmetric.ec import (
+    SECP256K1,
+    SECP256R1,
+    EllipticCurvePrivateKey,
+    generate_private_key,
+)
+from cryptography.hazmat.primitives.asymmetric.rsa import generate_private_key as generate_rsa_key
+from django.core.cache import cache
+from django.test import TestCase
+from jwt import encode as jwt_encode
+
+from authentik.providers.oauth2.dpop import (
+    DPoPError,
+    DPoPValidator,
+    jwk_thumbprint,
+)
+from authentik.providers.oauth2.utils import pkce_s256_challenge
+
+
+class DPoPProofBuilder:
+    """Helper to build DPoP proof JWTs for testing"""
+
+    def __init__(self, private_key: EllipticCurvePrivateKey | None = None):
+        if private_key is None:
+            private_key = generate_private_key(SECP256R1())
+        self.private_key = private_key
+        self.public_key = private_key.public_key()
+        nums = self.public_key.public_numbers()
+        self.x = base64.urlsafe_b64encode(nums.x.to_bytes(32, "big")).rstrip(b"=").decode()
+        self.y = base64.urlsafe_b64encode(nums.y.to_bytes(32, "big")).rstrip(b"=").decode()
+        self.jwk = {"kty": "EC", "crv": "P-256", "x": self.x, "y": self.y}
+
+    def build(  # noqa: PLR0913
+        self,
+        htm: str = "POST",
+        htu: str = "https://server.example.com/token",
+        c_s256: str | None = None,
+        iat: int | None = None,
+        jti: str = "test-jti-001",
+        alg: str = "ES256",
+        typ: str = "dpop+jwt",
+        include_private: bool = False,
+    ) -> str:
+        headers = {"typ": typ, "jwk": self.jwk.copy(), "alg": alg}
+        if include_private:
+            # Add a fake private key component to test rejection
+            headers["jwk"]["d"] = "fake-private-key"
+
+        payload = {
+            "htm": htm,
+            "htu": htu,
+            "iat": iat if iat is not None else int(time.time()),
+            "jti": jti,
+        }
+        if c_s256 is not None:
+            payload["c_s256"] = c_s256
+
+        return jwt_encode(payload, self.private_key, algorithm=alg, headers=headers)
+
+    def make_header(self, htu: str, c_s256: str | None = None) -> str:
+        """Build a DPoP proof header for the given token endpoint"""
+        return self.build(htm="POST", htu=htu, c_s256=c_s256)
+
+    @property
+    def jkt(self) -> str:
+        return jwk_thumbprint(self.jwk)
+
+
+def _craft_jwt(payload: dict, private_key, algorithm: str, headers: dict) -> str:
+    """Craft a JWT manually, bypassing PyJWT's algorithm validation.
+
+    This allows creating JWTs where the header claims a different algorithm
+    than what was used to sign, for testing rejection of such proofs.
+    """
+    from cryptography.hazmat.primitives import hashes
+    from cryptography.hazmat.primitives.asymmetric import ec, padding
+
+    header_bytes = json.dumps(headers, separators=(",", ":")).encode()
+    payload_bytes = json.dumps(payload, separators=(",", ":")).encode()
+    segments = [
+        base64.urlsafe_b64encode(header_bytes).rstrip(b"="),
+        base64.urlsafe_b64encode(payload_bytes).rstrip(b"="),
+    ]
+    signing_input = b".".join(segments)
+
+    if algorithm == "ES256":
+        signature = private_key.sign(signing_input, ec.ECDSA(hashes.SHA256()))
+    elif algorithm == "ES384":
+        signature = private_key.sign(signing_input, ec.ECDSA(hashes.SHA384()))
+    elif algorithm == "ES512":
+        signature = private_key.sign(signing_input, ec.ECDSA(hashes.SHA512()))
+    elif algorithm in ("RS256", "PS256"):
+        signature = private_key.sign(
+            signing_input,
+            (
+                padding.PSS(mgf=padding.MGF1(hashes.SHA256()), salt_length=padding.PSS.MAX_LENGTH)
+                if algorithm.startswith("PS")
+                else padding.PKCS1v15()
+            ),
+            hashes.SHA256(),
+        )
+    else:
+        raise ValueError(f"Unsupported test algorithm: {algorithm}")
+
+    segments.append(base64.urlsafe_b64encode(signature).rstrip(b"="))
+    return b".".join(segments).decode()
+
+
+def _b64_uint(value: int) -> str:
+    """Encode an unsigned integer as a base64url string (no padding)."""
+    length = (value.bit_length() + 7) // 8
+    return base64.urlsafe_b64encode(value.to_bytes(length, "big")).rstrip(b"=").decode()
+
+
+class RSAProofBuilder:
+    """Helper to build RSA-signed DPoP proof JWTs for testing"""
+
+    def __init__(self, key_size: int = 2048):
+        self.private_key = generate_rsa_key(public_exponent=65537, key_size=key_size)
+        nums = self.private_key.public_key().public_numbers()
+        self.jwk = {"kty": "RSA", "n": _b64_uint(nums.n), "e": _b64_uint(nums.e)}
+
+    def build(
+        self,
+        htm: str = "POST",
+        htu: str = "https://server.example.com/token",
+        c_s256: str | None = None,
+        iat: int | None = None,
+        jti: str = "rsa-jti-001",
+        alg: str = "RS256",
+    ) -> str:
+        headers = {"typ": "dpop+jwt", "jwk": self.jwk.copy(), "alg": alg}
+        payload = {
+            "htm": htm,
+            "htu": htu,
+            "iat": iat if iat is not None else int(time.time()),
+            "jti": jti,
+        }
+        if c_s256 is not None:
+            payload["c_s256"] = c_s256
+        return jwt_encode(payload, self.private_key, algorithm=alg, headers=headers)
+
+    @property
+    def jkt(self) -> str:
+        return jwk_thumbprint(self.jwk)
+
+
+class TestJWKThumbprint(TestCase):
+    """Test JWK thumbprint computation"""
+
+    def test_ec_thumbprint(self):
+        """Test EC P-256 thumbprint computation."""
+        builder = DPoPProofBuilder()
+        thumbprint = jwk_thumbprint(builder.jwk)
+        self.assertIsInstance(thumbprint, str)
+        self.assertGreater(len(thumbprint), 0)
+        # Verify consistency
+        self.assertEqual(thumbprint, jwk_thumbprint(builder.jwk))
+
+    def test_rsa_thumbprint(self):
+        """Test RSA thumbprint computation."""
+        from cryptography.hazmat.primitives.asymmetric.rsa import generate_private_key
+
+        private_key = generate_private_key(public_exponent=65537, key_size=2048)
+        public_key = private_key.public_key()
+        n = (
+            base64.urlsafe_b64encode(public_key.public_numbers().n.to_bytes(256, "big"))
+            .rstrip(b"=")
+            .decode()
+        )
+        e = (
+            base64.urlsafe_b64encode(public_key.public_numbers().e.to_bytes(3, "big"))
+            .rstrip(b"=")
+            .decode()
+        )
+        jwk = {"kty": "RSA", "n": n, "e": e}
+        thumbprint = jwk_thumbprint(jwk)
+        self.assertIsInstance(thumbprint, str)
+        self.assertGreater(len(thumbprint), 0)
+
+    def test_unsupported_kty(self):
+        """Test unsupported key type — jwcrypto may accept it but DPoP validator rejects it"""
+        validator = DPoPValidator()
+        with self.assertRaises(DPoPError):
+            validator._validate_jwk({"kty": "oct", "k": "foo"})
+
+
+class TestComputeCS256(TestCase):
+    """Test c_s256 computation"""
+
+    def test_code_sha256(self):
+        """Test c_s256 matches expected BASE64URL(SHA256(ASCII(value)))"""
+        value = "SplxlOBeZQQYbYS6WxSbIA"
+        expected = (
+            base64.urlsafe_b64encode(hashlib.sha256(value.encode("ascii")).digest())
+            .rstrip(b"=")
+            .decode("ascii")
+        )
+        self.assertEqual(pkce_s256_challenge(value), expected)
+
+    def test_code_sha256_empty(self):
+        """Test c_s256 with empty string"""
+        expected = (
+            base64.urlsafe_b64encode(hashlib.sha256(b"").digest()).rstrip(b"=").decode("ascii")
+        )
+        self.assertEqual(pkce_s256_challenge(""), expected)
+
+
+class TestDPoPValidator(TestCase):
+    """Test DPoP proof validation."""
+
+    def setUp(self):
+        self.validator = DPoPValidator()
+        self.builder = DPoPProofBuilder()
+        self.htu = "https://server.example.com/token"
+        cache.clear()
+
+    def tearDown(self):
+        cache.clear()
+
+    def test_valid_proof(self):
+        """Test a completely valid DPoP proof"""
+        c_s256 = pkce_s256_challenge("test-code")
+        proof = self.builder.build(c_s256=c_s256)
+        result = self.validator.validate(
+            proof,
+            expected_htm="POST",
+            expected_htu=self.htu,
+            expected_jkt=self.builder.jkt,
+            expected_c_s256=c_s256,
+        )
+        self.assertEqual(result["kty"], "EC")
+        self.assertEqual(result["crv"], "P-256")
+
+    def test_valid_proof_without_expected_jkt(self):
+        """Test valid proof without expected_jkt check"""
+        proof = self.builder.build(htu=self.htu)
+        result = self.validator.validate(proof, expected_htm="POST", expected_htu=self.htu)
+        self.assertEqual(result["kty"], "EC")
+
+    def test_invalid_signature(self):
+        """Test proof signed with different key than embedded jwk"""
+        other_builder = DPoPProofBuilder()
+        # Sign with other key but embed original jwk
+        payload = {
+            "htm": "POST",
+            "htu": self.htu,
+            "iat": int(time.time()),
+            "jti": "test-jti-002",
+        }
+        proof = jwt_encode(
+            payload,
+            other_builder.private_key,
+            algorithm="ES256",
+            headers={"typ": "dpop+jwt", "jwk": self.builder.jwk},
+        )
+        with self.assertRaises(DPoPError) as cm:
+            self.validator.validate(proof, expected_htm="POST", expected_htu=self.htu)
+        self.assertIn("signature", str(cm.exception).lower())
+
+    def test_missing_typ_header(self):
+        """Test rejection when typ header is not dpop+jwt"""
+        proof = self.builder.build(typ="jwt")
+        with self.assertRaises(DPoPError) as cm:
+            self.validator.validate(proof, expected_htm="POST", expected_htu=self.htu)
+        self.assertIn("typ", str(cm.exception).lower())
+
+    def test_private_key_material(self):
+        """Test rejection when jwk contains private key material"""
+        proof = self.builder.build(include_private=True)
+        with self.assertRaises(DPoPError) as cm:
+            self.validator.validate(proof, expected_htm="POST", expected_htu=self.htu)
+        self.assertIn("private", str(cm.exception).lower())
+
+    def test_htm_mismatch(self):
+        """Test rejection when htm doesn't match"""
+        proof = self.builder.build(htm="GET", htu=self.htu)
+        with self.assertRaises(DPoPError) as cm:
+            self.validator.validate(proof, expected_htm="POST", expected_htu=self.htu)
+        self.assertIn("htm", str(cm.exception).lower())
+
+    def test_htu_mismatch(self):
+        """Test rejection when htu doesn't match"""
+        proof = self.builder.build(htu="https://other.example.com/token")
+        with self.assertRaises(DPoPError) as cm:
+            self.validator.validate(proof, expected_htm="POST", expected_htu=self.htu)
+        self.assertIn("htu", str(cm.exception).lower())
+
+    def test_htu_ignores_query_and_fragment(self):
+        """Test htu matching ignores query and fragment"""
+        proof = self.builder.build(htu=self.htu + "?foo=bar#baz")
+        result = self.validator.validate(proof, expected_htm="POST", expected_htu=self.htu)
+        self.assertEqual(result["kty"], "EC")
+
+    def test_expired_iat(self):
+        """Test rejection when iat is too old"""
+        proof = self.builder.build(htu=self.htu, iat=int(time.time()) - 120)
+        with self.assertRaises(DPoPError) as cm:
+            self.validator.validate(proof, expected_htm="POST", expected_htu=self.htu)
+        self.assertIn("iat", str(cm.exception).lower())
+
+    def test_future_iat(self):
+        """Test rejection when iat is in the future"""
+        proof = self.builder.build(htu=self.htu, iat=int(time.time()) + 120)
+        with self.assertRaises(DPoPError) as cm:
+            self.validator.validate(proof, expected_htm="POST", expected_htu=self.htu)
+        self.assertIn("iat", str(cm.exception).lower())
+
+    def test_missing_jti(self):
+        """Test rejection when jti is missing"""
+        proof = self.builder.build(htu=self.htu, jti="")
+        with self.assertRaises(DPoPError) as cm:
+            self.validator.validate(proof, expected_htm="POST", expected_htu=self.htu)
+        self.assertIn("jti", str(cm.exception).lower())
+
+    def test_expected_jkt_mismatch(self):
+        """Test rejection when JWK thumbprint doesn't match expected_jkt"""
+        other_builder = DPoPProofBuilder()
+        proof = other_builder.build(htu=self.htu)
+        with self.assertRaises(DPoPError) as cm:
+            self.validator.validate(
+                proof, expected_htm="POST", expected_htu=self.htu, expected_jkt=self.builder.jkt
+            )
+        self.assertIn("thumbprint", str(cm.exception).lower())
+
+    def test_expected_jkt_match(self):
+        """Test acceptance when JWK thumbprint matches expected_jkt"""
+        proof = self.builder.build(htu=self.htu)
+        result = self.validator.validate(
+            proof, expected_htm="POST", expected_htu=self.htu, expected_jkt=self.builder.jkt
+        )
+        self.assertEqual(result["kty"], "EC")
+
+    def test_c_s256_mismatch(self):
+        """Test rejection when c_s256 doesn't match"""
+        proof = self.builder.build(c_s256="wrong-hash")
+        with self.assertRaises(DPoPError) as cm:
+            self.validator.validate(
+                proof,
+                expected_htm="POST",
+                expected_htu=self.htu,
+                expected_c_s256=pkce_s256_challenge("correct-code"),
+            )
+        self.assertIn("c_s256", str(cm.exception).lower())
+
+    def test_c_s256_match(self):
+        """Test acceptance when c_s256 matches"""
+        code = "test-auth-code-123"
+        c_s256 = pkce_s256_challenge(code)
+        proof = self.builder.build(c_s256=c_s256)
+        result = self.validator.validate(
+            proof, expected_htm="POST", expected_htu=self.htu, expected_c_s256=c_s256
+        )
+        self.assertEqual(result["kty"], "EC")
+
+    def test_c_s256_missing_when_required(self):
+        """Test rejection when c_s256 param provided but claim absent"""
+        proof = self.builder.build(htu=self.htu)
+        with self.assertRaises(DPoPError) as cm:
+            self.validator.validate(
+                proof,
+                expected_htm="POST",
+                expected_htu=self.htu,
+                expected_c_s256=pkce_s256_challenge("some-code"),
+            )
+        self.assertIn("c_s256", str(cm.exception).lower())
+
+    def test_symmetric_key_rejected(self):
+        """Test rejection of symmetric keys"""
+        with self.assertRaises(DPoPError) as cm:
+            self.validator._validate_jwk({"kty": "oct", "k": "foo"})
+        self.assertIn("Unsupported", str(cm.exception))
+
+    def test_invalid_jwt(self):
+        """Test rejection of malformed JWT"""
+        with self.assertRaises(DPoPError):
+            self.validator.validate("not-a-jwt", expected_htm="POST", expected_htu=self.htu)
+
+    def test_missing_jwk(self):
+        """Test rejection when jwk is missing from header"""
+        payload = {
+            "htm": "POST",
+            "htu": self.htu,
+            "iat": int(time.time()),
+            "jti": "test-jti-004",
+        }
+        proof = jwt_encode(
+            payload,
+            self.builder.private_key,
+            algorithm="ES256",
+            headers={"typ": "dpop+jwt"},
+        )
+        with self.assertRaises(DPoPError) as cm:
+            self.validator.validate(proof, expected_htm="POST", expected_htu=self.htu)
+        self.assertIn("jwk", str(cm.exception).lower())
+
+    def test_cnf_strips_extra_jwk_members(self):
+        """Test that any claims in the JWK not covered by the JKT are removed"""
+        builder = DPoPProofBuilder()
+        builder.jwk["jku"] = "https://attacker.example/keys"
+        builder.jwk["alg"] = "RS256"
+        result = self.validator.validate(
+            builder.build(htu=self.htu), expected_htm="POST", expected_htu=self.htu
+        )
+        self.assertEqual(set(result), {"crv", "kty", "x", "y"})
+
+    def test_jti_replay_rejected(self):
+        """Test that reusing the same jti is rejected"""
+        jti = "unique-jti-12345"
+        proof = self.builder.build(htu=self.htu, jti=jti)
+        # First use should succeed
+        self.validator.validate(proof, expected_htm="POST", expected_htu=self.htu)
+        # Second use with same jti should fail
+        with self.assertRaises(DPoPError) as cm:
+            self.validator.validate(proof, expected_htm="POST", expected_htu=self.htu)
+        self.assertIn("replay", str(cm.exception).lower())
+
+    def test_jti_different_accepted(self):
+        """Test that different jti values are both accepted"""
+        proof1 = self.builder.build(htu=self.htu, jti="jti-first")
+        proof2 = self.builder.build(htu=self.htu, jti="jti-second")
+        self.validator.validate(proof1, expected_htm="POST", expected_htu=self.htu)
+        self.validator.validate(proof2, expected_htm="POST", expected_htu=self.htu)
+
+    def test_symmetric_alg_rejected(self):
+        """Test rejection when alg header claims a symmetric algorithm"""
+        payload = {
+            "htm": "POST",
+            "htu": self.htu,
+            "iat": int(time.time()),
+            "jti": "test-jti-sym",
+        }
+        # Craft a JWT with HS256 in the header but signed with EC key
+        proof = _craft_jwt(
+            payload,
+            self.builder.private_key,
+            algorithm="ES256",
+            headers={"typ": "dpop+jwt", "jwk": self.builder.jwk, "alg": "HS256"},
+        )
+        with self.assertRaises(DPoPError) as cm:
+            self.validator.validate(proof, expected_htm="POST", expected_htu=self.htu)
+        self.assertIn("algorithm", str(cm.exception).lower())
+
+    def test_alg_jwk_mismatch_rejected(self):
+        """Test rejection when alg doesn't match JWK key type"""
+        payload = {
+            "htm": "POST",
+            "htu": self.htu,
+            "iat": int(time.time()),
+            "jti": "test-jti-mismatch",
+        }
+        # Craft a JWT with RS256 in the header but EC jwk and signature
+        proof = _craft_jwt(
+            payload,
+            self.builder.private_key,
+            algorithm="ES256",
+            headers={"typ": "dpop+jwt", "jwk": self.builder.jwk, "alg": "RS256"},
+        )
+        with self.assertRaises(DPoPError) as cm:
+            self.validator.validate(proof, expected_htm="POST", expected_htu=self.htu)
+        self.assertIn("signature", str(cm.exception).lower())
+
+    def test_rsa_proof_accepted(self):
+        """A valid 2048-bit RSA DPoP proof should validate."""
+        builder = RSAProofBuilder(key_size=2048)
+        proof = builder.build(htu=self.htu)
+        result = self.validator.validate(proof, expected_htm="POST", expected_htu=self.htu)
+        self.assertEqual(result["kty"], "RSA")
+
+    def test_rsa_key_too_small_rejected(self):
+        """An RSA key below the minimum size (2048 bits) must be rejected."""
+        builder = RSAProofBuilder(key_size=1024)
+        proof = builder.build(htu=self.htu)
+        with self.assertRaises(DPoPError) as cm:
+            self.validator.validate(proof, expected_htm="POST", expected_htu=self.htu)
+        self.assertIn("too small", str(cm.exception).lower())
+
+    def test_unsupported_ec_curve_rejected(self):
+        """An EC key on a curve outside the supported set must be rejected."""
+        priv = generate_private_key(SECP256K1())
+        nums = priv.public_key().public_numbers()
+        x = base64.urlsafe_b64encode(nums.x.to_bytes(32, "big")).rstrip(b"=").decode()
+        y = base64.urlsafe_b64encode(nums.y.to_bytes(32, "big")).rstrip(b"=").decode()
+        jwk = {"kty": "EC", "crv": "secp256k1", "x": x, "y": y}
+        payload = {
+            "htm": "POST",
+            "htu": self.htu,
+            "iat": int(time.time()),
+            "jti": "curve-jti-001",
+        }
+        proof = _craft_jwt(
+            payload,
+            priv,
+            algorithm="ES256",
+            headers={"typ": "dpop+jwt", "jwk": jwk, "alg": "ES256"},
+        )
+        with self.assertRaises(DPoPError) as cm:
+            self.validator.validate(proof, expected_htm="POST", expected_htu=self.htu)
+        self.assertIn("curve", str(cm.exception).lower())

authentik/providers/oauth2/tests/test_key_binding_auth_code.py

@@ -0,0 +1,316 @@
+"""Test OpenID Connect Key Binding for Authorization Code flow"""
+
+from base64 import b64encode
+from json import dumps, loads
+
+from django.test import RequestFactory
+from django.urls import reverse
+from django.utils import timezone
+from jwt import decode as jwt_decode
+from jwt import decode_complete as jwt_decode_complete
+
+from authentik.blueprints.tests import apply_blueprint
+from authentik.common.oauth.constants import (
+    GRANT_TYPE_AUTHORIZATION_CODE,
+    GRANT_TYPE_REFRESH_TOKEN,
+    JWT_TYPE_DPOP_ID_TOKEN,
+    SCOPE_BOUND_KEY,
+    SCOPE_OFFLINE_ACCESS,
+    SCOPE_OPENID,
+)
+from authentik.core.models import Application
+from authentik.core.tests.utils import create_test_admin_user, create_test_flow
+from authentik.lib.generators import generate_id
+from authentik.providers.oauth2.models import (
+    AuthorizationCode,
+    GrantType,
+    OAuth2Provider,
+    RedirectURI,
+    RedirectURIMatchingMode,
+    RefreshToken,
+    ScopeMapping,
+)
+from authentik.providers.oauth2.tests.test_dpop import DPoPProofBuilder
+from authentik.providers.oauth2.tests.utils import OAuthTestCase
+from authentik.providers.oauth2.utils import pkce_s256_challenge
+
+
+class TestKeyBindingAuthCode(OAuthTestCase):
+    """Test key-bound ID Tokens in authorization code flow"""
+
+    @apply_blueprint("system/providers-oauth2.yaml")
+    def setUp(self) -> None:
+        super().setUp()
+        self.factory = RequestFactory()
+        self.provider = OAuth2Provider.objects.create(
+            name=generate_id(),
+            authorization_flow=create_test_flow(),
+            grant_types=[GrantType.AUTHORIZATION_CODE, GrantType.REFRESH_TOKEN],
+            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://local.invalid")],
+            signing_key=self.keypair,
+        )
+        self.provider.property_mappings.set(ScopeMapping.objects.all())
+        self.app = Application.objects.create(
+            name=generate_id(), slug="test", provider=self.provider
+        )
+        self.user = create_test_admin_user()
+        self.dpop_builder = DPoPProofBuilder()
+        self.token_url = "http://testserver/application/o/token/"
+
+    def test_successful_key_binding(self):
+        """Valid bound_key + DPoP = ID Token has cnf and typ dpop+id_token"""
+        code = AuthorizationCode.objects.create(
+            code="foobar",
+            provider=self.provider,
+            user=self.user,
+            auth_time=timezone.now(),
+            scope=[SCOPE_OPENID, SCOPE_OFFLINE_ACCESS, SCOPE_BOUND_KEY],
+            dpop_jkt=self.dpop_builder.jkt,
+        )
+        c_s256 = pkce_s256_challenge("foobar")
+        header = b64encode(
+            f"{self.provider.client_id}:{self.provider.client_secret}".encode()
+        ).decode()
+        response = self.client.post(
+            reverse("authentik_providers_oauth2:token"),
+            data={
+                "grant_type": GRANT_TYPE_AUTHORIZATION_CODE,
+                "code": code.code,
+                "redirect_uri": "http://local.invalid",
+            },
+            HTTP_AUTHORIZATION=f"Basic {header}",
+            HTTP_DPOP=self.dpop_builder.make_header(self.token_url, c_s256),
+        )
+        self.assertEqual(response.status_code, 200)
+        body = loads(response.content.decode())
+        self.assertIn("id_token", body)
+        self.assertIn("refresh_token", body)
+
+        # Decode ID token
+        id_token = body["id_token"]
+        id_token_payload = jwt_decode(
+            id_token, "", options={"verify_signature": False}, algorithms=["ES256"]
+        )
+        id_token_header = jwt_decode_complete(
+            id_token, "", options={"verify_signature": False}, algorithms=["ES256"]
+        )["header"]
+
+        self.assertEqual(id_token_header.get("typ"), JWT_TYPE_DPOP_ID_TOKEN)
+        self.assertIn("cnf", id_token_payload)
+        self.assertEqual(id_token_payload["cnf"]["jwk"]["kty"], "EC")
+
+        # Refresh token should be bound
+        refresh = RefreshToken.objects.filter(user=self.user, provider=self.provider).first()
+        self.assertIsNotNone(refresh)
+        self.assertEqual(refresh.dpop_jkt, self.dpop_builder.jkt)
+
+    def test_dpop_present_without_dpop_jkt_no_cnf(self):
+        """DPoP header but no dpop_jkt in auth request so no cnf in ID Token"""
+        code = AuthorizationCode.objects.create(
+            code="foobar",
+            provider=self.provider,
+            user=self.user,
+            auth_time=timezone.now(),
+            scope=[SCOPE_OPENID],
+            dpop_jkt=None,
+        )
+        header = b64encode(
+            f"{self.provider.client_id}:{self.provider.client_secret}".encode()
+        ).decode()
+        response = self.client.post(
+            reverse("authentik_providers_oauth2:token"),
+            data={
+                "grant_type": GRANT_TYPE_AUTHORIZATION_CODE,
+                "code": code.code,
+                "redirect_uri": "http://local.invalid",
+            },
+            HTTP_AUTHORIZATION=f"Basic {header}",
+            HTTP_DPOP=self.dpop_builder.make_header(self.token_url),
+        )
+        self.assertEqual(response.status_code, 200)
+        body = loads(response.content.decode())
+        id_token_payload = jwt_decode(
+            body["id_token"], "", options={"verify_signature": False}, algorithms=["ES256"]
+        )
+        self.assertNotIn("cnf", id_token_payload)
+        id_token_header = jwt_decode_complete(
+            body["id_token"], "", options={"verify_signature": False}, algorithms=["ES256"]
+        )["header"]
+        self.assertNotEqual(id_token_header.get("typ"), JWT_TYPE_DPOP_ID_TOKEN)
+
+    def test_bound_key_missing_dpop_fails(self):
+        """bound_key scope but no DPoP header so invalid_request"""
+        code = AuthorizationCode.objects.create(
+            code="foobar",
+            provider=self.provider,
+            user=self.user,
+            auth_time=timezone.now(),
+            scope=[SCOPE_OPENID, SCOPE_BOUND_KEY],
+            dpop_jkt=self.dpop_builder.jkt,
+        )
+        header = b64encode(
+            f"{self.provider.client_id}:{self.provider.client_secret}".encode()
+        ).decode()
+        response = self.client.post(
+            reverse("authentik_providers_oauth2:token"),
+            data={
+                "grant_type": GRANT_TYPE_AUTHORIZATION_CODE,
+                "code": code.code,
+                "redirect_uri": "http://local.invalid",
+            },
+            HTTP_AUTHORIZATION=f"Basic {header}",
+        )
+        self.assertEqual(response.status_code, 400)
+        body = loads(response.content.decode())
+        self.assertEqual(body["error"], "invalid_request")
+
+    def test_wrong_c_s256_fails(self):
+        """DPoP proof with wrong c_s256 so invalid_request"""
+        code = AuthorizationCode.objects.create(
+            code="foobar",
+            provider=self.provider,
+            user=self.user,
+            auth_time=timezone.now(),
+            scope=[SCOPE_OPENID, SCOPE_BOUND_KEY],
+            dpop_jkt=self.dpop_builder.jkt,
+        )
+        header = b64encode(
+            f"{self.provider.client_id}:{self.provider.client_secret}".encode()
+        ).decode()
+        response = self.client.post(
+            reverse("authentik_providers_oauth2:token"),
+            data={
+                "grant_type": GRANT_TYPE_AUTHORIZATION_CODE,
+                "code": code.code,
+                "redirect_uri": "http://local.invalid",
+            },
+            HTTP_AUTHORIZATION=f"Basic {header}",
+            HTTP_DPOP=self.dpop_builder.make_header(self.token_url, c_s256="wrong-hash"),
+        )
+        self.assertEqual(response.status_code, 400)
+        body = loads(response.content.decode())
+        self.assertEqual(body["error"], "invalid_request")
+
+    def test_refresh_with_same_key_succeeds(self):
+        """Refresh with matching DPoP key so new ID Token with same cnf"""
+        # Create initial key-bound refresh token
+        refresh = RefreshToken.objects.create(
+            provider=self.provider,
+            user=self.user,
+            token=generate_id(),
+            auth_time=timezone.now(),
+            scope=[SCOPE_OPENID, SCOPE_OFFLINE_ACCESS, SCOPE_BOUND_KEY],
+            dpop_jkt=self.dpop_builder.jkt,
+            _id_token=dumps({}),
+        )
+        header = b64encode(
+            f"{self.provider.client_id}:{self.provider.client_secret}".encode()
+        ).decode()
+        response = self.client.post(
+            reverse("authentik_providers_oauth2:token"),
+            data={
+                "grant_type": GRANT_TYPE_REFRESH_TOKEN,
+                "refresh_token": refresh.token,
+                "redirect_uri": "http://local.invalid",
+                "scope": f"{SCOPE_OPENID} {SCOPE_OFFLINE_ACCESS}",
+            },
+            HTTP_AUTHORIZATION=f"Basic {header}",
+            HTTP_DPOP=self.dpop_builder.make_header(self.token_url),
+        )
+        self.assertEqual(response.status_code, 200)
+        body = loads(response.content.decode())
+        id_token_payload = jwt_decode(
+            body["id_token"], "", options={"verify_signature": False}, algorithms=["ES256"]
+        )
+        self.assertIn("cnf", id_token_payload)
+        self.assertEqual(id_token_payload["cnf"]["jwk"]["kty"], "EC")
+
+    def test_refresh_with_wrong_key_fails(self):
+        """Refresh with different DPoP key"""
+        other_builder = DPoPProofBuilder()
+        refresh = RefreshToken.objects.create(
+            provider=self.provider,
+            user=self.user,
+            token=generate_id(),
+            auth_time=timezone.now(),
+            scope=[SCOPE_OPENID, SCOPE_OFFLINE_ACCESS, SCOPE_BOUND_KEY],
+            dpop_jkt=self.dpop_builder.jkt,
+            _id_token=dumps({}),
+        )
+        header = b64encode(
+            f"{self.provider.client_id}:{self.provider.client_secret}".encode()
+        ).decode()
+        response = self.client.post(
+            reverse("authentik_providers_oauth2:token"),
+            data={
+                "grant_type": GRANT_TYPE_REFRESH_TOKEN,
+                "refresh_token": refresh.token,
+                "redirect_uri": "http://local.invalid",
+                "scope": f"{SCOPE_OPENID} {SCOPE_OFFLINE_ACCESS}",
+            },
+            HTTP_AUTHORIZATION=f"Basic {header}",
+            HTTP_DPOP=other_builder.make_header(self.token_url),
+        )
+        self.assertEqual(response.status_code, 400)
+        body = loads(response.content.decode())
+        self.assertEqual(body["error"], "invalid_request")
+
+    def test_refresh_without_dpop_when_bound_fails(self):
+        """Key-bound refresh token used without DPoP header"""
+        refresh = RefreshToken.objects.create(
+            provider=self.provider,
+            user=self.user,
+            token=generate_id(),
+            auth_time=timezone.now(),
+            scope=[SCOPE_OPENID, SCOPE_OFFLINE_ACCESS, SCOPE_BOUND_KEY],
+            dpop_jkt=self.dpop_builder.jkt,
+            _id_token=dumps({}),
+        )
+        header = b64encode(
+            f"{self.provider.client_id}:{self.provider.client_secret}".encode()
+        ).decode()
+        response = self.client.post(
+            reverse("authentik_providers_oauth2:token"),
+            data={
+                "grant_type": GRANT_TYPE_REFRESH_TOKEN,
+                "refresh_token": refresh.token,
+                "redirect_uri": "http://local.invalid",
+                "scope": f"{SCOPE_OPENID} {SCOPE_OFFLINE_ACCESS}",
+            },
+            HTTP_AUTHORIZATION=f"Basic {header}",
+        )
+        self.assertEqual(response.status_code, 400)
+        body = loads(response.content.decode())
+        self.assertEqual(body["error"], "invalid_request")
+
+    def test_non_bound_refresh_ignores_dpop(self):
+        """Non-bound refresh token ignores DPoP header"""
+        refresh = RefreshToken.objects.create(
+            provider=self.provider,
+            user=self.user,
+            token=generate_id(),
+            auth_time=timezone.now(),
+            scope=[SCOPE_OPENID, SCOPE_OFFLINE_ACCESS],
+            dpop_jkt=None,
+            _id_token=dumps({}),
+        )
+        header = b64encode(
+            f"{self.provider.client_id}:{self.provider.client_secret}".encode()
+        ).decode()
+        response = self.client.post(
+            reverse("authentik_providers_oauth2:token"),
+            data={
+                "grant_type": GRANT_TYPE_REFRESH_TOKEN,
+                "refresh_token": refresh.token,
+                "redirect_uri": "http://local.invalid",
+                "scope": f"{SCOPE_OPENID} {SCOPE_OFFLINE_ACCESS}",
+            },
+            HTTP_AUTHORIZATION=f"Basic {header}",
+            HTTP_DPOP=self.dpop_builder.make_header(self.token_url),
+        )
+        self.assertEqual(response.status_code, 200)
+        body = loads(response.content.decode())
+        id_token_payload = jwt_decode(
+            body["id_token"], "", options={"verify_signature": False}, algorithms=["ES256"]
+        )
+        self.assertNotIn("cnf", id_token_payload)

authentik/providers/oauth2/tests/test_key_binding_device.py

@@ -0,0 +1,171 @@
+"""Test OpenID Connect Key Binding for Device Authorization flow"""
+
+from base64 import b64encode
+from json import dumps, loads
+
+from django.urls import reverse
+from django.utils import timezone
+from jwt import decode as jwt_decode
+from jwt import decode_complete as jwt_decode_complete
+
+from authentik.blueprints.tests import apply_blueprint
+from authentik.common.oauth.constants import (
+    GRANT_TYPE_DEVICE_CODE,
+    GRANT_TYPE_REFRESH_TOKEN,
+    JWT_TYPE_DPOP_ID_TOKEN,
+    SCOPE_BOUND_KEY,
+    SCOPE_OFFLINE_ACCESS,
+    SCOPE_OPENID,
+)
+from authentik.core.models import Application
+from authentik.core.tests.utils import create_test_admin_user, create_test_flow
+from authentik.lib.generators import generate_id
+from authentik.providers.oauth2.models import (
+    DeviceToken,
+    GrantType,
+    OAuth2Provider,
+    RedirectURI,
+    RedirectURIMatchingMode,
+    RefreshToken,
+    ScopeMapping,
+)
+from authentik.providers.oauth2.tests.test_dpop import DPoPProofBuilder
+from authentik.providers.oauth2.tests.utils import OAuthTestCase
+from authentik.providers.oauth2.utils import pkce_s256_challenge
+
+
+class TestKeyBindingDevice(OAuthTestCase):
+    """Test key-bound ID Tokens in device authorization flow"""
+
+    @apply_blueprint("system/providers-oauth2.yaml")
+    def setUp(self) -> None:
+        super().setUp()
+        self.provider = OAuth2Provider.objects.create(
+            name=generate_id(),
+            authorization_flow=create_test_flow(),
+            grant_types=[GrantType.DEVICE_CODE, GrantType.REFRESH_TOKEN],
+            redirect_uris=[RedirectURI(RedirectURIMatchingMode.STRICT, "http://local.invalid")],
+            signing_key=self.keypair,
+        )
+        self.provider.property_mappings.set(ScopeMapping.objects.all())
+        self.app = Application.objects.create(
+            name=generate_id(), slug="test", provider=self.provider
+        )
+        self.user = create_test_admin_user()
+        self.dpop_builder = DPoPProofBuilder()
+        self.token_url = "http://testserver/application/o/token/"
+
+    def test_device_successful_key_binding(self):
+        """Valid key bound + DPoP in device flow = ID Token has cnf and typ dpop+id_token"""
+        device_token = DeviceToken.objects.create(
+            provider=self.provider,
+            user=self.user,
+            scope=[SCOPE_OPENID, SCOPE_OFFLINE_ACCESS, SCOPE_BOUND_KEY],
+            dpop_jkt=self.dpop_builder.jkt,
+        )
+        c_s256 = pkce_s256_challenge(device_token.device_code)
+        response = self.client.post(
+            reverse("authentik_providers_oauth2:token"),
+            data={
+                "grant_type": GRANT_TYPE_DEVICE_CODE,
+                "device_code": device_token.device_code,
+                "client_id": self.provider.client_id,
+                "client_secret": self.provider.client_secret,
+            },
+            HTTP_DPOP=self.dpop_builder.make_header(self.token_url, c_s256),
+        )
+        self.assertEqual(response.status_code, 200)
+        body = loads(response.content.decode())
+        self.assertIn("id_token", body)
+        self.assertIn("refresh_token", body)
+
+        id_token_payload = jwt_decode(
+            body["id_token"], "", options={"verify_signature": False}, algorithms=["ES256"]
+        )
+        id_token_header = jwt_decode_complete(
+            body["id_token"], "", options={"verify_signature": False}, algorithms=["ES256"]
+        )["header"]
+
+        self.assertEqual(id_token_header.get("typ"), JWT_TYPE_DPOP_ID_TOKEN)
+        self.assertIn("cnf", id_token_payload)
+        self.assertEqual(id_token_payload["cnf"]["jwk"]["kty"], "EC")
+
+        # Refresh token should be bound
+        refresh = RefreshToken.objects.filter(user=self.user, provider=self.provider).first()
+        self.assertIsNotNone(refresh)
+        self.assertEqual(refresh.dpop_jkt, self.dpop_builder.jkt)
+
+    def test_device_wrong_c_s256_fails(self):
+        """DPoP proof with wrong c_s256 in device flow = invalid_request"""
+        device_token = DeviceToken.objects.create(
+            provider=self.provider,
+            user=self.user,
+            scope=[SCOPE_OPENID, SCOPE_BOUND_KEY],
+            dpop_jkt=self.dpop_builder.jkt,
+        )
+        response = self.client.post(
+            reverse("authentik_providers_oauth2:token"),
+            data={
+                "grant_type": GRANT_TYPE_DEVICE_CODE,
+                "device_code": device_token.device_code,
+                "client_id": self.provider.client_id,
+                "client_secret": self.provider.client_secret,
+            },
+            HTTP_DPOP=self.dpop_builder.make_header(self.token_url, c_s256="wrong-hash"),
+        )
+        self.assertEqual(response.status_code, 400)
+        body = loads(response.content.decode())
+        self.assertEqual(body["error"], "invalid_request")
+
+    def test_device_missing_dpop_fails(self):
+        """key bound scope but no DPoP header in device flow = invalid_request"""
+        device_token = DeviceToken.objects.create(
+            provider=self.provider,
+            user=self.user,
+            scope=[SCOPE_OPENID, SCOPE_BOUND_KEY],
+            dpop_jkt=self.dpop_builder.jkt,
+        )
+        response = self.client.post(
+            reverse("authentik_providers_oauth2:token"),
+            data={
+                "grant_type": GRANT_TYPE_DEVICE_CODE,
+                "device_code": device_token.device_code,
+                "client_id": self.provider.client_id,
+                "client_secret": self.provider.client_secret,
+            },
+        )
+        self.assertEqual(response.status_code, 400)
+        body = loads(response.content.decode())
+        self.assertEqual(body["error"], "invalid_request")
+
+    def test_device_refresh_with_same_key_succeeds(self):
+        """Refresh key bound token with matching DPoP key = success, same cnf"""
+        refresh = RefreshToken.objects.create(
+            provider=self.provider,
+            user=self.user,
+            token=generate_id(),
+            auth_time=timezone.now(),
+            scope=[SCOPE_OPENID, SCOPE_OFFLINE_ACCESS, SCOPE_BOUND_KEY],
+            dpop_jkt=self.dpop_builder.jkt,
+            _id_token=dumps({}),
+        )
+        header = b64encode(
+            f"{self.provider.client_id}:{self.provider.client_secret}".encode()
+        ).decode()
+        response = self.client.post(
+            reverse("authentik_providers_oauth2:token"),
+            data={
+                "grant_type": GRANT_TYPE_REFRESH_TOKEN,
+                "refresh_token": refresh.token,
+                "redirect_uri": "http://local.invalid",
+                "scope": f"{SCOPE_OPENID} {SCOPE_OFFLINE_ACCESS}",
+            },
+            HTTP_AUTHORIZATION=f"Basic {header}",
+            HTTP_DPOP=self.dpop_builder.make_header(self.token_url),
+        )
+        self.assertEqual(response.status_code, 200)
+        body = loads(response.content.decode())
+        id_token_payload = jwt_decode(
+            body["id_token"], "", options={"verify_signature": False}, algorithms=["ES256"]
+        )
+        self.assertIn("cnf", id_token_payload)

authentik/providers/oauth2/tests/test_key_binding_discovery.py

@@ -0,0 +1,64 @@
+"""Test OpenID Connect Key Binding discovery metadata"""
+
+from django.urls import reverse
+
+from authentik.blueprints.models import BlueprintInstance
+from authentik.blueprints.v1.importer import Importer
+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, ScopeMapping
+from authentik.providers.oauth2.tests.utils import OAuthTestCase
+
+
+class TestKeyBindingDiscovery(OAuthTestCase):
+    """Test discovery metadata includes key binding support"""
+
+    def setUp(self) -> None:
+        super().setUp()
+        content = BlueprintInstance(path="system/providers-oauth2.yaml").retrieve()
+        Importer.from_string(content).apply()
+        self.provider = OAuth2Provider.objects.create(
+            name=generate_id(),
+            client_id="test",
+            authorization_flow=create_test_flow(),
+            signing_key=self.keypair,
+        )
+        self.provider.property_mappings.set(
+            ScopeMapping.objects.filter(
+                managed__in=[
+                    "goauthentik.io/providers/oauth2/scope-openid",
+                    "goauthentik.io/providers/oauth2/scope-bound_key",
+                ]
+            )
+        )
+        self.app = Application.objects.create(
+            name=generate_id(), slug="test", provider=self.provider
+        )
+
+    def test_discovery_includes_bound_key(self):
+        """scopes_supported should include bound_key"""
+        response = self.client.get(
+            reverse(
+                "authentik_providers_oauth2:provider-info",
+                kwargs={"application_slug": self.app.slug},
+            )
+        )
+        self.assertEqual(response.status_code, 200)
+        body = response.json()
+        self.assertIn("scopes_supported", body)
+        self.assertIn("bound_key", body["scopes_supported"])
+
+    def test_discovery_includes_dpop_algs(self):
+        """dpop_signing_alg_values_supported should be present"""
+        response = self.client.get(
+            reverse(
+                "authentik_providers_oauth2:provider-info",
+                kwargs={"application_slug": self.app.slug},
+            )
+        )
+        self.assertEqual(response.status_code, 200)
+        body = response.json()
+        self.assertIn("dpop_signing_alg_values_supported", body)
+        self.assertIsInstance(body["dpop_signing_alg_values_supported"], list)
+        self.assertGreater(len(body["dpop_signing_alg_values_supported"]), 0)

authentik/providers/oauth2/views/authorize.py

@@ -26,6 +26,7 @@ from authentik.common.oauth.constants import (
     PROMPT_LOGIN,
     PROMPT_NONE,
     QS_LOGIN_HINT,
+    SCOPE_BOUND_KEY,
     SCOPE_GITHUB,
     SCOPE_OFFLINE_ACCESS,
     SCOPE_OPENID,
@@ -48,6 +49,7 @@ from authentik.lib.utils.time import timedelta_from_string
 from authentik.lib.views import bad_request_message
 from authentik.policies.types import PolicyRequest
 from authentik.policies.views import PolicyAccessView, RequestValidationError
+from authentik.providers.oauth2.dpop import is_valid_jkt
 from authentik.providers.oauth2.errors import (
     AuthorizeError,
     ClientIdError,
@@ -103,6 +105,7 @@ class OAuthAuthorizationParams:
 
     code_challenge: str | None = None
     code_challenge_method: str | None = None
+    dpop_jkt: str | None = None
 
     github_compat: InitVar[bool] = False
 
@@ -127,6 +130,7 @@ class OAuthAuthorizationParams:
         response_mode = query_dict.get("response_mode", False)
 
         max_age = query_dict.get("max_age")
+        dpop_jkt = query_dict.get("dpop_jkt")
         return OAuthAuthorizationParams(
             client_id=query_dict.get("client_id", ""),
             redirect_uri=redirect_uri,
@@ -141,6 +145,7 @@ class OAuthAuthorizationParams:
             max_age=int(max_age) if max_age else None,
             code_challenge=query_dict.get("code_challenge"),
             code_challenge_method=query_dict.get("code_challenge_method", "plain"),
+            dpop_jkt=dpop_jkt,
             github_compat=github_compat,
         )
 
@@ -168,6 +173,7 @@ class OAuthAuthorizationParams:
             )
         self.check_grant()
         self.check_scope(github_compat)
+        self.check_dpop_jkt()
         self.check_nonce()
         self.check_code_challenge()
 
@@ -321,6 +327,35 @@ class OAuthAuthorizationParams:
                 # Spec says to ignore the scope when the response_type wouldn't result
                 # in an authorization code being generated
                 self.scope.remove(SCOPE_OFFLINE_ACCESS)
+        # Key binding requires dpop_jkt at authorization time
+        if SCOPE_BOUND_KEY in self.scope and not self.dpop_jkt:
+            raise AuthorizeError(
+                self.redirect_uri,
+                error="invalid_request",
+                grant_type=self.grant_type,
+                state=self.state,
+                description="dpop_jkt is required when bound_key scope is requested",
+            )
+        # dpop_jkt should only be set if requesting the key binding scope
+        if SCOPE_BOUND_KEY not in self.scope and self.dpop_jkt:
+            raise AuthorizeError(
+                self.redirect_uri,
+                error="invalid_request",
+                grant_type=self.grant_type,
+                state=self.state,
+                description="dpop_jkt is set when bound_key scope is not requested",
+            )
+
+    def check_dpop_jkt(self):
+        """Validate dpop_jkt format if provided."""
+        if self.dpop_jkt and not is_valid_jkt(self.dpop_jkt):
+            raise AuthorizeError(
+                self.redirect_uri,
+                error="invalid_request",
+                grant_type=self.grant_type,
+                state=self.state,
+                description="dpop_jkt must be a base64url-encoded SHA-256 JWK thumbprint",
+            )
 
     def check_nonce(self):
         """Nonce parameter validation."""
@@ -383,6 +418,9 @@ class OAuthAuthorizationParams:
             code.code_challenge = self.code_challenge
             code.code_challenge_method = self.code_challenge_method
 
+        if self.dpop_jkt:
+            code.dpop_jkt = self.dpop_jkt
+
         return code
 
 

authentik/providers/oauth2/views/device_backchannel.py

@@ -11,9 +11,11 @@ from django.views.decorators.csrf import csrf_exempt
 from rest_framework.throttling import AnonRateThrottle
 from structlog.stdlib import get_logger
 
+from authentik.common.oauth.constants import SCOPE_BOUND_KEY
 from authentik.core.models import Application
 from authentik.lib.config import CONFIG
 from authentik.lib.utils.time import timedelta_from_string
+from authentik.providers.oauth2.dpop import is_valid_jkt
 from authentik.providers.oauth2.errors import DeviceCodeError
 from authentik.providers.oauth2.models import DeviceToken, GrantType, OAuth2Provider, ScopeMapping
 from authentik.providers.oauth2.utils import TokenResponse, extract_client_auth
@@ -29,6 +31,7 @@ class DeviceView(View):
     client_id: str
     provider: OAuth2Provider
     scopes: set[str] = []
+    dpop_jkt: str | None = None
 
     def parse_request(self):
         """Parse incoming request"""
@@ -62,6 +65,17 @@ class DeviceView(View):
             )
             self.scopes = self.scopes.intersection(default_scope_names)
 
+        self.dpop_jkt = self.request.POST.get("dpop_jkt")
+        if self.dpop_jkt and not is_valid_jkt(self.dpop_jkt):
+            raise DeviceCodeError("invalid_dpop_jkt")
+
+        # Key binding requires dpop_jkt at authorization time
+        if SCOPE_BOUND_KEY in self.scopes and not self.dpop_jkt:
+            raise DeviceCodeError("dpop_jkt_required")
+        # dpop_jkt should only be set if requesting the key binding scope
+        if SCOPE_BOUND_KEY not in self.scopes and self.dpop_jkt:
+            raise DeviceCodeError("dpop_jkt_not_allowed")
+
     def dispatch(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
         throttle = AnonRateThrottle()
         throttle.rate = CONFIG.get("throttle.providers.oauth2.device", "20/hour")
@@ -78,7 +92,10 @@ class DeviceView(View):
             return TokenResponse(exc.create_dict(request), status=400)
         until = timedelta_from_string(self.provider.access_code_validity)
         token: DeviceToken = DeviceToken.objects.create(
-            expires=now() + until, provider=self.provider, _scope=" ".join(self.scopes)
+            expires=now() + until,
+            provider=self.provider,
+            _scope=" ".join(self.scopes),
+            dpop_jkt=self.dpop_jkt,
         )
         device_url = self.request.build_absolute_uri(
             reverse("authentik_providers_oauth2_root:device-login")

authentik/providers/oauth2/views/introspection.py

@@ -80,6 +80,7 @@ class TokenIntrospectionView(View):
             response = {}
             if self.params.id_token:
                 response.update(self.params.id_token.to_dict())
+                response.pop("cnf", None)
             response["active"] = not self.params.token.is_expired and not self.params.token.revoked
             response["scope"] = " ".join(self.params.token.scope)
             response["client_id"] = self.params.provider.client_id

authentik/providers/oauth2/views/provider.py

@@ -22,6 +22,7 @@ from authentik.common.oauth.constants import (
 )
 from authentik.core.expression.exceptions import PropertyMappingExpressionException
 from authentik.core.models import Application
+from authentik.providers.oauth2.dpop import DPOP_SUPPORTED_ALGS
 from authentik.providers.oauth2.models import (
     OAuth2Provider,
     ResponseMode,
@@ -117,6 +118,7 @@ class ProviderInfoView(View):
             "claims_supported": self.get_claims(provider),
             "claims_parameter_supported": False,
             "code_challenge_methods_supported": [PKCE_METHOD_PLAIN, PKCE_METHOD_S256],
+            "dpop_signing_alg_values_supported": sorted(DPOP_SUPPORTED_ALGS),
         }
         if provider.encryption_key:
             config["id_token_encryption_alg_values_supported"] = ["RSA-OAEP-256"]

authentik/providers/oauth2/views/token.py

@@ -11,6 +11,7 @@ from typing import Any
 from urllib.parse import urlparse
 
 from django.http import HttpRequest, HttpResponse
+from django.urls import reverse
 from django.utils import timezone
 from django.utils.decorators import method_decorator
 from django.views import View
@@ -30,7 +31,9 @@ from authentik.common.oauth.constants import (
     GRANT_TYPE_PASSWORD,
     GRANT_TYPE_REFRESH_TOKEN,
     GRANT_TYPE_TOKEN_EXCHANGE,
+    JWT_TYPE_DPOP_ID_TOKEN,
     PKCE_METHOD_S256,
+    SCOPE_BOUND_KEY,
     SCOPE_OFFLINE_ACCESS,
     TOKEN_EXCHANGE_TOKEN_TYPES,
     TOKEN_TYPE,
@@ -56,6 +59,7 @@ from authentik.events.signals import get_login_event
 from authentik.flows.planner import PLAN_CONTEXT_APPLICATION
 from authentik.lib.utils.time import timedelta_from_string
 from authentik.policies.engine import PolicyEngine
+from authentik.providers.oauth2.dpop import DPoPError, DPoPValidator
 from authentik.providers.oauth2.errors import (
     DeviceCodeError,
     TokenError,
@@ -106,6 +110,8 @@ class TokenParams:
     user: User | None = None
 
     code_verifier: str | None = None
+    dpop_proof: str | None = None
+    dpop_jwk: dict | None = None
 
     requested_token_type: str | None = None
 
@@ -136,6 +142,8 @@ class TokenParams:
             scope=set(request.POST.get("scope", "").split()),
             # PKCE parameter.
             code_verifier=request.POST.get("code_verifier"),
+            # DPoP proof-of-possession header (RFC 9449)
+            dpop_proof=request.headers.get("DPoP"),
             # Token exchange parameter.
             requested_token_type=request.POST.get("requested_token_type"),
         )
@@ -177,6 +185,43 @@ class TokenParams:
                 )
                 raise TokenError("invalid_grant")
 
+    def _validate_dpop(
+        self,
+        request: HttpRequest,
+        dpop_jkt: str | None,
+        raw_code: str | None = None,
+        flow_name: str = "token",
+    ) -> None:
+        """Validate DPoP proof for key-bound tokens.
+
+        :param request: The current HTTP request
+        :param dpop_jkt: The expected JWK thumbprint (from auth request or previous token)
+        :param raw_code: The raw authorization code or device code (for c_s256 computation)
+        :param flow_name: Description of flow for logging (e.g., "authorization code")
+        :raises TokenError: If DPoP validation fails
+        """
+        if not self.dpop_proof:
+            LOGGER.warning("Missing DPoP proof for key-bound token", flow_name=flow_name)
+            raise TokenError("invalid_request")
+        if dpop_jkt is None:
+            LOGGER.warning("bound_key scope requested but no dpop_jkt", flow_name=flow_name)
+            raise TokenError("invalid_request")
+        try:
+            kwargs = {}
+            if raw_code is not None:
+                kwargs["expected_c_s256"] = pkce_s256_challenge(raw_code)
+            token_url = request.build_absolute_uri(reverse("authentik_providers_oauth2:token"))
+            self.dpop_jwk = DPoPValidator().validate(
+                self.dpop_proof,
+                expected_htm="POST",
+                expected_htu=token_url,
+                expected_jkt=dpop_jkt,
+                **kwargs,
+            )
+        except DPoPError as exc:
+            LOGGER.warning("DPoP validation failed", flow_name=flow_name, exc=str(exc))
+            raise TokenError("invalid_request") from exc
+
     def __post_init__(self, raw_code: str, raw_token: str, request: HttpRequest):
         if self.grant_type not in self.provider.grant_types:
             LOGGER.warning("Invalid grant_type for provider", grant_type=self.grant_type)
@@ -272,6 +317,14 @@ class TokenParams:
         if not self.authorization_code.code_challenge and self.code_verifier:
             raise TokenError("invalid_grant")
 
+        if SCOPE_BOUND_KEY in self.authorization_code.scope:
+            self._validate_dpop(
+                request,
+                self.authorization_code.dpop_jkt,
+                raw_code=raw_code,
+                flow_name="authorization code",
+            )
+
     def __check_redirect_uri(self, request: HttpRequest):
         allowed_redirect_urls = self.provider.authorization_redirect_uris
         # At this point, no provider should have a blank redirect_uri, in case they do
@@ -348,6 +401,13 @@ class TokenParams:
             ).from_http(request, user=self.refresh_token.user)
             raise TokenError("invalid_grant")
 
+        if self.refresh_token.dpop_jkt:
+            self._validate_dpop(
+                request,
+                self.refresh_token.dpop_jkt,
+                flow_name="refresh token",
+            )
+
     def __post_init_client_credentials(self, request: HttpRequest):
         # client_credentials flow with client assertion
         if request.POST.get(CLIENT_ASSERTION_TYPE, "") != "":
@@ -565,6 +625,14 @@ class TokenParams:
             raise TokenError("invalid_grant")
         self.device_code = code
 
+        if SCOPE_BOUND_KEY in self.device_code.scope:
+            self._validate_dpop(
+                request,
+                self.device_code.dpop_jkt,
+                raw_code=device_code,
+                flow_name="device code",
+            )
+
     def __post_init_token_exchange(self, request: HttpRequest):
         """See https://datatracker.ietf.org/doc/html/rfc8693#section-2.1"""
         # Delegation is not implemented. An actor token is rejected rather than ignored, so a
@@ -725,6 +793,17 @@ class TokenView(View):
         except UserAuthError as error:
             return TokenResponse(error.create_dict(request), status=403)
 
+    def _get_id_token_jwt_type(self) -> str | None:
+        """Return dpop+id_token if key binding is active, else None"""
+        if self.params.dpop_jwk is not None:
+            return JWT_TYPE_DPOP_ID_TOKEN
+        return None
+
+    def _add_cnf_to_id_token(self, id_token: IDToken) -> None:
+        """Add cnf claim to ID Token when key binding is active"""
+        if self.params.dpop_jwk is not None:
+            id_token.cnf = {"jwk": self.params.dpop_jwk}
+
     def create_code_response(self) -> dict[str, Any]:
         """See https://datatracker.ietf.org/doc/html/rfc6749#section-4.1"""
         now = timezone.now()
@@ -744,9 +823,11 @@ class TokenView(View):
             self.request,
         )
         access_id_token.nonce = self.params.authorization_code.nonce
+        self._add_cnf_to_id_token(access_id_token)
         access_token.id_token = access_id_token
         access_token.save()
 
+        id_token_jwt_type = self._get_id_token_jwt_type()
         response = {
             "access_token": access_token.token,
             "token_type": TOKEN_TYPE,
@@ -754,7 +835,7 @@ class TokenView(View):
             "expires_in": int(
                 timedelta_from_string(self.provider.access_token_validity).total_seconds()
             ),
-            "id_token": access_token.id_token.to_jwt(self.provider),
+            "id_token": access_token.id_token.to_jwt(self.provider, jwt_type=id_token_jwt_type),
         }
 
         if SCOPE_OFFLINE_ACCESS in self.params.authorization_code.scope:
@@ -766,6 +847,7 @@ class TokenView(View):
                 provider=self.provider,
                 auth_time=self.params.authorization_code.auth_time,
                 session=self.params.authorization_code.session,
+                dpop_jkt=self.params.authorization_code.dpop_jkt,
             )
             id_token = IDToken.new(
                 self.provider,
@@ -774,6 +856,7 @@ class TokenView(View):
             )
             id_token.nonce = self.params.authorization_code.nonce
             id_token.at_hash = access_token.at_hash
+            self._add_cnf_to_id_token(id_token)
             refresh_token.id_token = id_token
             refresh_token.save()
             response["refresh_token"] = refresh_token.token
@@ -800,21 +883,24 @@ class TokenView(View):
             auth_time=self.params.refresh_token.auth_time,
             session=self.params.refresh_token.session,
         )
-        access_token.id_token = IDToken.new(
+        access_id_token = IDToken.new(
             self.provider,
             access_token,
             self.request,
         )
+        self._add_cnf_to_id_token(access_id_token)
+        access_token.id_token = access_id_token
         access_token.save()
 
-        res = {
+        id_token_jwt_type = self._get_id_token_jwt_type()
+        response = {
             "access_token": access_token.token,
             "token_type": TOKEN_TYPE,
             "scope": " ".join(access_token.scope),
             "expires_in": int(
                 timedelta_from_string(self.provider.access_token_validity).total_seconds()
             ),
-            "id_token": access_token.id_token.to_jwt(self.provider),
+            "id_token": access_token.id_token.to_jwt(self.provider, jwt_type=id_token_jwt_type),
         }
 
         refresh_token_threshold = timedelta_from_string(self.provider.refresh_token_threshold)
@@ -830,6 +916,7 @@ class TokenView(View):
                 provider=self.provider,
                 auth_time=self.params.refresh_token.auth_time,
                 session=self.params.refresh_token.session,
+                dpop_jkt=self.params.refresh_token.dpop_jkt,
             )
             id_token = IDToken.new(
                 self.provider,
@@ -838,15 +925,16 @@ class TokenView(View):
             )
             id_token.nonce = self.params.refresh_token.id_token.nonce
             id_token.at_hash = access_token.at_hash
+            self._add_cnf_to_id_token(id_token)
             refresh_token.id_token = id_token
             refresh_token.save()
 
             # Mark old token as revoked
             self.params.refresh_token.revoked = True
             self.params.refresh_token.save()
-            res["refresh_token"] = refresh_token.token
+            response["refresh_token"] = refresh_token.token
 
-        return res
+        return response
 
     def create_client_credentials_response(self) -> dict[str, Any]:
         """See https://datatracker.ietf.org/doc/html/rfc6749#section-4.4"""
@@ -890,13 +978,16 @@ class TokenView(View):
             auth_time=auth_event.created if auth_event else now,
             session=self.params.device_code.session,
         )
-        access_token.id_token = IDToken.new(
+        access_id_token = IDToken.new(
             self.provider,
             access_token,
             self.request,
         )
+        self._add_cnf_to_id_token(access_id_token)
+        access_token.id_token = access_id_token
         access_token.save()
 
+        id_token_jwt_type = self._get_id_token_jwt_type()
         response = {
             "access_token": access_token.token,
             "token_type": TOKEN_TYPE,
@@ -904,7 +995,7 @@ class TokenView(View):
             "expires_in": int(
                 timedelta_from_string(self.provider.access_token_validity).total_seconds()
             ),
-            "id_token": access_token.id_token.to_jwt(self.provider),
+            "id_token": access_token.id_token.to_jwt(self.provider, jwt_type=id_token_jwt_type),
         }
 
         if SCOPE_OFFLINE_ACCESS in self.params.device_code.scope:
@@ -915,6 +1006,7 @@ class TokenView(View):
                 expires=refresh_token_expiry,
                 provider=self.provider,
                 auth_time=auth_event.created if auth_event else now,
+                dpop_jkt=self.params.device_code.dpop_jkt,
             )
             id_token = IDToken.new(
                 self.provider,
@@ -922,6 +1014,7 @@ class TokenView(View):
                 self.request,
             )
             id_token.at_hash = access_token.at_hash
+            self._add_cnf_to_id_token(id_token)
             refresh_token.id_token = id_token
             refresh_token.save()
             response["refresh_token"] = refresh_token.token

blueprints/system/providers-oauth2.yaml

@@ -79,3 +79,12 @@ entries:
         # This scope grants the application the ability to access the authentik API
         # on behalf of the authorizing user
         return {}
+  - identifiers:
+      managed: goauthentik.io/providers/oauth2/scope-bound_key
+    model: authentik_providers_oauth2.scopemapping
+    attrs:
+      name: "authentik default OAuth Mapping: OpenID 'bound_key'"
+      scope_name: bound_key
+      description: "Request a key-bound ID Token (OpenID Connect Key Binding)"
+      expression: |
+        return {}