Support a custom SSL CA bundle for KubernetesHook connections

providers/cncf/kubernetes/docs/connections/kubernetes.rst

@@ -63,6 +63,12 @@ Cluster context
 Disable verify SSL
   Can optionally disable SSL certificate verification.  By default SSL is verified.
 
+Ssl ca cert
+  Path to a CA bundle used to verify the Kubernetes API server certificate, for clusters
+  whose certificate is signed by a custom or intermediate CA. Ignored when ``Disable verify SSL``
+  is set. The file must be present on the filesystem where the connection is used, e.g. mounted
+  into the worker/scheduler pod via a Secret or ConfigMap volume.
+
 Disable TCP keepalive
   TCP keepalive is a feature (enabled by default) that tries to keep long-running connections
   alive. Set this parameter to True to disable this feature.

providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py

@@ -136,6 +136,9 @@ class KubernetesHook(BaseHook, PodOperatorHookProtocol):
     :param in_cluster: Set to ``True`` if running from within a kubernetes cluster.
     :param disable_verify_ssl: Set to ``True`` if SSL verification should be disabled.
     :param disable_tcp_keepalive: Set to ``True`` if you want to disable keepalive logic.
+    :param ssl_ca_cert: Path to a CA bundle used to verify the Kubernetes API server
+        certificate, for clusters whose certificate is signed by a custom or intermediate CA.
+        Ignored when ``disable_verify_ssl`` is set.
     """
 
     conn_name_attr = "kubernetes_conn_id"
@@ -162,6 +165,7 @@ def get_connection_form_widgets(cls) -> dict[str, Any]:
             "cluster_context": StringField(lazy_gettext("Cluster context"), widget=BS3TextFieldWidget()),
             "disable_verify_ssl": BooleanField(lazy_gettext("Disable SSL")),
             "disable_tcp_keepalive": BooleanField(lazy_gettext("Disable TCP keepalive")),
+            "ssl_ca_cert": StringField(lazy_gettext("SSL CA cert"), widget=BS3TextFieldWidget()),
             "xcom_sidecar_container_image": StringField(
                 lazy_gettext("XCom sidecar image"), widget=BS3TextFieldWidget()
             ),
@@ -192,6 +196,7 @@ def __init__(
         in_cluster: bool | None = None,
         disable_verify_ssl: bool | None = None,
         disable_tcp_keepalive: bool | None = None,
+        ssl_ca_cert: str | None = None,
     ) -> None:
         super().__init__()
         self.conn_id = conn_id or kubernetes_conn_id
@@ -202,6 +207,7 @@ def __init__(
         self.in_cluster = in_cluster
         self.disable_verify_ssl = disable_verify_ssl
         self.disable_tcp_keepalive = disable_tcp_keepalive
+        self.ssl_ca_cert = ssl_ca_cert
         self._is_in_cluster: bool | None = None
 
     @staticmethod
@@ -275,6 +281,7 @@ def get_conn(self) -> client.ApiClient:
         disable_tcp_keepalive = self._coalesce_param(
             self.disable_tcp_keepalive, _get_bool(self._get_field("disable_tcp_keepalive"))
         )
+        ssl_ca_cert = self._coalesce_param(self.ssl_ca_cert, self._get_field("ssl_ca_cert"))
 
         if disable_verify_ssl is True:
             _disable_verify_ssl()
@@ -288,6 +295,7 @@ def get_conn(self) -> client.ApiClient:
             return _TimeoutK8sApiClient(
                 configuration=self.client_configuration,
                 disable_verify_ssl=disable_verify_ssl is True,
+                ssl_ca_cert=ssl_ca_cert,
             )
 
         if kubeconfig_path is not None:
@@ -301,6 +309,7 @@ def get_conn(self) -> client.ApiClient:
             return _TimeoutK8sApiClient(
                 configuration=self.client_configuration,
                 disable_verify_ssl=disable_verify_ssl is True,
+                ssl_ca_cert=ssl_ca_cert,
             )
 
         if kubeconfig is not None:
@@ -319,6 +328,7 @@ def get_conn(self) -> client.ApiClient:
             return _TimeoutK8sApiClient(
                 configuration=self.client_configuration,
                 disable_verify_ssl=disable_verify_ssl is True,
+                ssl_ca_cert=ssl_ca_cert,
             )
 
         if self.config_dict:
@@ -332,14 +342,21 @@ def get_conn(self) -> client.ApiClient:
             return _TimeoutK8sApiClient(
                 configuration=self.client_configuration,
                 disable_verify_ssl=disable_verify_ssl is True,
+                ssl_ca_cert=ssl_ca_cert,
             )
 
         return self._get_default_client(
-            cluster_context=cluster_context, disable_verify_ssl=disable_verify_ssl
+            cluster_context=cluster_context,
+            disable_verify_ssl=disable_verify_ssl,
+            ssl_ca_cert=ssl_ca_cert,
         )
 
     def _get_default_client(
-        self, *, cluster_context: str | None = None, disable_verify_ssl: bool | None = None
+        self,
+        *,
+        cluster_context: str | None = None,
+        disable_verify_ssl: bool | None = None,
+        ssl_ca_cert: str | None = None,
     ) -> client.ApiClient:
         # if we get here, then no configuration has been supplied
         # we should try in_cluster since that's most likely
@@ -358,6 +375,7 @@ def _get_default_client(
         return _TimeoutK8sApiClient(
             configuration=self.client_configuration,
             disable_verify_ssl=disable_verify_ssl is True,
+            ssl_ca_cert=ssl_ca_cert,
         )
 
     @property

providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/kube_client.py

@@ -62,19 +62,26 @@ class _TimeoutK8sApiClient(client.ApiClient):
 
         When *disable_verify_ssl* is True the TLS certificate check is turned off
         on the *client_configuration* that is passed (or on a fresh default copy)
-        so that callers do not need to repeat this logic at every call-site.
+        so that callers do not need to repeat this logic at every call-site. Likewise,
+        when *ssl_ca_cert* is provided, it is set as the CA bundle used to verify the
+        Kubernetes API server certificate, so a custom/intermediate CA can be trusted
+        without disabling verification altogether.
         """
 
         def __init__(
             self,
             configuration: client.Configuration | None = None,
             *,
             disable_verify_ssl: bool = False,
+            ssl_ca_cert: str | None = None,
         ) -> None:
-            if disable_verify_ssl:
+            if disable_verify_ssl or ssl_ca_cert:
                 if configuration is None:
                     configuration = client.Configuration.get_default_copy()
-                configuration.verify_ssl = False
+                if disable_verify_ssl:
+                    configuration.verify_ssl = False
+                if ssl_ca_cert:
+                    configuration.ssl_ca_cert = ssl_ca_cert
             super().__init__(configuration=configuration)
 
         def call_api(self, *args: Any, **kwargs: Any) -> Any:

providers/cncf/kubernetes/tests/unit/cncf/kubernetes/hooks/test_kubernetes.py

@@ -126,6 +126,24 @@ def test_call_api_timeout_inject(self, kwargs, expected_timeout):
             assert call_kwargs["_request_timeout"] == expected_timeout
             assert out == "ok"
 
+    @pytest.mark.parametrize(
+        ("kwargs", "expected_verify_ssl", "expected_ssl_ca_cert"),
+        [
+            pytest.param({"ssl_ca_cert": "/path/to/ca.pem"}, True, "/path/to/ca.pem", id="ssl-ca-cert-only"),
+            pytest.param(
+                {"ssl_ca_cert": "/path/to/ca.pem", "disable_verify_ssl": True},
+                False,
+                "/path/to/ca.pem",
+                id="ssl-ca-cert-and-disable-verify-ssl",
+            ),
+        ],
+    )
+    def test_ssl_ca_cert_sets_configuration(self, kwargs, expected_verify_ssl, expected_ssl_ca_cert):
+        """ssl_ca_cert is applied to the client configuration, independently of disable_verify_ssl."""
+        cli = _TimeoutK8sApiClient(**kwargs)
+        assert cli.configuration.ssl_ca_cert == expected_ssl_ca_cert
+        assert cli.configuration.verify_ssl is expected_verify_ssl
+
 
 class TestTimeoutAsyncK8sApiClient:
     @pytest.mark.asyncio
@@ -222,6 +240,8 @@ def setup_connections(self, create_connection_without_db):
             ("default_kube_config", {}),
             ("disable_verify_ssl", {"disable_verify_ssl": True}),
             ("disable_verify_ssl_empty", {"disable_verify_ssl": ""}),
+            ("ssl_ca_cert", {"ssl_ca_cert": "/extra/ca.pem"}),
+            ("ssl_ca_cert_empty", {"ssl_ca_cert": ""}),
             ("disable_tcp_keepalive", {"disable_tcp_keepalive": True}),
             ("disable_tcp_keepalive_empty", {"disable_tcp_keepalive": ""}),
             ("sidecar_container_image", {"xcom_sidecar_container_image": "private.repo.com/alpine:3.16"}),
