You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by vi...@apache.org on 2026/04/23 20:52:52 UTC
(airflow) branch main updated: Add multi-team lookup to Kubernetes secrets backend (#65694)
This is an automated email from the ASF dual-hosted git repository.
vincbeck 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 5d6f1c12c6f Add multi-team lookup to Kubernetes secrets backend (#65694)
5d6f1c12c6f is described below
commit 5d6f1c12c6f531963c8b86460336d24246baf13c
Author: PrithviBadiga <45...@users.noreply.github.com>
AuthorDate: Thu Apr 23 13:52:45 2026 -0700
Add multi-team lookup to Kubernetes secrets backend (#65694)
---
.../kubernetes-secrets-backend.rst | 60 +++++++++++
.../secrets/kubernetes_secrets_backend.py | 77 ++++++++++----
.../secrets/test_kubernetes_secrets_backend.py | 118 ++++++++++++++++++++-
3 files changed, 230 insertions(+), 25 deletions(-)
diff --git a/providers/cncf/kubernetes/docs/secrets-backends/kubernetes-secrets-backend.rst b/providers/cncf/kubernetes/docs/secrets-backends/kubernetes-secrets-backend.rst
index 4f68de3128d..5c75078e7b4 100644
--- a/providers/cncf/kubernetes/docs/secrets-backends/kubernetes-secrets-backend.rst
+++ b/providers/cncf/kubernetes/docs/secrets-backends/kubernetes-secrets-backend.rst
@@ -81,6 +81,7 @@ The following parameters can be passed via ``backend_kwargs`` as a JSON dictiona
* ``connections_label``: Label key used to discover connection secrets. Default: ``"airflow.apache.org/connection-id"``
* ``variables_label``: Label key used to discover variable secrets. Default: ``"airflow.apache.org/variable-key"``
* ``config_label``: Label key used to discover config secrets. Default: ``"airflow.apache.org/config-key"``
+* ``team_label``: Label key used to discover team-scoped secrets in multi-team mode. Default: ``"airflow.apache.org/team"``
* ``connections_data_key``: The data key in the Kubernetes secret that holds the connection value. Default: ``"value"``
* ``variables_data_key``: The data key in the Kubernetes secret that holds the variable value. Default: ``"value"``
* ``config_data_key``: The data key in the Kubernetes secret that holds the config value. Default: ``"value"``
@@ -207,6 +208,65 @@ You can create a variable secret with ``kubectl``:
airflow.apache.org/variable-key=my_var \
--namespace=airflow
+Multi-team lookup
+"""""""""""""""""
+
+In multi-team mode, when ``team_name`` is provided, this backend first looks for a secret whose
+identifier label matches the requested connection or variable and whose ``team_label`` matches the
+current team. If no team-scoped secret is found, it falls back to a global secret with the same
+identifier label and no team label.
+
+To use this mode, keep the backend enabled as usual and make sure your Kubernetes secrets include
+both:
+
+* the regular identifier label (for example ``airflow.apache.org/connection-id=my_db``)
+* the configured ``team_label`` with the team name (for example ``airflow.apache.org/team=team_a``)
+
+For example, this configuration keeps the default team label:
+
+.. code-block:: ini
+
+ [secrets]
+ backend = airflow.providers.cncf.kubernetes.secrets.kubernetes_secrets_backend.KubernetesSecretsBackend
+ backend_kwargs = {"team_label": "airflow.apache.org/team"}
+
+If you use a custom team label instead, configure it in ``backend_kwargs`` and apply the same label
+to your Kubernetes secrets.
+
+For example, with ``team_label="airflow.apache.org/team"``, ``team_name="team_a"``, and
+``conn_id="my_db"``, the backend queries:
+
+* Team-scoped: ``airflow.apache.org/connection-id=my_db,airflow.apache.org/team=team_a``
+* Global fallback: ``airflow.apache.org/connection-id=my_db,!airflow.apache.org/team``
+
+Example team-scoped connection secret:
+
+.. code-block:: yaml
+
+ apiVersion: v1
+ kind: Secret
+ metadata:
+ name: my-team-db-secret
+ labels:
+ airflow.apache.org/connection-id: my_db
+ airflow.apache.org/team: team_a
+ data:
+ value: <base64-encoded-connection-uri>
+
+If you also want a default value for workloads that do not run with a team context, create a second
+secret with the same connection identifier label but without ``airflow.apache.org/team``:
+
+.. code-block:: yaml
+
+ apiVersion: v1
+ kind: Secret
+ metadata:
+ name: my-global-db-secret
+ labels:
+ airflow.apache.org/connection-id: my_db
+ data:
+ value: <base64-encoded-global-connection-uri>
+
Using with External Secrets Operator
"""""""""""""""""""""""""""""""""""""
diff --git a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/secrets/kubernetes_secrets_backend.py b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/secrets/kubernetes_secrets_backend.py
index ea67f7c3da6..447a4c48a9e 100644
--- a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/secrets/kubernetes_secrets_backend.py
+++ b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/secrets/kubernetes_secrets_backend.py
@@ -20,6 +20,7 @@
from __future__ import annotations
import base64
+import re
from functools import cached_property
from pathlib import Path
@@ -96,14 +97,16 @@ class KubernetesSecretsBackend(BaseSecretsBackend, LoggingMixin):
DEFAULT_CONNECTIONS_LABEL = "airflow.apache.org/connection-id"
DEFAULT_VARIABLES_LABEL = "airflow.apache.org/variable-key"
DEFAULT_CONFIG_LABEL = "airflow.apache.org/config-key"
+ DEFAULT_TEAM_LABEL = "airflow.apache.org/team"
SERVICE_ACCOUNT_NAMESPACE_PATH = "/var/run/secrets/kubernetes.io/serviceaccount/namespace"
def __init__(
self,
namespace: str | None = None,
- connections_label: str = DEFAULT_CONNECTIONS_LABEL,
- variables_label: str = DEFAULT_VARIABLES_LABEL,
- config_label: str = DEFAULT_CONFIG_LABEL,
+ connections_label: str | None = DEFAULT_CONNECTIONS_LABEL,
+ variables_label: str | None = DEFAULT_VARIABLES_LABEL,
+ config_label: str | None = DEFAULT_CONFIG_LABEL,
+ team_label: str | None = DEFAULT_TEAM_LABEL,
connections_data_key: str = "value",
variables_data_key: str = "value",
config_data_key: str = "value",
@@ -114,6 +117,7 @@ class KubernetesSecretsBackend(BaseSecretsBackend, LoggingMixin):
self.connections_label = connections_label
self.variables_label = variables_label
self.config_label = config_label
+ self.team_label = team_label
self.connections_data_key = connections_data_key
self.variables_data_key = variables_data_key
self.config_data_key = config_data_key
@@ -143,26 +147,28 @@ class KubernetesSecretsBackend(BaseSecretsBackend, LoggingMixin):
"""
Get serialized representation of Connection from a Kubernetes secret.
- Multi-team isolation is not currently supported; ``team_name`` is accepted
- for API compatibility but ignored.
-
:param conn_id: connection id
- :param team_name: Team name (unused — multi-team is not currently supported)
+ :param team_name: Team name associated to the task trying to access the connection (if any)
"""
- return self._get_secret(self.connections_label, conn_id, self.connections_data_key)
+ if self._is_team_specific_accessed_as_global(conn_id, team_name):
+ return None
+
+ return self._get_secret(
+ self.connections_label, conn_id, self.connections_data_key, team_name=team_name
+ )
def get_variable(self, key: str, team_name: str | None = None) -> str | None:
"""
Get Airflow Variable from a Kubernetes secret.
- Multi-team isolation is not currently supported; ``team_name`` is accepted
- for API compatibility but ignored.
-
:param key: Variable Key
- :param team_name: Team name (unused — multi-team is not currently supported)
+ :param team_name: Team name associated to the task trying to access the variable (if any)
:return: Variable Value
"""
- return self._get_secret(self.variables_label, key, self.variables_data_key)
+ if self._is_team_specific_accessed_as_global(key, team_name):
+ return None
+
+ return self._get_secret(self.variables_label, key, self.variables_data_key, team_name=team_name)
def get_config(self, key: str) -> str | None:
"""
@@ -173,7 +179,13 @@ class KubernetesSecretsBackend(BaseSecretsBackend, LoggingMixin):
"""
return self._get_secret(self.config_label, key, self.config_data_key)
- def _get_secret(self, label_key: str | None, label_value: str, data_key: str) -> str | None:
+ @staticmethod
+ def _is_team_specific_accessed_as_global(secret_id: str, team_name: str | None = None) -> bool:
+ return team_name is None and bool(re.fullmatch(r"_[^_]+___.+", secret_id))
+
+ def _get_secret(
+ self, label_key: str | None, label_value: str, data_key: str, team_name: str | None = None
+ ) -> str | None:
"""
Get secret value from Kubernetes by label selector.
@@ -188,18 +200,43 @@ class KubernetesSecretsBackend(BaseSecretsBackend, LoggingMixin):
"""
if label_key is None:
return None
- label_selector = f"{label_key}={label_value}"
+
+ if team_name and self.team_label:
+ team_secret = self._get_secret_by_selector(
+ label_key, label_value, data_key, f"{self.team_label}={team_name}", warn_if_missing=False
+ )
+ if team_secret is not None:
+ return team_secret
+
+ team_selector = f"!{self.team_label}" if self.team_label else None
+ return self._get_secret_by_selector(label_key, label_value, data_key, team_selector)
+
+ def _get_secret_by_selector(
+ self,
+ label_key: str,
+ label_value: str,
+ data_key: str,
+ extra_selector: str | None,
+ *,
+ warn_if_missing: bool = True,
+ ) -> str | None:
+ """Get secret value from Kubernetes by the given base and optional extra selectors."""
+ selectors = [f"{label_key}={label_value}"]
+ if extra_selector:
+ selectors.append(extra_selector)
+ label_selector = ",".join(selectors)
secret_list = self.client.list_namespaced_secret(
self.namespace,
label_selector=label_selector,
resource_version="0",
)
if not secret_list.items:
- self.log.warning(
- "No secret found with label %s in namespace %s.",
- label_selector,
- self.namespace,
- )
+ if warn_if_missing:
+ self.log.warning(
+ "No secret found with label %s in namespace %s.",
+ label_selector,
+ self.namespace,
+ )
return None
if len(secret_list.items) > 1:
self.log.warning(
diff --git a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/secrets/test_kubernetes_secrets_backend.py b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/secrets/test_kubernetes_secrets_backend.py
index 6dced3d340b..e7538631b24 100644
--- a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/secrets/test_kubernetes_secrets_backend.py
+++ b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/secrets/test_kubernetes_secrets_backend.py
@@ -63,7 +63,7 @@ class TestKubernetesSecretsBackendConnections:
assert result == uri
mock_client.return_value.list_namespaced_secret.assert_called_once_with(
"default",
- label_selector="airflow.apache.org/connection-id=my_db",
+ label_selector="airflow.apache.org/connection-id=my_db,!airflow.apache.org/team",
resource_version="0",
)
@@ -103,6 +103,60 @@ class TestKubernetesSecretsBackendConnections:
assert result is None
+ @mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default")
+ @mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock)
+ def test_get_conn_value_uses_team_specific_secret_first(self, mock_client, mock_namespace):
+ mock_client.return_value.list_namespaced_secret.side_effect = [
+ _make_secret_list([_make_secret({"value": "team-conn"})]),
+ ]
+
+ backend = KubernetesSecretsBackend()
+ result = backend.get_conn_value("my_db", team_name="team_a")
+
+ assert result == "team-conn"
+ mock_client.return_value.list_namespaced_secret.assert_called_once_with(
+ "default",
+ label_selector="airflow.apache.org/connection-id=my_db,airflow.apache.org/team=team_a",
+ resource_version="0",
+ )
+
+ @mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default")
+ @mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock)
+ def test_get_conn_value_falls_back_to_global_secret_when_team_secret_is_missing(
+ self, mock_client, mock_namespace
+ ):
+ mock_client.return_value.list_namespaced_secret.side_effect = [
+ _make_secret_list([]),
+ _make_secret_list([_make_secret({"value": "global-conn"})]),
+ ]
+
+ backend = KubernetesSecretsBackend()
+ result = backend.get_conn_value("my_db", team_name="team_a")
+
+ assert result == "global-conn"
+ assert mock_client.return_value.list_namespaced_secret.call_args_list == [
+ mock.call(
+ "default",
+ label_selector="airflow.apache.org/connection-id=my_db,airflow.apache.org/team=team_a",
+ resource_version="0",
+ ),
+ mock.call(
+ "default",
+ label_selector="airflow.apache.org/connection-id=my_db,!airflow.apache.org/team",
+ resource_version="0",
+ ),
+ ]
+
+ @mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default")
+ @mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock)
+ def test_get_conn_value_returns_none_for_team_scoped_id_without_team_name(
+ self, mock_client, mock_namespace
+ ):
+ backend = KubernetesSecretsBackend()
+
+ assert backend.get_conn_value("_teama___my_db") is None
+ mock_client.return_value.list_namespaced_secret.assert_not_called()
+
class TestKubernetesSecretsBackendVariables:
@mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default")
@@ -119,7 +173,7 @@ class TestKubernetesSecretsBackendVariables:
assert result == "my-value"
mock_client.return_value.list_namespaced_secret.assert_called_once_with(
"default",
- label_selector="airflow.apache.org/variable-key=api_key",
+ label_selector="airflow.apache.org/variable-key=api_key,!airflow.apache.org/team",
resource_version="0",
)
@@ -134,6 +188,43 @@ class TestKubernetesSecretsBackendVariables:
assert result is None
+ @mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default")
+ @mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock)
+ def test_get_variable_falls_back_to_global_secret_when_team_secret_is_missing(
+ self, mock_client, mock_namespace
+ ):
+ mock_client.return_value.list_namespaced_secret.side_effect = [
+ _make_secret_list([]),
+ _make_secret_list([_make_secret({"value": "global-value"})]),
+ ]
+
+ backend = KubernetesSecretsBackend()
+ result = backend.get_variable("api_key", team_name="team_a")
+
+ assert result == "global-value"
+ assert mock_client.return_value.list_namespaced_secret.call_args_list == [
+ mock.call(
+ "default",
+ label_selector="airflow.apache.org/variable-key=api_key,airflow.apache.org/team=team_a",
+ resource_version="0",
+ ),
+ mock.call(
+ "default",
+ label_selector="airflow.apache.org/variable-key=api_key,!airflow.apache.org/team",
+ resource_version="0",
+ ),
+ ]
+
+ @mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default")
+ @mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock)
+ def test_get_variable_returns_none_for_team_scoped_key_without_team_name(
+ self, mock_client, mock_namespace
+ ):
+ backend = KubernetesSecretsBackend()
+
+ assert backend.get_variable("_teama___api_key") is None
+ mock_client.return_value.list_namespaced_secret.assert_not_called()
+
class TestKubernetesSecretsBackendConfig:
@mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default")
@@ -150,7 +241,7 @@ class TestKubernetesSecretsBackendConfig:
assert result == "sqlite:///airflow.db"
mock_client.return_value.list_namespaced_secret.assert_called_once_with(
"default",
- label_selector="airflow.apache.org/config-key=sql_alchemy_conn",
+ label_selector="airflow.apache.org/config-key=sql_alchemy_conn,!airflow.apache.org/team",
resource_version="0",
)
@@ -181,7 +272,7 @@ class TestKubernetesSecretsBackendCustomConfig:
assert result == "postgresql://localhost/db"
mock_client.return_value.list_namespaced_secret.assert_called_once_with(
"default",
- label_selector="my-org/conn=my_db",
+ label_selector="my-org/conn=my_db,!airflow.apache.org/team",
resource_version="0",
)
@@ -225,6 +316,23 @@ class TestKubernetesSecretsBackendCustomConfig:
assert result is None
+ @mock.patch(f"{MODULE_PATH}.namespace", new_callable=mock.PropertyMock, return_value="default")
+ @mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock)
+ def test_team_specific_lookup_uses_custom_team_label(self, mock_client, mock_namespace):
+ mock_client.return_value.list_namespaced_secret.return_value = _make_secret_list(
+ [_make_secret({"value": "team-conn"})]
+ )
+
+ backend = KubernetesSecretsBackend(team_label="my-org.io/team")
+ result = backend.get_conn_value("my_db", team_name="team_a")
+
+ assert result == "team-conn"
+ mock_client.return_value.list_namespaced_secret.assert_called_once_with(
+ "default",
+ label_selector="airflow.apache.org/connection-id=my_db,my-org.io/team=team_a",
+ resource_version="0",
+ )
+
class TestKubernetesSecretsBackendLabelNone:
@mock.patch(f"{MODULE_PATH}.client", new_callable=mock.PropertyMock)
@@ -332,7 +440,7 @@ class TestKubernetesSecretsBackendNamespace:
mock_client.return_value.list_namespaced_secret.assert_called_once_with(
"airflow",
- label_selector="airflow.apache.org/connection-id=my_db",
+ label_selector="airflow.apache.org/connection-id=my_db,!airflow.apache.org/team",
resource_version="0",
)