diff --git a/src/keycloak/keycloak_openid.py b/src/keycloak/keycloak_openid.py index 5eb2f8e0..e67f29cc 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 +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=``, 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) diff --git a/src/keycloak/openid_connection.py b/src/keycloak/openid_connection.py index a2a42991..f0bef750 100644 --- a/src/keycloak/openid_connection.py +++ b/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 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. diff --git a/tests/test_keycloak_openid.py b/tests/test_keycloak_openid.py index 680b1bc7..8a86ac7f 100644 --- a/tests/test_keycloak_openid.py +++ b/tests/test_keycloak_openid.py @@ -54,6 +54,113 @@ 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", # noqa: S106 + 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", # noqa: S106 + ) + 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", # noqa: S106 + ) + 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.