Make oidc ssl cert verification configurable

Signed-off-by: Andrea Lamparelli <[email protected]>

Author: lampajr

src/horreum/configs.py

@@ -1,6 +1,6 @@
 from dataclasses import dataclass
-from typing import Optional
 from enum import Enum
+from typing import Optional
 
 import httpx
 from kiota_abstractions.request_option import RequestOption
@@ -27,3 +27,5 @@ class ClientConfiguration:
     options: Optional[dict[str, RequestOption]] = None
     # which authentication method to use
     auth_method: AuthMethod = AuthMethod.BEARER
+    # SSL cert verification against the oidc provider
+    auth_verify: bool = True

src/horreum/horreum_client.py

@@ -1,9 +1,9 @@
+import base64
+import logging
 from importlib.metadata import version
 from typing import Optional
 
-import base64
 import httpx
-import logging
 from kiota_abstractions.authentication import AuthenticationProvider, ApiKeyAuthenticationProvider, KeyLocation
 from kiota_abstractions.authentication.access_token_provider import AccessTokenProvider
 from kiota_abstractions.authentication.anonymous_authentication_provider import AnonymousAuthenticationProvider
@@ -21,16 +21,17 @@
 
 logger = logging.getLogger(__name__)
 
-async def setup_auth_provider(base_url: str, username: str, password: str) -> AccessTokenProvider:
+
+async def setup_auth_provider(base_url: str, username: str, password: str, http_client: httpx.AsyncClient = None,
+                              verify: bool = True) -> AccessTokenProvider:
     # Use not authenticated client to fetch the auth mechanism
     auth_provider = AnonymousAuthenticationProvider()
-    req_adapter = HttpxRequestAdapter(auth_provider)
+    req_adapter = HttpxRequestAdapter(authentication_provider=auth_provider, http_client=http_client)
     req_adapter.base_url = base_url
     auth_client = HorreumRawClient(req_adapter)
 
     auth_config = await auth_client.api.config.keycloak.get()
-    # TODO: we could generalize using a generic OIDC client
-    return KeycloakAccessProvider(auth_config, username, password)
+    return KeycloakAccessProvider(auth_config, username, password, verify)
 
 
 class HorreumClient:
@@ -49,6 +50,7 @@ def __init__(self, base_url: str, credentials: Optional[HorreumCredentials],
         self.__base_url = base_url
         self.__credentials = credentials
         self.__client_config = client_config
+        self.__auth_verify = client_config.auth_verify if client_config is not None else True
 
         if client_config and client_config.http_client and client_config.use_default_middlewares:
             self.__http_client = KiotaClientFactory.create_with_default_middleware(client=client_config.http_client,
@@ -62,20 +64,25 @@ async def setup(self):
         """
 
         if self.__credentials:
-            if self.__credentials.apikey is not None and (self.__client_config is None or self.__client_config.auth_method == AuthMethod.API_KEY):
+            if self.__credentials.apikey is not None and (
+                    self.__client_config is None or self.__client_config.auth_method == AuthMethod.API_KEY):
                 # API key authentication
-                self.auth_provider = ApiKeyAuthenticationProvider(KeyLocation.Header, self.__credentials.apikey, "X-Horreum-API-Key")
+                self.auth_provider = ApiKeyAuthenticationProvider(KeyLocation.Header, self.__credentials.apikey,
+                                                                  "X-Horreum-API-Key")
                 logger.info('Using API Key authentication')
 
             elif self.__credentials.username is not None:
                 if self.__client_config is None or self.__client_config.auth_method == AuthMethod.BEARER:
                     # Bearer token authentication
-                    access_provider = await setup_auth_provider(self.__base_url, self.__credentials.username, self.__credentials.password)
+                    access_provider = await setup_auth_provider(self.__base_url, self.__credentials.username,
+                                                                self.__credentials.password, self.__http_client,
+                                                                self.__auth_verify)
                     self.auth_provider = BaseBearerTokenAuthenticationProvider(access_provider)
                     logger.info('Using OIDC bearer token authentication')
                 elif self.__client_config.auth_method == AuthMethod.BASIC:
                     # Basic authentication
-                    basic = "Basic " + base64.b64encode((self.__credentials.username + ":" + self.__credentials.password).encode()).decode()
+                    basic = "Basic " + base64.b64encode(
+                        (self.__credentials.username + ":" + self.__credentials.password).encode()).decode()
                     self.auth_provider = ApiKeyAuthenticationProvider(KeyLocation.Header, basic, "Authentication")
                     logger.info('Using Basic HTTP authentication')
             elif self.__credentials.password is not None:

src/horreum/keycloak_access_provider.py

@@ -12,15 +12,16 @@ class KeycloakAccessProvider(AccessTokenProvider):
     username: str
     password: str
 
-    def __init__(self, config: KeycloakConfig, username: str, password: str):
+    def __init__(self, config: KeycloakConfig, username: str, password: str, verify: bool = True):
         super()
         self.config = config
         self.username = username
         self.password = password
         self.keycloak_openid = KeycloakOpenID(
             server_url=config.url,
             client_id=config.client_id,
-            realm_name=config.realm
+            realm_name=config.realm,
+            verify=verify
         )
 
     async def get_authorization_token(self, uri: str, additional_authentication_context: Dict[str, Any] = {}) -> str: