feat: Add verify_ssl support to OIDC auth flow for self-signed certif…

…icates Signed-off-by: Aniket Paluskar <apaluska@redhat.com> Made-with: Cursor
ntkathole · Apr 7, 2026 · Mar 10, 2026 · Mar 10, 2026 · Mar 10, 2026 · Mar 10, 2026
commit 13310576492133297095de49b51a7d57afa80d28
@@ -1,5 +1,6 @@
 import logging
 import os
+import ssl
 from typing import Optional
 from unittest.mock import Mock
 
@@ -40,7 +41,8 @@ def __init__(
         self._auth_config = auth_config
         self._k8s_parser = k8s_parser
         self.oidc_discovery_service = OIDCDiscoveryService(
-            self._auth_config.auth_discovery_url
+            self._auth_config.auth_discovery_url,
+            verify_ssl=self._auth_config.verify_ssl,
         )
 
     async def _validate_token(self, access_token: str):
@@ -91,8 +93,14 @@ def _extract_claim(data: dict, *keys: str, expected_type: type = list):
     def _decode_token(self, access_token: str) -> dict:
         """Fetch the JWKS signing key and decode + verify the JWT."""
         optional_custom_headers = {"User-agent": "custom-user-agent"}
+        ssl_ctx = ssl.create_default_context()
+        if not self._auth_config.verify_ssl:
+            ssl_ctx.check_hostname = False
+            ssl_ctx.verify_mode = ssl.CERT_NONE
         jwks_client = PyJWKClient(
-            self.oidc_discovery_service.get_jwks_url(), headers=optional_custom_headers
+            self.oidc_discovery_service.get_jwks_url(),
+            headers=optional_custom_headers,
+            ssl_context=ssl_ctx,
         )
         signing_key = jwks_client.get_signing_key_from_jwt(access_token)
         return jwt.decode(

@@ -36,6 +36,7 @@ class AuthConfig(FeastConfigBaseModel):
 class OidcAuthConfig(AuthConfig):
     auth_discovery_url: str
     client_id: str
+    verify_ssl: bool = True
 
 
 class OidcClientAuthConfig(OidcAuthConfig):

@@ -69,7 +69,8 @@ def _fetch_token_from_idp(self) -> str:
         """Obtain an access token via client_credentials or ROPG flow."""
         assert self.auth_config.auth_discovery_url is not None
         token_endpoint = OIDCDiscoveryService(
-            self.auth_config.auth_discovery_url
+            self.auth_config.auth_discovery_url,
+            verify_ssl=self.auth_config.verify_ssl,
         ).get_token_url()
 
         if self.auth_config.client_secret and not (
@@ -91,7 +92,10 @@ def _fetch_token_from_idp(self) -> str:
 
         headers = {"Content-Type": "application/x-www-form-urlencoded"}
         token_response = requests.post(
-            token_endpoint, data=token_request_body, headers=headers
+            token_endpoint,
+            data=token_request_body,
+            headers=headers,
+            verify=self.auth_config.verify_ssl,
         )
 
         if token_response.status_code == 200:

@@ -2,8 +2,9 @@
 
 
 class OIDCDiscoveryService:
-    def __init__(self, discovery_url: str):
+    def __init__(self, discovery_url: str, verify_ssl: bool = True):
         self.discovery_url = discovery_url
+        self._verify_ssl = verify_ssl
         self._discovery_data = None  # Initialize it lazily.
 
     @property
@@ -15,7 +16,7 @@ def discovery_data(self):
 
     def _fetch_discovery_data(self) -> dict:
         try:
-            response = requests.get(self.discovery_url)
+            response = requests.get(self.discovery_url, verify=self._verify_ssl)
             response.raise_for_status()
             return response.json()
         except requests.RequestException as e: