From 99fcd00eba29f0889e7ad331cdbd5dc4ddd71beb Mon Sep 17 00:00:00 2001 From: Andrei Berenda Date: Wed, 6 May 2026 08:50:47 +0400 Subject: [PATCH 1/4] feat(openid): support private_key_jwt client authentication - Allow client_id=None on KeycloakOpenID (sanctions the runtime behavior required when the auth server derives client identity from the assertion's `sub` claim). - Add `client_assertion` constructor parameter (str or zero-arg callable). When set, _add_secret_key emits client_assertion_type/client_assertion and skips client_secret. Closes #717. - Add per-call `client_assertion` parameter on token(). - Tests for instance/callable/per-call assertions, suppression of client_secret, and the unchanged secret path. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/keycloak/keycloak_openid.py | 76 +++++++++++++++++++--- tests/test_keycloak_openid.py | 109 ++++++++++++++++++++++++++++++++ 2 files changed, 175 insertions(+), 10 deletions(-) diff --git a/src/keycloak/keycloak_openid.py b/src/keycloak/keycloak_openid.py index 5eb2f8e0..fc1ca3b0 100644 --- a/src/keycloak/keycloak_openid.py +++ b/src/keycloak/keycloak_openid.py @@ -31,7 +31,7 @@ class to handle authentication and token manipulation. import json import pathlib -from typing import Any +from typing import Any, Callable import aiofiles from jwcrypto import jwk, jwt @@ -76,9 +76,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=``, 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 +101,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 +110,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 +141,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 +165,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 +192,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 +250,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 +391,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 +420,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 +441,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) diff --git a/tests/test_keycloak_openid.py b/tests/test_keycloak_openid.py index 680b1bc7..1649b7ae 100644 --- a/tests/test_keycloak_openid.py +++ b/tests/test_keycloak_openid.py @@ -54,6 +54,115 @@ def test_keycloak_openid_init(env: KeycloakTestEnv) -> None: assert oid_default.connection.pool_maxsize is None +def test_keycloak_openid_init_client_assertion(env: KeycloakTestEnv) -> None: + """Test that ``client_id=None`` and ``client_assertion=...`` are accepted.""" + oid = KeycloakOpenID( + server_url=f"http://{env.keycloak_host}:{env.keycloak_port}", + realm_name="master", + client_id=None, + client_assertion="signed.jwt.value", + ) + assert oid.client_id is None + assert oid.client_assertion == "signed.jwt.value" + assert oid.client_secret_key is None + + +def test_add_secret_key_with_instance_assertion(env: KeycloakTestEnv) -> None: + """An instance-level assertion adds the assertion fields and skips client_secret.""" + oid = KeycloakOpenID( + server_url=f"http://{env.keycloak_host}:{env.keycloak_port}", + realm_name="master", + client_id=None, + client_secret_key="should-be-ignored", + client_assertion="signed.jwt.value", + ) + payload = oid._add_secret_key({"grant_type": "client_credentials"}) + assert payload == { + "grant_type": "client_credentials", + "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer", + "client_assertion": "signed.jwt.value", + } + assert "client_secret" not in payload + + +def test_add_secret_key_with_callable_assertion(env: KeycloakTestEnv) -> None: + """A callable assertion is invoked lazily and its return value is sent.""" + calls = [] + + def make_assertion() -> str: + calls.append(1) + return "fresh.jwt.value" + + oid = KeycloakOpenID( + server_url=f"http://{env.keycloak_host}:{env.keycloak_port}", + realm_name="master", + client_id=None, + client_assertion=make_assertion, + ) + payload = oid._add_secret_key({}) + assert payload["client_assertion"] == "fresh.jwt.value" + assert payload["client_assertion_type"] == ( + "urn:ietf:params:oauth:client-assertion-type:jwt-bearer" + ) + assert calls == [1] + + +def test_add_secret_key_assertion_in_payload_suppresses_secret(env: KeycloakTestEnv) -> None: + """A per-call ``client_assertion`` in the payload suppresses ``client_secret``.""" + oid = KeycloakOpenID( + server_url=f"http://{env.keycloak_host}:{env.keycloak_port}", + realm_name="master", + client_id="some-client", + client_secret_key="should-be-ignored", + ) + payload = oid._add_secret_key({"client_assertion": "per.call.jwt"}) + assert payload == { + "client_assertion": "per.call.jwt", + "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer", + } + assert "client_secret" not in payload + + +def test_add_secret_key_secret_path_unchanged(env: KeycloakTestEnv) -> None: + """Without an assertion, ``client_secret_key`` is sent as before.""" + oid = KeycloakOpenID( + server_url=f"http://{env.keycloak_host}:{env.keycloak_port}", + realm_name="master", + client_id="some-client", + client_secret_key="the-secret", + ) + payload = oid._add_secret_key({"grant_type": "client_credentials"}) + assert payload == { + "grant_type": "client_credentials", + "client_secret": "the-secret", + } + + +def test_token_sends_private_key_jwt_payload(env: KeycloakTestEnv) -> None: + """``token()`` with ``client_assertion=`` produces a private_key_jwt wire payload.""" + oid = KeycloakOpenID( + server_url=f"http://{env.keycloak_host}:{env.keycloak_port}", + realm_name="master", + client_id=None, + ) + + fake_response = mock.Mock(status_code=200) + fake_response.json.return_value = {"access_token": "x", "token_type": "Bearer"} + with mock.patch.object( + oid.connection, "raw_post", return_value=fake_response + ) as raw_post: + oid.token(grant_type="client_credentials", client_assertion="signed.jwt.value") + + sent_payload = raw_post.call_args.kwargs["data"] + assert sent_payload["grant_type"] == "client_credentials" + assert sent_payload["client_assertion"] == "signed.jwt.value" + assert sent_payload["client_assertion_type"] == ( + "urn:ietf:params:oauth:client-assertion-type:jwt-bearer" + ) + assert "client_secret" not in sent_payload + assert sent_payload["client_id"] is None + + def test_well_known(oid: KeycloakOpenID) -> None: """ Test the well_known method. From d755cfc966f47d5c78e2f1c45f6f8cad71f59bb8 Mon Sep 17 00:00:00 2001 From: Andrei Berenda Date: Wed, 6 May 2026 09:06:14 +0400 Subject: [PATCH 2/4] feat(admin): support private_key_jwt in KeycloakOpenIDConnection Plumbs `client_assertion` through KeycloakOpenIDConnection into the underlying KeycloakOpenID, so KeycloakAdmin works natively with private_key_jwt clients (including auto-refresh, which re-invokes the configured assertion callable). - `client_id: str` -> `client_id: str | None` on KeycloakOpenIDConnection. - New `client_assertion: str | Callable[[], str] | None` parameter, forwarded to the inner KeycloakOpenID constructor. - `grant_type` defaults to `client_credentials` when an assertion is set. - `keycloak_openid` accessor now requires either `client_id` or `client_assertion` (not just `client_id`). - Tests for assertion plumbing, callable laziness, and the missing-creds error path. Verified end-to-end against a live Keycloak realm with a Kubernetes SA JWT: get_realm/get_users/get_clients all succeed; callable assertion is invoked once per token mint and re-invoked on forced re-issue. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/keycloak/openid_connection.py | 40 +++++++++++++++++++----- tests/test_keycloak_admin.py | 51 +++++++++++++++++++++++++++++++ 2 files changed, 84 insertions(+), 7 deletions(-) diff --git a/src/keycloak/openid_connection.py b/src/keycloak/openid_connection.py index a2a42991..64080c04 100644 --- a/src/keycloak/openid_connection.py +++ b/src/keycloak/openid_connection.py @@ -31,7 +31,7 @@ from __future__ import annotations from datetime import datetime, timedelta, timezone -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, Callable if TYPE_CHECKING: from httpx import Response as AsyncResponse @@ -74,7 +74,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 +83,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 +102,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 +125,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 +145,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 +155,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 +242,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 +376,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 +396,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 diff --git a/tests/test_keycloak_admin.py b/tests/test_keycloak_admin.py index be5fd4f8..d0be0248 100644 --- a/tests/test_keycloak_admin.py +++ b/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. From fa79a29d5bd990dcdae859efa270030c4f32304d Mon Sep 17 00:00:00 2001 From: Andrei Berenda Date: Wed, 6 May 2026 09:08:50 +0400 Subject: [PATCH 3/4] chore: appease ruff (UP035, S106, TC003) and format Resolves CI lint failures introduced by the previous two commits: - Move `Callable` import to `collections.abc` and into a TYPE_CHECKING block (UP035 + TC003); both modules already use `from __future__ import annotations`, so it is type-only. - Suppress S106 on three test fixtures that intentionally pass dummy values to `client_secret_key` to verify wire-payload behavior. - Apply `ruff format` to the new test block. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/keycloak/keycloak_openid.py | 5 ++++- src/keycloak/openid_connection.py | 4 +++- tests/test_keycloak_openid.py | 10 ++++------ 3 files changed, 11 insertions(+), 8 deletions(-) diff --git a/src/keycloak/keycloak_openid.py b/src/keycloak/keycloak_openid.py index fc1ca3b0..043c1994 100644 --- a/src/keycloak/keycloak_openid.py +++ b/src/keycloak/keycloak_openid.py @@ -31,7 +31,10 @@ class to handle authentication and token manipulation. import json import pathlib -from typing import Any, Callable +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from collections.abc import Callable import aiofiles from jwcrypto import jwk, jwt diff --git a/src/keycloak/openid_connection.py b/src/keycloak/openid_connection.py index 64080c04..f0bef750 100644 --- a/src/keycloak/openid_connection.py +++ b/src/keycloak/openid_connection.py @@ -31,9 +31,11 @@ from __future__ import annotations from datetime import datetime, timedelta, timezone -from typing import TYPE_CHECKING, Any, Callable +from typing import TYPE_CHECKING, Any if TYPE_CHECKING: + from collections.abc import Callable + from httpx import Response as AsyncResponse from requests import Response diff --git a/tests/test_keycloak_openid.py b/tests/test_keycloak_openid.py index 1649b7ae..8a86ac7f 100644 --- a/tests/test_keycloak_openid.py +++ b/tests/test_keycloak_openid.py @@ -73,7 +73,7 @@ def test_add_secret_key_with_instance_assertion(env: KeycloakTestEnv) -> None: server_url=f"http://{env.keycloak_host}:{env.keycloak_port}", realm_name="master", client_id=None, - client_secret_key="should-be-ignored", + client_secret_key="should-be-ignored", # noqa: S106 client_assertion="signed.jwt.value", ) payload = oid._add_secret_key({"grant_type": "client_credentials"}) @@ -113,7 +113,7 @@ def test_add_secret_key_assertion_in_payload_suppresses_secret(env: KeycloakTest server_url=f"http://{env.keycloak_host}:{env.keycloak_port}", realm_name="master", client_id="some-client", - client_secret_key="should-be-ignored", + client_secret_key="should-be-ignored", # noqa: S106 ) payload = oid._add_secret_key({"client_assertion": "per.call.jwt"}) assert payload == { @@ -129,7 +129,7 @@ def test_add_secret_key_secret_path_unchanged(env: KeycloakTestEnv) -> None: server_url=f"http://{env.keycloak_host}:{env.keycloak_port}", realm_name="master", client_id="some-client", - client_secret_key="the-secret", + client_secret_key="the-secret", # noqa: S106 ) payload = oid._add_secret_key({"grant_type": "client_credentials"}) assert payload == { @@ -148,9 +148,7 @@ def test_token_sends_private_key_jwt_payload(env: KeycloakTestEnv) -> None: fake_response = mock.Mock(status_code=200) fake_response.json.return_value = {"access_token": "x", "token_type": "Bearer"} - with mock.patch.object( - oid.connection, "raw_post", return_value=fake_response - ) as raw_post: + with mock.patch.object(oid.connection, "raw_post", return_value=fake_response) as raw_post: oid.token(grant_type="client_credentials", client_assertion="signed.jwt.value") sent_payload = raw_post.call_args.kwargs["data"] From 15fc4e80276a8800643fea41a86e632e925001bb Mon Sep 17 00:00:00 2001 From: Andrei Berenda Date: Wed, 6 May 2026 09:16:57 +0400 Subject: [PATCH 4/4] fix(openid): add client_assertion to a_token() to match sync signature `test_counter_part` enforces parameter parity between sync `token()` and async `a_token()`. The previous commit added `client_assertion` only to the sync method, which broke that invariant on Python 3.10/22.0 CI. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/keycloak/keycloak_openid.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/src/keycloak/keycloak_openid.py b/src/keycloak/keycloak_openid.py index 043c1994..e67f29cc 100644 --- a/src/keycloak/keycloak_openid.py +++ b/src/keycloak/keycloak_openid.py @@ -1289,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: """ @@ -1317,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 @@ -1334,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)