@@ -413,6 +433,62 @@ def test_disable_verify_ssl_applies_to_client_configuration(self, config_source)
         assert isinstance(api_conn, kubernetes.client.api_client.ApiClient)
         assert api_conn.configuration.verify_ssl is False
 
+    @pytest.mark.parametrize(
+        ("ssl_ca_cert_param", "conn_id", "expected"),
+        (
+            ("/param/ca.pem", None, "/param/ca.pem"),
+            (None, "ssl_ca_cert", "/extra/ca.pem"),
+            ("/param/ca.pem", "ssl_ca_cert", "/param/ca.pem"),
+            (None, "ssl_ca_cert_empty", None),
+        ),
+    )
+    @patch("kubernetes.config.incluster_config.InClusterConfigLoader", new=MagicMock())
+    def test_ssl_ca_cert(self, ssl_ca_cert_param, conn_id, expected):
+        """
+        Verifies that ssl_ca_cert from the hook param or connection extra is applied to the
+        returned ApiClient's configuration. Hook param should beat extra.
+        """
+        kubernetes_hook = KubernetesHook(conn_id=conn_id, ssl_ca_cert=ssl_ca_cert_param)
+        api_conn = kubernetes_hook.get_conn()
+        assert isinstance(api_conn, kubernetes.client.api_client.ApiClient)
+        assert api_conn.configuration.ssl_ca_cert == expected
+
+    @pytest.mark.parametrize(
+        "config_source",
+        [
+            pytest.param("in_cluster", id="in_cluster"),
+            pytest.param("kube_config_path", id="kube_config_path"),
+            pytest.param("kube_config", id="kube_config"),
+            pytest.param("config_dict", id="config_dict"),
+            pytest.param("default", id="default_client"),
+        ],
+    )
+    @patch("kubernetes.config.incluster_config.InClusterConfigLoader", new=MagicMock())
+    @patch("kubernetes.config.kube_config.KubeConfigLoader", new=MagicMock())
+    @patch("kubernetes.config.kube_config.KubeConfigMerger", new=MagicMock())
+    def test_ssl_ca_cert_applies_to_client_configuration(self, config_source):
+        """
+        Verifies that ssl_ca_cert is propagated to the returned ApiClient's configuration
+        regardless of which configuration-loading branch of get_conn() is taken.
+        """
+        if config_source == "in_cluster":
+            kubernetes_hook = KubernetesHook(conn_id="in_cluster", ssl_ca_cert="/param/ca.pem")
+        elif config_source == "kube_config_path":
+            kubernetes_hook = KubernetesHook(conn_id="kube_config_path", ssl_ca_cert="/param/ca.pem")
+        elif config_source == "kube_config":
+            kubernetes_hook = KubernetesHook(conn_id="kube_config", ssl_ca_cert="/param/ca.pem")
+        elif config_source == "config_dict":
+            kubernetes_hook = KubernetesHook(
+                config_dict={"apiVersion": "v1", "kind": "Config"},
+                ssl_ca_cert="/param/ca.pem",
+            )
+        else:
+            kubernetes_hook = KubernetesHook(ssl_ca_cert="/param/ca.pem")
+
+        api_conn = kubernetes_hook.get_conn()
+        assert isinstance(api_conn, kubernetes.client.api_client.ApiClient)
+        assert api_conn.configuration.ssl_ca_cert == "/param/ca.pem"
+
     @pytest.mark.parametrize(
         ("disable_tcp_keepalive", "conn_id", "expected"),
         (
@@ -619,7 +695,9 @@ def test_prefixed_names_still_work(self, mock_get_client):
         with mock.patch.dict("os.environ", AIRFLOW_CONN_KUBERNETES_DEFAULT=conn_uri):
             kubernetes_hook = KubernetesHook(conn_id="kubernetes_default")
             kubernetes_hook.get_conn()
-            mock_get_client.assert_called_with(cluster_context="test", disable_verify_ssl=None)
+            mock_get_client.assert_called_with(
+                cluster_context="test", disable_verify_ssl=None, ssl_ca_cert=None
+            )
             assert kubernetes_hook.get_namespace() == "test"
 
     def test_missing_default_connection_is_ok(self, remove_default_conn, sdk_connection_not_found):