This is an automated email from the ASF dual-hosted git repository.
amoghdesai pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 26ead210674 Do not cache kubeconfig for exec based auth in
AsyncKubernetesHook (#65212)
26ead210674 is described below
commit 26ead2106741c5b36a3d22ca894408f4aaa4bfc0
Author: Amogh Desai <[email protected]>
AuthorDate: Thu Apr 16 20:42:51 2026 +0530
Do not cache kubeconfig for exec based auth in AsyncKubernetesHook (#65212)
---
.../providers/cncf/kubernetes/hooks/kubernetes.py | 115 +++++++++++++---
.../unit/cncf/kubernetes/hooks/test_kubernetes.py | 152 +++++++++++++++++++++
2 files changed, 247 insertions(+), 20 deletions(-)
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 23b4cc94b1f..3f740981aa7 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
@@ -855,9 +855,45 @@ class AsyncKubernetesHook(KubernetesHook):
self._extras: dict | None = connection_extras
self._event_polling_fallback = False
self._config_loaded = False
+ # Cached result of exec-auth detection. None means not yet detected.
+ # This is to optimise and not calling _uses_exec_auth repeatedly on
every _load_config() call.
+ self._is_exec_auth: bool | None = None
+
+ def _uses_exec_auth(self, kubeconfig_data: dict, context: str | None =
None) -> bool:
+ """
+ Detect if the active kubeconfig context uses exec-based authentication.
+
+ Exec plugins return short-lived tokens (EKS, GKE, etc). Only the user
+ tied to the active context is checked. If the active context or its
user
+ cannot be resolved, falls back to scanning all users and returns True
if
+ any use exec auth — erring on the side of not caching.
+ """
+ active_context = context or kubeconfig_data.get("current-context")
+
+ active_user: str | None = None
+ for ctx in kubeconfig_data.get("contexts", []):
+ if ctx.get("name") == active_context:
+ active_user = (ctx.get("context") or {}).get("user")
+ break
+
+ users = kubeconfig_data.get("users", [])
+ if active_user is not None:
+ for user in users:
+ if user.get("name") == active_user:
+ return "exec" in (user.get("user") or {})
+ return False
+
+ # fallback to check all users if active context user cannot be
resolved; this is a safe fallback since it errs on the side of not caching
+ return any("exec" in (u.get("user") or {}) for u in users)
async def _load_config(self):
- """Load Kubernetes configuration once per hook instance."""
+ """
+ Load Kubernetes configuration.
+
+ For static auth (token, certificate), configuration is loaded once per
hook instance
+ and cached. For exec-based auth (EKS, GKE), the config is reloaded on
every call so
+ that short-lived tokens are always refreshed.
+ """
if self._config_loaded:
return
@@ -884,39 +920,68 @@ class AsyncKubernetesHook(KubernetesHook):
self._config_loaded = True
return
- # If above block does not return, we are not in a cluster.
self._is_in_cluster = False
-
if self.config_dict:
self.log.debug(LOADING_KUBE_CONFIG_FILE_RESOURCE.format("config
dictionary"))
- await async_config.load_kube_config_from_dict(self.config_dict,
context=cluster_context)
- self._config_loaded = True
- return
+ await async_config.load_kube_config_from_dict(
+ self.config_dict,
+ context=cluster_context,
+ )
+
+ if self._is_exec_auth is None:
+ self._is_exec_auth = self._uses_exec_auth(self.config_dict,
context=cluster_context)
+
+ if not self._is_exec_auth:
+ self._config_loaded = True
+
+ return
if kubeconfig_path is not None:
self.log.debug("loading kube_config from: %s", kubeconfig_path)
+
await async_config.load_kube_config(
config_file=kubeconfig_path,
client_configuration=self.client_configuration,
context=cluster_context,
)
- self._config_loaded = True
- return
+ if self._is_exec_auth is None:
+ try:
+ async with aiofiles.open(kubeconfig_path) as f:
+ content = await f.read()
+ data = yaml.safe_load(content)
+ self._is_exec_auth = self._uses_exec_auth(data,
context=cluster_context)
+ except Exception as exc:
+ self.log.warning(
+ "Error while parsing kube_config from %s to detect
exec auth; "
+ "continuing without caching the config: %s",
+ kubeconfig_path,
+ exc,
+ )
+ self._is_exec_auth = True
+
+ if not self._is_exec_auth:
+ self._config_loaded = True
+
+ return
if kubeconfig is not None:
async with aiofiles.tempfile.NamedTemporaryFile() as temp_config:
- self.log.debug(
- "Reading kubernetes configuration file from connection "
- "object and writing temporary config file with its
content",
- )
if isinstance(kubeconfig, dict):
- self.log.debug(
- LOADING_KUBE_CONFIG_FILE_RESOURCE.format(
- "connection kube_config dictionary (serializing)"
+ kubeconfig_data = kubeconfig
+ kubeconfig_str = json.dumps(kubeconfig)
+ else:
+ try:
+ kubeconfig_data = _load_body_to_dict(kubeconfig)
+ except AirflowException as exc:
+ self.log.debug(
+ "Could not parse kubeconfig string to detect exec
auth; "
+ "config will not be cached: %s",
+ exc,
)
- )
- kubeconfig = json.dumps(kubeconfig)
- await temp_config.write(kubeconfig.encode())
+ kubeconfig_data = None
+ kubeconfig_str = kubeconfig
+
+ await temp_config.write(kubeconfig_str.encode())
await temp_config.flush()
await async_config.load_kube_config(
@@ -924,10 +989,20 @@ class AsyncKubernetesHook(KubernetesHook):
client_configuration=self.client_configuration,
context=cluster_context,
)
- self._config_loaded = True
- return
+ if self._is_exec_auth is None:
+ self._is_exec_auth = (
+ self._uses_exec_auth(kubeconfig_data,
context=cluster_context)
+ if kubeconfig_data
+ else True
+ )
+
+ if not self._is_exec_auth:
+ self._config_loaded = True
+
+ return
self.log.debug(LOADING_KUBE_CONFIG_FILE_RESOURCE.format("default
configuration file"))
+
await async_config.load_kube_config(
client_configuration=self.client_configuration,
context=cluster_context,
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 850610ace27..2d38c697954 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
@@ -1133,6 +1133,158 @@ class TestAsyncKubernetesHook:
with pytest.raises(AirflowException):
await hook._load_config()
+ @pytest.mark.parametrize(
+ ("kubeconfig_data", "context", "expected"),
+ [
+ pytest.param(
+ {
+ "current-context": "ctx1",
+ "contexts": [{"name": "ctx1", "context": {"user":
"user1"}}],
+ "users": [{"name": "user1", "user": {"exec": {"command":
"aws"}}}],
+ },
+ None,
+ True,
+ id="active_context_user_has_exec",
+ ),
+ pytest.param(
+ {
+ "current-context": "ctx1",
+ "contexts": [{"name": "ctx1", "context": {"user":
"user1"}}],
+ "users": [{"name": "user1", "user": {"token": "abc123"}}],
+ },
+ None,
+ False,
+ id="active_context_user_has_static_token",
+ ),
+ pytest.param(
+ {
+ "current-context": "ctx1",
+ "contexts": [{"name": "ctx1", "context": {"user":
"user1"}}],
+ "users": [
+ {"name": "user1", "user": {"token": "abc123"}},
+ {"name": "user2", "user": {"exec": {"command":
"aws"}}},
+ ],
+ },
+ None,
+ False,
+ id="only_inactive_user_has_exec",
+ ),
+ pytest.param(
+ {
+ "users": [{"name": "user1", "user": {"exec": {"command":
"aws"}}}],
+ },
+ None,
+ True,
+ id="no_contexts_fallback_any_user_has_exec",
+ ),
+ ],
+ )
+ def test_uses_exec_auth(self, kubeconfig_data, context, expected):
+ hook = AsyncKubernetesHook(conn_id=None, in_cluster=False)
+ assert hook._uses_exec_auth(kubeconfig_data, context=context) is
expected
+
+ @pytest.mark.asyncio
+
@mock.patch("airflow.providers.cncf.kubernetes.hooks.kubernetes.async_config.load_kube_config_from_dict")
+ async def test_load_config_no_caching_when_exec_auth(self, mock_load):
+ mock_load.return_value = None
+ exec_config = {
+ "current-context": "ctx1",
+ "contexts": [{"name": "ctx1", "context": {"user": "user1"}}],
+ "users": [{"name": "user1", "user": {"exec": {"command": "aws eks
get-token"}}}],
+ }
+ hook = AsyncKubernetesHook(conn_id=None, in_cluster=False,
config_dict=exec_config)
+ await hook._load_config()
+ assert hook._config_loaded is False
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize(
+ ("kubeconfig_content", "expected_cached"),
+ [
+ pytest.param(
+ (
+ "current-context: ctx1\n"
+ "contexts:\n- name: ctx1\n context:\n user: user1\n"
+ "users:\n- name: user1\n user:\n exec:\n command:
aws eks get-token\n"
+ ),
+ False,
+ id="kube_config_path_with_exec",
+ ),
+ pytest.param(
+ (
+ "current-context: ctx1\n"
+ "contexts:\n- name: ctx1\n context:\n user: user1\n"
+ "users:\n- name: user1\n user:\n token: static-token\n"
+ ),
+ True,
+ id="kube_config_path_no_exec",
+ ),
+ ],
+ )
+
@mock.patch("airflow.providers.cncf.kubernetes.hooks.kubernetes.async_config.load_kube_config")
+ async def test_load_config_caching_behavior_kube_config_path(
+ self, mock_load_file, tmp_path, kubeconfig_content, expected_cached
+ ):
+ mock_load_file.return_value = None
+ kubeconfig_file = tmp_path / "kubeconfig.yaml"
+ kubeconfig_file.write_text(kubeconfig_content)
+
+ hook = AsyncKubernetesHook(conn_id=None, in_cluster=False)
+ hook._get_field = mock.AsyncMock(
+ side_effect=lambda field: str(kubeconfig_file) if field ==
"kube_config_path" else None
+ )
+ await hook._load_config()
+ assert hook._config_loaded is expected_cached
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize(
+ ("kubeconfig", "expected_cached"),
+ [
+ pytest.param(
+ {
+ "current-context": "ctx1",
+ "contexts": [{"name": "ctx1", "context": {"user":
"user1"}}],
+ "users": [{"name": "user1", "user": {"token":
"static-token"}}],
+ },
+ True,
+ id="dict_kubeconfig_no_exec",
+ ),
+ pytest.param(
+ (
+ "current-context: ctx1\n"
+ "contexts:\n- name: ctx1\n context:\n user: user1\n"
+ "users:\n- name: user1\n user:\n exec:\n command:
aws\n"
+ ),
+ False,
+ id="string_kubeconfig_with_exec",
+ ),
+ pytest.param(
+ (
+ "current-context: ctx1\n"
+ "contexts:\n- name: ctx1\n context:\n user: user1\n"
+ "users:\n- name: user1\n user:\n token: static-token\n"
+ ),
+ True,
+ id="string_kubeconfig_no_exec",
+ ),
+ ],
+ )
+
@mock.patch("airflow.providers.cncf.kubernetes.hooks.kubernetes.async_config.load_kube_config")
+
@mock.patch("airflow.providers.cncf.kubernetes.hooks.kubernetes.async_config.load_kube_config_from_dict")
+ async def test_load_config_caching_behavior(
+ self, mock_load_dict, mock_load_file, kubeconfig, expected_cached
+ ):
+ mock_load_dict.return_value = None
+ mock_load_file.return_value = None
+ if isinstance(kubeconfig, dict):
+ hook = AsyncKubernetesHook(conn_id=None, in_cluster=False,
config_dict=kubeconfig)
+ else:
+ hook = AsyncKubernetesHook(conn_id=None, in_cluster=False)
+ hook._get_field = mock.AsyncMock(
+ side_effect=lambda field: kubeconfig if field == "kube_config"
else None
+ )
+ await hook._load_config()
+ assert hook._config_loaded is expected_cached
+
@pytest.mark.asyncio
@mock.patch(KUBE_API.format("list_namespaced_event"))
async def test_async_get_pod_events_with_resource_version(
