feat: decode token with all keys, added get client certificate info method (#704)

* fix: do not throw when processing signed userinfo response

* chore: remove custom pyproject commit

* fix: add all keys for decoding the token

* feat: new methods for certificate key info

* chore: lint fix

* fix: use async

* test: updated regex

Author: ryshoooo

poetry.lock

@@ -118,7 +118,7 @@ def type(self) -> str:
         """
         return self._type
 
-    @type.setter
+    @type.setter  # noqa: A003
     def type(self, value: str) -> None:
         self._type = value
 

src/keycloak/authorization/permission.py

@@ -115,7 +115,7 @@ def type(self) -> str:
         """
         return self._type
 
-    @type.setter
+    @type.setter  # noqa: A003
     def type(self, value: str) -> None:
         self._type = value
 

src/keycloak/authorization/policy.py

@@ -7518,6 +7518,35 @@ def get_role_client_level_children(self, client_id: str, role_id: str) -> list:
 
         return res
 
+    def get_client_certificate_key_info(self, client_id: str, attribute: str) -> dict:
+        """
+        Get client certificate key info.
+
+        :param client_id: id of the client
+        :type client_id: str
+        :param attribute: attribute to query for
+        :type attribute: str
+        :return: Certificate key info
+        :rtype: dict
+        """
+        params_path = {
+            "realm-name": self.connection.realm_name,
+            "id": client_id,
+            "attr": attribute,
+        }
+        data_raw = self.connection.raw_get(
+            urls_patterns.URL_ADMIN_CLIENT_CERTS.format(**params_path)
+        )
+        res = raise_error_from_response(data_raw, KeycloakGetError)
+        if not isinstance(res, dict):
+            msg = (
+                "Unexpected response type. Expected 'dict', received "
+                f"'{type(res)}', value '{res}'."
+            )
+            raise TypeError(msg)
+
+        return res
+
     def upload_certificate(self, client_id: str, certcont: str) -> dict:
         """
         Upload a new certificate for the client.
@@ -14313,6 +14342,35 @@ async def a_get_role_client_level_children(self, client_id: str, role_id: str) -
 
         return res
 
+    async def a_get_client_certificate_key_info(self, client_id: str, attribute: str) -> dict:
+        """
+        Get client certificate key info.
+
+        :param client_id: id of the client
+        :type client_id: str
+        :param attribute: attribute to query for
+        :type attribute: str
+        :return: Certificate key info
+        :rtype: dict
+        """
+        params_path = {
+            "realm-name": self.connection.realm_name,
+            "id": client_id,
+            "attr": attribute,
+        }
+        data_raw = await self.connection.a_raw_get(
+            urls_patterns.URL_ADMIN_CLIENT_CERTS.format(**params_path)
+        )
+        res = raise_error_from_response(data_raw, KeycloakGetError)
+        if not isinstance(res, dict):
+            msg = (
+                "Unexpected response type. Expected 'dict', received "
+                f"'{type(res)}', value '{res}'."
+            )
+            raise TypeError(msg)
+
+        return res
+
     async def a_upload_certificate(self, client_id: str, certcont: str) -> dict:
         """
         Upload a new certificate for the client asynchronously.

src/keycloak/keycloak_admin.py

@@ -791,12 +791,9 @@ def decode_token(self, token: str, validate: bool = True, **kwargs: Any) -> dict
         key = kwargs.pop("key", None)
         if validate:
             if key is None:
-                key = (
-                    "-----BEGIN PUBLIC KEY-----\n"
-                    + self.public_key()
-                    + "\n-----END PUBLIC KEY-----"
-                )
-                key = jwk.JWK.from_pem(key.encode("utf-8"))
+                key = jwk.JWKSet()
+                for cert in self.certs()["keys"]:
+                    key.add(jwk.JWK(**cert))
         else:
             key = None
 
@@ -1656,12 +1653,9 @@ async def a_decode_token(self, token: str, validate: bool = True, **kwargs: Any)
         key = kwargs.pop("key", None)
         if validate:
             if key is None:
-                key = (
-                    "-----BEGIN PUBLIC KEY-----\n"
-                    + await self.a_public_key()
-                    + "\n-----END PUBLIC KEY-----"
-                )
-                key = jwk.JWK.from_pem(key.encode("utf-8"))
+                key = jwk.JWKSet()
+                for cert in (await self.a_certs())["keys"]:
+                    key.add(jwk.JWK(**cert))
         else:
             key = None
 

src/keycloak/keycloak_openid.py

@@ -89,7 +89,7 @@ def test_counter_part() -> None:
         assert sync_sign.parameters == async_sign.parameters
 
     for async_method in async_methods:
-        if async_method in ["aclose"]:
+        if async_method == "aclose":
             continue
         if async_method[2:].startswith("_"):
             continue

tests/test_connection.py

@@ -40,6 +40,7 @@
 NO_CLIENT_SCOPE_REGEX = '404: b\'{"error":"Could not find client scope".*}\''
 UNKOWN_ERROR_REGEX = 'b\'{"error":"unknown_error".*}\''
 USER_NOT_FOUND_REGEX = '404: b\'{"error":"User not found".*}\''
+COULD_NOT_FIND_CLIENT_REGEX = '404: b\'{"error":"Could not find client".*}\''
 
 
 def test_keycloak_version() -> None:
@@ -7697,6 +7698,23 @@ async def test_a_consents(
     assert err.match(CONSENT_NOT_FOUND_REGEX)
 
 
+def test_keycloak_client_get_cert_info(admin: KeycloakAdmin, client: str) -> None:
+    """Test get cert info."""
+    assert admin.get_client_certificate_key_info(client, "jwt.credential") == {}
+    with pytest.raises(KeycloakGetError) as res:
+        admin.get_client_certificate_key_info("blah", "blah")
+    assert res.match(COULD_NOT_FIND_CLIENT_REGEX)
+
+
+@pytest.mark.asyncio
+async def test_a_keycloak_client_get_cert_info(admin: KeycloakAdmin, client: str) -> None:
+    """Test get cert info."""
+    assert await admin.a_get_client_certificate_key_info(client, "jwt.credential") == {}
+    with pytest.raises(KeycloakGetError) as res:
+        await admin.a_get_client_certificate_key_info("blah", "blah")
+    assert res.match(COULD_NOT_FIND_CLIENT_REGEX)
+
+
 def test_counter_part() -> None:
     """Test that each function has its async counter part."""
     admin_methods = [func for func in dir(KeycloakAdmin) if callable(getattr(KeycloakAdmin, func))]

tests/test_keycloak_admin.py

@@ -5,6 +5,7 @@
 
 import jwcrypto.jwk
 import jwcrypto.jws
+import jwcrypto.jwt
 import pytest
 
 from keycloak import KeycloakAdmin, KeycloakOpenID
@@ -379,7 +380,7 @@ def test_decode_token_invalid_token(oid_with_credentials: tuple[KeycloakOpenID,
     key = jwcrypto.jwk.JWK.from_pem(key.encode("utf-8"))
 
     invalid_access_token = access_token + "a"
-    with pytest.raises(jwcrypto.jws.InvalidJWSSignature):
+    with pytest.raises(jwcrypto.jwt.JWTMissingKey):
         decoded_invalid_access_token = oid.decode_token(token=invalid_access_token, validate=True)
 
     with pytest.raises(jwcrypto.jws.InvalidJWSSignature):
@@ -941,7 +942,7 @@ async def test_a_decode_token_invalid_token(
     key = jwcrypto.jwk.JWK.from_pem(key.encode("utf-8"))
 
     invalid_access_token = access_token + "a"
-    with pytest.raises(jwcrypto.jws.InvalidJWSSignature):
+    with pytest.raises(jwcrypto.jwt.JWTMissingKey):
         decoded_invalid_access_token = await oid.a_decode_token(
             token=invalid_access_token,
             validate=True,
@@ -1219,3 +1220,54 @@ def test_counter_part() -> None:
             continue
 
         assert async_method[2:] in sync_methods
+
+
+def test_other_signing_methods(
+    admin: KeycloakAdmin, oid_with_credentials: tuple[KeycloakOpenID, str, str]
+) -> None:
+    """Test other signing algs."""
+    oid, username, password = oid_with_credentials
+
+    admin.change_current_realm(oid.realm_name)
+    client_id = admin.get_client_id(oid.client_id)
+    assert client_id is not None
+    client_def = admin.get_client(client_id)
+    client_def["attributes"].update(
+        {
+            "access.token.signed.response.alg": "RS512",
+            "id.token.signed.response.alg": "RS512",
+            "userinfo.signed.response.alg": "RS512",
+        }
+    )
+    res = admin.update_client(client_id, client_def)
+    assert res == {}
+
+    token = oid.token(username, password)
+    res = oid.decode_token(token["access_token"])
+    assert res != {}
+
+
+@pytest.mark.asyncio
+async def test_a_other_signing_methods(
+    admin: KeycloakAdmin, oid_with_credentials: tuple[KeycloakOpenID, str, str]
+) -> None:
+    """Test other signing algs."""
+    oid, username, password = oid_with_credentials
+
+    await admin.a_change_current_realm(oid.realm_name)
+    client_id = await admin.a_get_client_id(oid.client_id)
+    assert client_id is not None
+    client_def = await admin.a_get_client(client_id)
+    client_def["attributes"].update(
+        {
+            "access.token.signed.response.alg": "RS512",
+            "id.token.signed.response.alg": "RS512",
+            "userinfo.signed.response.alg": "RS512",
+        }
+    )
+    res = await admin.a_update_client(client_id, client_def)
+    assert res == {}
+
+    token = await oid.a_token(username, password)
+    res = await oid.a_decode_token(token["access_token"])
+    assert res != {}