Remove ModelMesh and Serverless from get_ca_bundle (#1173)

Author: threcc

tests/conftest.py

@@ -690,9 +690,7 @@ def prometheus(admin_client: DynamicClient) -> Prometheus:
     return Prometheus(
         client=admin_client,
         resource_name="thanos-querier",
-        verify_ssl=create_ca_bundle_file(
-            client=admin_client, ca_type="openshift"
-        ),  # TODO: Verify SSL with appropriate certs
+        verify_ssl=create_ca_bundle_file(client=admin_client),  # TODO: Verify SSL with appropriate certs
         bearer_token=get_openshift_token(),
     )
 

tests/model_explainability/evalhub/conftest.py

@@ -66,4 +66,4 @@ def evalhub_ca_bundle_file(
     admin_client: DynamicClient,
 ) -> str:
     """Create a CA bundle file for verifying the EvalHub route TLS certificate."""
-    return create_ca_bundle_file(client=admin_client, ca_type="openshift")
+    return create_ca_bundle_file(client=admin_client)

tests/model_explainability/guardrails/conftest.py

@@ -107,7 +107,7 @@ def prompt_injection_detector_route(
 def openshift_ca_bundle_file(
     admin_client: DynamicClient,
 ) -> str:
-    return create_ca_bundle_file(client=admin_client, ca_type="openshift")
+    return create_ca_bundle_file(client=admin_client)
 
 
 @pytest.fixture(scope="class")

tests/model_explainability/trustyai_service/trustyai_service_utils.py

@@ -59,7 +59,7 @@ def __init__(self, token: str, service: TrustyAIService, client: DynamicClient):
         self.service_route = Route(
             client=client, namespace=service.namespace, name=TRUSTYAI_SERVICE_NAME, ensure_exists=True
         )
-        self.cert_path = create_ca_bundle_file(client=client, ca_type="openshift")
+        self.cert_path = create_ca_bundle_file(client=client)
 
     def _get_metric_base_url(self, metric_name: str) -> str:
         """Gets base URL for a given metric type (fairness or drift).

utilities/certificates_utils.py

@@ -9,98 +9,61 @@
 from simple_logger.logger import get_logger
 
 from utilities.constants import (
-    ISTIO_CA_BUNDLE_FILENAME,
     OPENSHIFT_CA_BUNDLE_FILENAME,
-    KServeDeploymentType,
 )
-from utilities.infra import is_managed_cluster, is_self_managed_operator
+from utilities.infra import is_managed_cluster
 
 LOGGER = get_logger(name=__name__)
 
 
-def create_ca_bundle_file(client: DynamicClient, ca_type: str) -> str:
+def create_ca_bundle_file(client: DynamicClient) -> str:
     """
     Creates a ca bundle file from a secret
 
     Args:
         client (DynamicClient): DynamicClient object
-        ca_type (str): The type of ca bundle to create. Can be "knative" or "openshift"
-
     Returns:
-        str: The path to the ca bundle file. If cert is not created, return empty string
+        str: The path to the ca bundle file.
 
     Raises:
-        ValueError: If ca_type is not "knative" or "openshift"
-
+        AttributeError: If the router-certs-default secret does not exist in the cluster.
     """
-    if ca_type == "knative":
-        certs_secret = Secret(
-            client=client,
-            name="knative-serving-cert",
-            namespace="istio-system",
-        )
-        filename = ISTIO_CA_BUNDLE_FILENAME
-
-    elif ca_type == "openshift":
-        certs_secret = Secret(
-            client=client,
-            name="router-certs-default",
-            namespace="openshift-ingress",
-        )
-        filename = OPENSHIFT_CA_BUNDLE_FILENAME
 
-    else:
-        raise ValueError("Invalid ca_type")
+    certs_secret = Secret(
+        client=client,
+        name="router-certs-default",
+        namespace="openshift-ingress",
+    )
 
-    if certs_secret.exists:
-        bundle = base64.b64decode(certs_secret.instance.data["tls.crt"]).decode()
-        filepath = os.path.join(py_config["tmp_base_dir"], filename)
-        with open(filepath, "w") as fd:
-            fd.write(bundle)
+    filename = OPENSHIFT_CA_BUNDLE_FILENAME
+    bundle = base64.b64decode(certs_secret.instance.data["tls.crt"]).decode()
+    filepath = os.path.join(py_config["tmp_base_dir"], filename)
 
-        return filepath
+    with open(filepath, "w") as fd:
+        fd.write(bundle)
 
-    LOGGER.warning(f"Could not find {certs_secret.name} secret")
-    return ""
+    return filepath
 
 
 @cache
-def get_ca_bundle(client: DynamicClient, deployment_mode: str) -> str:
+def get_ca_bundle(client: DynamicClient) -> str:
     """
-    Get the ca bundle for the given deployment mode.
+    Get the CA bundle for TLS verification.
 
-    If running on managed cluster and deployment in serverless or raw deployment, return empty string.
-    If running on self-managed operator and deployment is model mesh, return ca bundle.
+    On managed clusters, no CA bundle is needed (returns empty string).
+    On self-managed clusters, creates a CA bundle file.
 
     Args:
         client (DynamicClient): DynamicClient object
-        deployment_mode (str): The deployment mode. Can be "serverless", "model-mesh" or "raw-deployment"
 
     Returns:
-        str: The path to the ca bundle file. If cert is not created, return empty string
-
-    Raises:
-            ValueError: If deployment_mode is not "serverless", "model-mesh" or "raw-deployment"
-
+        str: The path to the ca bundle file, or empty string if not needed or not found.
     """
-    if deployment_mode in (
-        KServeDeploymentType.SERVERLESS,
-        KServeDeploymentType.RAW_DEPLOYMENT,
-    ):
-        if is_managed_cluster(client):
-            LOGGER.info("Running on managed cluster, not using ca bundle")
-            return ""
-        else:
-            return create_ca_bundle_file(client=client, ca_type="knative")
-
-    elif deployment_mode == KServeDeploymentType.MODEL_MESH:
-        if is_self_managed_operator(client=client):
-            return create_ca_bundle_file(client=client, ca_type="openshift")
-
+    if is_managed_cluster(client):
+        LOGGER.info("Running on managed cluster, not using ca bundle")
         return ""
 
-    else:
-        raise ValueError(f"Unknown deployment mode: {deployment_mode}")
+    return create_ca_bundle_file(client=client)
 
 
 def create_k8s_secret(

utilities/inference_utils.py

@@ -327,7 +327,7 @@ def generate_command(
         else:
             # admin client is needed to check if cluster is managed
             _client = get_client()
-            if ca := get_ca_bundle(client=_client, deployment_mode=self.deployment_mode):
+            if ca := get_ca_bundle(client=_client):
                 cmd += f" --cacert {ca} "
 
             else:

utilities/infra.py

@@ -23,7 +23,6 @@
     ResourceNotFoundError,
 )
 from ocp_resources.authentication_config_openshift_io import Authentication
-from ocp_resources.catalog_source import CatalogSource
 from ocp_resources.cluster_service_version import ClusterServiceVersion
 from ocp_resources.config_imageregistry_operator_openshift_io import Config
 from ocp_resources.config_map import ConfigMap
@@ -468,23 +467,6 @@ def login_with_user_password(api_address: str, user: str, password: str | None =
     return bool(re.search(r"Login successful|Logged into", out))
 
 
-@cache
-def is_self_managed_operator(client: DynamicClient) -> bool:
-    """
-    Check if the operator is self-managed.
-    """
-    if py_config["distribution"] == "upstream":
-        return True
-
-    return not bool(
-        CatalogSource(
-            client=client,
-            name="addon-managed-odh-catalog",
-            namespace=py_config["applications_namespace"],
-        ).exists
-    )
-
-
 @cache
 def is_managed_cluster(client: DynamicClient) -> bool:
     """

utilities/llmd_utils.py

@@ -583,7 +583,7 @@ def generate_command(
                 from ocp_resources.resource import get_client
 
                 client = get_client()
-                ca_bundle = get_ca_bundle(client=client, deployment_mode="raw")
+                ca_bundle = get_ca_bundle(client=client)
                 if ca_bundle:
                     cmd += f" --cacert {ca_bundle}"
                 else: