Check Kubernetes cluster endpoint for defining the ssl_ca_cert

src/clusterfuzz/_internal/k8s/service.py

@@ -14,6 +14,7 @@
 """Kubernetes batch client."""
 import base64
 import collections
+import ipaddress
 import os
 import typing
 import uuid
@@ -206,7 +207,16 @@ def _load_gke_credentials(self):
 
     configuration = k8s_client.Configuration()
     configuration.host = f'https://{endpoint}'
-    configuration.ssl_ca_cert = ca_cert_path
+
+    try:
+      ipaddress.ip_address(endpoint)
+      configuration.ssl_ca_cert = ca_cert_path
+    except ValueError:
+      # If the endpoint is a hostname, we assume it's using a public CA or
+      # the system trust store should be used.
+      logs.info(f'Endpoint {endpoint} is a hostname. '
+                'Skipping custom CA configuration.')
+
     configuration.verify_ssl = True
 
     def get_token(creds):

src/clusterfuzz/_internal/tests/core/k8s/k8s_credentials_test.py

@@ -0,0 +1,96 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Tests for KubernetesService credential loading."""
+
+import os
+import unittest
+from unittest import mock
+
+from clusterfuzz._internal.k8s import service
+from clusterfuzz._internal.tests.test_libs import helpers
+
+
+class KubernetesCredentialsTest(unittest.TestCase):
+  """Tests for KubernetesService credential loading."""
+
+  def setUp(self):
+    helpers.patch(self, [
+        'clusterfuzz._internal.system.environment.get_value',
+        'google.auth.default',
+        'googleapiclient.discovery.build',
+        'kubernetes.client.Configuration',
+        'kubernetes.config.load_kube_config',
+    ])
+    self.mock.get_value.return_value = 'test-project'
+    creds = mock.Mock()
+    creds.token = 'test-token'
+    self.mock.default.return_value = (creds, 'test-project')
+
+    self.mock_discovery = self.mock.build.return_value
+    self.mock_clusters = self.mock_discovery.projects.return_value.locations.return_value.clusters.return_value
+
+    self.mock_config_instance = self.mock.Configuration.return_value
+
+    os.environ['BOT_DIR'] = '/tmp'
+
+  def test_load_gke_credentials_ip_endpoint(self):
+    """Test loading credentials with an IP endpoint (should set ssl_ca_cert)."""
+    self.mock_clusters.list.return_value.execute.return_value = {
+        'clusters': [{
+            'name': 'clusterfuzz-cronjobs-gke',
+            'endpoint': '1.2.3.4',
+            'masterAuth': {
+                'clusterCaCertificate': 'dGVzdA=='  # base64 "test"
+            }
+        }]
+    }
+
+    # Bypass __init__ logic to call _load_gke_credentials directly
+    with mock.patch.object(
+        service.KubernetesService, '__init__', return_value=None):
+      kube_service = service.KubernetesService()
+
+    # pylint: disable=protected-access
+    kube_service._load_gke_credentials()
+
+    self.assertEqual(self.mock_config_instance.host, 'https://1.2.3.4')
+    self.assertIsNotNone(self.mock_config_instance.ssl_ca_cert)
+    self.assertTrue(self.mock_config_instance.verify_ssl)
+
+  def test_load_gke_credentials_hostname_endpoint(self):
+    """Test loading credentials with a hostname endpoint (should skip ssl_ca_cert)."""
+    self.mock_clusters.list.return_value.execute.return_value = {
+        'clusters': [{
+            'name': 'clusterfuzz-cronjobs-gke',
+            'endpoint': 'example.com',
+            'masterAuth': {
+                'clusterCaCertificate': 'dGVzdA=='
+            }
+        }]
+    }
+
+    # Bypass __init__ logic to call _load_gke_credentials directly
+    with mock.patch.object(
+        service.KubernetesService, '__init__', return_value=None):
+      kube_service = service.KubernetesService()
+
+    # Reset mock to ensure we capture the specific call
+    self.mock_config_instance.ssl_ca_cert = None
+
+    # pylint: disable=protected-access
+    kube_service._load_gke_credentials()
+
+    self.assertEqual(self.mock_config_instance.host, 'https://example.com')
+    self.assertIsNone(self.mock_config_instance.ssl_ca_cert)
+    self.assertTrue(self.mock_config_instance.verify_ssl)