feat: support private_key_jwt client authentication

src/keycloak/keycloak_openid.py

@@ -31,7 +31,10 @@ class to handle authentication and token manipulation.
 
 import json
 import pathlib
-from typing import Any
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
 
 import aiofiles
 from jwcrypto import jwk, jwt
@@ -76,9 +79,14 @@ class KeycloakOpenID:
     Keycloak OpenID client.
 
     :param server_url: Keycloak server url
-    :param client_id: client id
+    :param client_id: client id. May be ``None`` when authenticating with a
+        ``private_key_jwt`` client assertion whose ``sub`` claim identifies the client.
     :param realm_name: realm name
     :param client_secret_key: client secret key
+    :param client_assertion: signed JWT (or zero-arg callable returning one) used for
+        ``private_key_jwt`` client authentication. When set, the token endpoint receives
+        ``client_assertion_type=urn:ietf:params:oauth:client-assertion-type:jwt-bearer``
+        and ``client_assertion=<jwt>``, and ``client_secret`` is not sent.
     :param verify: Boolean value to enable or disable certificate validation or a string
         containing a path to a CA bundle to use
     :param custom_headers: dict of custom header to pass to each HTML request
@@ -96,7 +104,7 @@ def __init__(
         self,
         server_url: str,
         realm_name: str,
-        client_id: str,
+        client_id: str | None,
         client_secret_key: str | None = None,
         verify: bool | str = True,
         custom_headers: dict | None = None,
@@ -105,14 +113,16 @@ def __init__(
         cert: str | tuple | None = None,
         max_retries: int = 1,
         pool_maxsize: int | None = None,
+        client_assertion: str | Callable[[], str] | None = None,
     ) -> None:
         """
         Init method.
 
         :param server_url: Keycloak server url
         :type server_url: str
-        :param client_id: client id
-        :type client_id: str
+        :param client_id: client id. May be ``None`` for ``private_key_jwt`` clients
+            whose identity is derived from the assertion's ``sub`` claim.
+        :type client_id: str | None
         :param realm_name: realm name
         :type realm_name: str
         :param client_secret_key: client secret key
@@ -134,9 +144,14 @@ def __init__(
         :type max_retries: int
         :param pool_maxsize: The maximum number of connections to save in the pool.
         :type pool_maxsize: int
+        :param client_assertion: signed JWT, or zero-arg callable returning one, for
+            ``private_key_jwt`` client authentication. When set, ``client_secret`` is
+            never sent on the wire.
+        :type client_assertion: str | Callable[[], str] | None
         """
         self.client_id = client_id
         self.client_secret_key = client_secret_key
+        self.client_assertion = client_assertion
         self.realm_name = realm_name
         headers = custom_headers if custom_headers is not None else {}
         self.connection = ConnectionManager(
@@ -153,17 +168,17 @@ def __init__(
         self.authorization = Authorization()
 
     @property
-    def client_id(self) -> str:
+    def client_id(self) -> str | None:
         """
         Get client id.
 
         :returns: Client id
-        :rtype: str
+        :rtype: str | None
         """
         return self._client_id
 
     @client_id.setter
-    def client_id(self, value: str) -> None:
+    def client_id(self, value: str | None) -> None:
         self._client_id = value
 
     @property
@@ -180,6 +195,20 @@ def client_secret_key(self) -> str | None:
     def client_secret_key(self, value: str | None) -> None:
         self._client_secret_key = value
 
+    @property
+    def client_assertion(self) -> str | Callable[[], str] | None:
+        """
+        Get the client assertion (or callable producing one) for ``private_key_jwt``.
+
+        :returns: Client assertion
+        :rtype: str | Callable[[], str] | None
+        """
+        return self._client_assertion
+
+    @client_assertion.setter
+    def client_assertion(self, value: str | Callable[[], str] | None) -> None:
+        self._client_assertion = value
+
     @property
     def realm_name(self) -> str:
         """
@@ -224,13 +253,36 @@ def authorization(self, value: Authorization) -> None:
 
     def _add_secret_key(self, payload: dict) -> dict:
         """
-        Add secret key if exists.
+        Add client authentication credentials to the payload.
+
+        Emits a ``private_key_jwt`` client assertion when one is configured (or already
+        present in the payload), otherwise falls back to ``client_secret``. The two are
+        mutually exclusive on the wire.
 
         :param payload: Payload
         :type payload: dict
-        :returns: Payload with the secret key
+        :returns: Payload with the client authentication credentials
         :rtype: dict
         """
+        if "client_assertion" in payload:
+            payload.setdefault(
+                "client_assertion_type",
+                "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
+            )
+            return payload
+
+        if self.client_assertion is not None:
+            assertion = (
+                self.client_assertion()
+                if callable(self.client_assertion)
+                else self.client_assertion
+            )
+            payload["client_assertion_type"] = (
+                "urn:ietf:params:oauth:client-assertion-type:jwt-bearer"
+            )
+            payload["client_assertion"] = assertion
+            return payload
+
         if self.client_secret_key:
             payload.update({"client_secret": self.client_secret_key})
 
@@ -342,6 +394,7 @@ def token(
         totp: int | None = None,
         scope: str = "openid",
         code_verifier: str | None = None,
+        client_assertion: str | None = None,
         **extra: Any,  # noqa: ANN401
     ) -> dict:
         """
@@ -370,6 +423,10 @@ def token(
         :type scope: str
         :param code_verifier: PKCE code verifier
         :type code_verifier: str
+        :param client_assertion: per-call ``private_key_jwt`` assertion. Overrides the
+            instance-level ``client_assertion`` for this request and suppresses
+            ``client_secret``.
+        :type client_assertion: str | None
         :param extra: Additional extra arguments
         :type extra: dict
         :returns: Keycloak token
@@ -387,6 +444,8 @@ def token(
         }
         if code_verifier:
             payload["code_verifier"] = code_verifier
+        if client_assertion is not None:
+            payload["client_assertion"] = client_assertion
         if extra:
             payload.update(extra)
 
@@ -1230,6 +1289,7 @@ async def a_token(
         totp: int | None = None,
         scope: str = "openid",
         code_verifier: str | None = None,
+        client_assertion: str | None = None,
         **extra: Any,  # noqa: ANN401
     ) -> dict:
         """
@@ -1258,6 +1318,10 @@ async def a_token(
         :type scope: str
         :param code_verifier: PKCE code verifier
         :type code_verifier: str
+        :param client_assertion: per-call ``private_key_jwt`` assertion. Overrides the
+            instance-level ``client_assertion`` for this request and suppresses
+            ``client_secret``.
+        :type client_assertion: str | None
         :param extra: Additional extra arguments
         :type extra: dict
         :returns: Keycloak token
@@ -1275,6 +1339,8 @@ async def a_token(
         }
         if code_verifier:
             payload["code_verifier"] = code_verifier
+        if client_assertion is not None:
+            payload["client_assertion"] = client_assertion
         if extra:
             payload.update(extra)
 

src/keycloak/openid_connection.py

@@ -34,6 +34,8 @@
 from typing import TYPE_CHECKING, Any
 
 if TYPE_CHECKING:
+    from collections.abc import Callable
+
     from httpx import Response as AsyncResponse
     from requests import Response
 
@@ -74,7 +76,7 @@ def __init__(
         token: dict | None = None,
         totp: int | None = None,
         realm_name: str | None = "master",
-        client_id: str = "admin-cli",
+        client_id: str | None = "admin-cli",
         verify: str | bool = True,
         client_secret_key: str | None = None,
         custom_headers: dict | None = None,
@@ -83,6 +85,7 @@ def __init__(
         cert: str | tuple | None = None,
         max_retries: int = 1,
         pool_maxsize: int | None = None,
+        client_assertion: str | Callable[[], str] | None = None,
     ) -> None:
         """
         Init method.
@@ -101,8 +104,9 @@ def __init__(
         :type totp: str
         :param realm_name: realm name
         :type realm_name: str
-        :param client_id: client id
-        :type client_id: str
+        :param client_id: client id. May be ``None`` for ``private_key_jwt`` clients
+            whose identity is derived from the assertion's ``sub`` claim.
+        :type client_id: str | None
         :param verify: Boolean value to enable or disable certificate validation or a string
             containing a path to a CA bundle to use
         :type verify: Union[bool,str]
@@ -123,6 +127,11 @@ def __init__(
         :type max_retries: int
         :param pool_maxsize: The maximum number of connections to save in the pool.
         :type pool_maxsize: int
+        :param client_assertion: signed JWT, or zero-arg callable returning one, for
+            ``private_key_jwt`` client authentication. When set, the underlying
+            ``KeycloakOpenID`` is configured to authenticate with the assertion
+            (``client_secret_key`` is not sent).
+        :type client_assertion: str | Callable[[], str] | None
         """
         # token is renewed when it hits 90% of its lifetime. This is to account for any possible
         # clock skew.
@@ -138,6 +147,7 @@ def __init__(
         self.client_id = client_id
         self.verify = verify
         self.client_secret_key = client_secret_key
+        self.client_assertion = client_assertion
         self.user_realm_name = user_realm_name
         self.timeout = timeout
         self.custom_headers = custom_headers
@@ -147,7 +157,7 @@ def __init__(
         if not self.grant_type:
             if username and password:
                 self.grant_type = "password"
-            elif client_secret_key:
+            elif client_secret_key or client_assertion:
                 self.grant_type = "client_credentials"
 
         if self.server_url is None:
@@ -234,6 +244,20 @@ def client_secret_key(self) -> str | None:
     def client_secret_key(self, value: str | None) -> None:
         self._client_secret_key = value
 
+    @property
+    def client_assertion(self) -> str | Callable[[], str] | None:
+        """
+        Get the client assertion (or callable producing one) for ``private_key_jwt``.
+
+        :returns: Client assertion
+        :rtype: str | Callable[[], str] | None
+        """
+        return self._client_assertion
+
+    @client_assertion.setter
+    def client_assertion(self, value: str | Callable[[], str] | None) -> None:
+        self._client_assertion = value
+
     @property
     def username(self) -> str | None:
         """
@@ -354,8 +378,11 @@ def keycloak_openid(self) -> KeycloakOpenID:
             else:
                 token_realm_name = "master"  # noqa: S105
 
-            if self.client_id is None:
-                msg = "Unable to get KeycloakOpenID client without client_id set."
+            if self.client_id is None and self.client_assertion is None:
+                msg = (
+                    "Unable to get KeycloakOpenID client without client_id or "
+                    "client_assertion set."
+                )
                 raise AttributeError(msg)
 
             if self.server_url is None:
@@ -371,6 +398,7 @@ def keycloak_openid(self) -> KeycloakOpenID:
                 timeout=self.timeout,
                 custom_headers=self.custom_headers,
                 cert=self.cert,
+                client_assertion=self.client_assertion,
             )
 
         return self._keycloak_openid

tests/test_keycloak_admin.py

@@ -156,6 +156,57 @@ def test_keycloak_admin_init(env: KeycloakTestEnv) -> None:
     assert keycloak_admin.connection.token
 
 
+def test_openid_connection_client_assertion_init(env: KeycloakTestEnv) -> None:
+    """Test that ``client_assertion`` is plumbed through to the inner KeycloakOpenID."""
+    conn = KeycloakOpenIDConnection(
+        server_url=f"http://{env.keycloak_host}:{env.keycloak_port}",
+        realm_name="master",
+        client_id=None,
+        client_assertion="signed.jwt.value",
+    )
+    assert conn.client_id is None
+    assert conn.client_assertion == "signed.jwt.value"
+    # client_assertion alone is enough to default to client_credentials.
+    assert conn.grant_type == "client_credentials"
+
+    inner = conn.keycloak_openid
+    assert inner.client_id is None
+    assert inner.client_assertion == "signed.jwt.value"
+
+
+def test_openid_connection_callable_client_assertion_lazy(env: KeycloakTestEnv) -> None:
+    """A callable ``client_assertion`` is forwarded as-is (lazy)."""
+    calls = []
+
+    def make_assertion() -> str:
+        calls.append(1)
+        return "fresh.jwt.value"
+
+    conn = KeycloakOpenIDConnection(
+        server_url=f"http://{env.keycloak_host}:{env.keycloak_port}",
+        realm_name="master",
+        client_id=None,
+        client_assertion=make_assertion,
+    )
+    inner = conn.keycloak_openid
+    # The callable should not have been invoked just by constructing the connection
+    # or building the inner KeycloakOpenID.
+    assert calls == []
+    # And it's the same callable reference, not a captured value.
+    assert inner.client_assertion is make_assertion
+
+
+def test_openid_connection_requires_client_id_or_assertion(env: KeycloakTestEnv) -> None:
+    """Without ``client_id`` and without ``client_assertion``, building the inner client errors."""
+    conn = KeycloakOpenIDConnection(
+        server_url=f"http://{env.keycloak_host}:{env.keycloak_port}",
+        realm_name="master",
+        client_id=None,
+    )
+    with pytest.raises(AttributeError, match="client_id or client_assertion"):
+        _ = conn.keycloak_openid
+
+
 def test_realms(admin: KeycloakAdmin) -> None:
     """
     Test realms.