feat: support ssl cert from string

src/keycloak/connection.py

@@ -33,6 +33,8 @@
 
 import httpx
 import requests
+import ssl
+
 from httpx import Response as AsyncResponse
 from requests import Response
 from requests.adapters import HTTPAdapter
@@ -41,6 +43,21 @@
 from .exceptions import KeycloakConnectionError
 
 
+class _HTTPAdapterWithSSLContext(HTTPAdapter):
+    """Creates an adapter that uses a custom ssl context."""
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:  # noqa: ANN401
+        """Init method."""
+        self._ca_cert = kwargs.pop("ca_cert", None)
+        super().__init__(*args, **kwargs)
+
+    def init_poolmanager(self, *args: Any, **kwargs: Any) -> None:  # noqa: ANN401
+        """Init pool manager."""
+        super().init_poolmanager(
+            *args, **kwargs, ssl_context=ssl.create_default_context(cadata=self._ca_cert)
+        )
+
+
 class ConnectionManager:
     """
     Represents a simple server connection.
@@ -64,6 +81,8 @@ class ConnectionManager:
     :type max_retries: int
     :param pool_maxsize: The maximum number of connections to save in the pool.
     :type pool_maxsize: int
+    :param : An optional CA certificate bundle in PEM format
+    :type : str
     """
 
     def __init__(
@@ -76,6 +95,7 @@ def __init__(
         cert: str | tuple | None = None,
         max_retries: int = 1,
         pool_maxsize: int | None = None,
+        ca_cert: str | None = None,
     ) -> None:
         """
         Init method.
@@ -99,6 +119,8 @@ def __init__(
         :type max_retries: int
         :param pool_maxsize: The maximum number of connections to save in the pool.
         :type pool_maxsize: int
+        :param ca_cert: An optional CA certificate bundle in PEM format
+        :type ca_cert: str
         """
         self.base_url = base_url
         self.headers = headers
@@ -114,10 +136,10 @@ def __init__(
         # retry once to reset connection with Keycloak after  tomcat's ConnectionTimeout
         # see https://github.com/marcospereirampj/python-keycloak/issues/36
         for protocol in ("https://", "http://"):
-            adapter_kwargs = {"max_retries": max_retries}
+            adapter_kwargs = {"ca_cert": ca_cert, "max_retries": max_retries}
             if pool_maxsize is not None:
                 adapter_kwargs["pool_maxsize"] = pool_maxsize
-            adapter = HTTPAdapter(**adapter_kwargs)  # pyright: ignore[reportArgumentType]
+            adapter = _HTTPAdapterWithSSLContext(**adapter_kwargs)  # pyright: ignore[reportArgumentType]
             # adds POST to retry whitelist
             allowed_methods = (
                 set(adapter.max_retries.allowed_methods)

src/keycloak/keycloak_openid.py

@@ -90,6 +90,8 @@ class KeycloakOpenID:
     :param max_retries: The total number of times to retry HTTP requests.
     :param pool_maxsize: The maximum number of connections to save in the pool.
     :type pool_maxsize: int
+    :param ca_cert: An optional CA certificate bundle in PEM format
+    :type ca_cert: str
     """
 
     def __init__(
@@ -105,6 +107,7 @@ def __init__(
         cert: str | tuple | None = None,
         max_retries: int = 1,
         pool_maxsize: int | None = None,
+        ca_cert: str | None = None,
     ) -> None:
         """
         Init method.
@@ -134,6 +137,8 @@ def __init__(
         :type max_retries: int
         :param pool_maxsize: The maximum number of connections to save in the pool.
         :type pool_maxsize: int
+        :param ca_cert: An optional CA certificate bundle in PEM format
+        :type ca_cert: str
         """
         self.client_id = client_id
         self.client_secret_key = client_secret_key
@@ -148,6 +153,7 @@ def __init__(
             cert=cert,
             max_retries=max_retries,
             pool_maxsize=pool_maxsize,
+            ca_cert=ca_cert,
         )
 
         self.authorization = Authorization()