diff --git a/providers/cncf/kubernetes/docs/connections/kubernetes.rst b/providers/cncf/kubernetes/docs/connections/kubernetes.rst index 66b13263ccb8b..27e63e6b74463 100644 --- a/providers/cncf/kubernetes/docs/connections/kubernetes.rst +++ b/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. diff --git a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py index bccf6093ea067..a1955cff341f2 100644 --- a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py +++ b/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 diff --git a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/kube_client.py b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/kube_client.py index 08f91abe73ee9..869488bdf85d9 100644 --- a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/kube_client.py +++ b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/kube_client.py @@ -62,7 +62,10 @@ 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__( @@ -70,11 +73,15 @@ def __init__( 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: diff --git a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/hooks/test_kubernetes.py b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/hooks/test_kubernetes.py index 09d686821cacd..4b3afc51ed022 100644 --- a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/hooks/test_kubernetes.py +++ b/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):