Hide team scoped values of options registered as sensitive

airflow-core/newsfragments/70755.significant.rst

@@ -0,0 +1,29 @@
+Team scoped values of options registered as sensitive are now hidden
+
+Configuration options are registered as sensitive under their base section, so until now only the
+base spelling of an option was masked. A team scoped override -- set in a ``[<team>=<section>]``
+config file section, or through an ``AIRFLOW__<TEAM>___<SECTION>__<KEY>`` environment variable --
+was not recognised as the same option and was returned in full.
+
+Sensitivity is now decided after resolving the team scoped spelling back to the base option, so a
+team scoped value is masked exactly as the base value already was.
+
+**Behaviour changes:**
+
+- ``AirflowConfigParser.as_dict(display_sensitive=False)``, ``GET /config``,
+  ``GET /config/section/{section}/option/{option}`` and ``airflow config list`` now return
+  ``< hidden >`` for a team scoped value of an option registered as sensitive. Deployments that
+  read a team's real value through any of these will now receive the mask; use
+  ``display_sensitive=True`` where a real value is required and appropriate.
+- Team scoped ``_cmd`` and ``_secret`` entries are replaced with ``< hidden >`` in place, rather
+  than being resolved into their value and removed as they are in a base section. Resolving them
+  is not supported for a team, so the command string or secret path is no longer shown either.
+- Non team configuration is unaffected, and ``display_sensitive=True`` continues to return real
+  values.
+
+**New public helpers on the shared configuration parser** (additive; no signatures changed):
+
+- ``team_section_name(team_name, section)`` builds the team scoped config file section name.
+- ``base_section_name(section)`` recovers the base section from a possibly team scoped one.
+- ``AirflowConfigParser.is_sensitive_option(section, key)`` reports whether an option is
+  registered as sensitive under any of its spellings.

airflow-core/src/airflow/api_fastapi/core_api/routes/public/config.py

@@ -153,9 +153,7 @@ def get_config_value(
         )
 
     section_l, option_l = section.lower(), option.lower()
-    if (section_l, option_l) in conf.sensitive_config_values or _is_per_key_sensitive_option(
-        section_l, option_l
-    ):
+    if conf.is_sensitive_option(section_l, option_l) or _is_per_key_sensitive_option(section_l, option_l):
         value = "< hidden >"
     else:
         value = conf.get(section, option)

airflow-core/src/airflow/api_fastapi/core_api/services/public/config.py

@@ -31,6 +31,11 @@
 # (e.g. Vault role_id / secret_id) as the registered ``backend_kwargs``
 # option, so they need the same redaction treatment when
 # ``display_sensitive=False``.
+# These match literal section names only, so a team scoped spelling of the same option -- the
+# ``[<team>=secrets]`` section, or ``AIRFLOW__<TEAM>___SECRETS__BACKEND_KWARG__*``, which is
+# reported under a section named after the team -- is not recognised. Nothing leaks today because
+# the secrets backend is not team aware; tracked at
+# https://github.com/apache/airflow/issues/71037
 _PER_KEY_SENSITIVE_PREFIXES: dict[str, str] = {
     "secrets": "backend_kwarg__",
     "workers": "secrets_backend_kwarg__",

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_config.py

@@ -22,6 +22,8 @@
 
 import pytest
 
+from airflow.configuration import conf
+
 from tests_common.test_utils.config import conf_vars
 
 pytestmark = pytest.mark.db_test
@@ -559,3 +561,65 @@ def test_get_config_value_masks_per_key_secrets_backend_kwarg(self, test_client)
         )
         assert response.status_code == 200
         assert response.json()["sections"][0]["options"][0]["value"] == OPTION_VALUE_SENSITIVE_HIDDEN
+
+
+TEAM_NAME = "team_a"
+SECTION_TEAM_DATABASE = f"{TEAM_NAME}={SECTION_DATABASE}"
+SECTION_TEAM_CORE = f"{TEAM_NAME}={SECTION_CORE}"
+OPTION_VALUE_TEAM_SQL_ALCHEMY_CONN = "postgresql://team_a_user:team_a_password@team-a-db/airflow"
+OPTION_VALUE_TEAM_PARALLELISM = "512"
+
+
+class TestTeamScopedOptionMasking(TestConfigEndpoint):
+    """A team scoped override of a sensitive option lives in a ``[<team>=<section>]`` section rather
+    than in the section the option is registered under. The single-option endpoint has to resolve
+    that spelling back to the registered option, so that it redacts a team scoped value exactly as
+    it redacts the base one."""
+
+    @pytest.fixture(autouse=True)
+    def setup_team_scoped(self) -> Generator[None, None, None]:
+        team_config = {
+            SECTION_TEAM_DATABASE: {OPTION_KEY_SQL_ALCHEMY_CONN: OPTION_VALUE_TEAM_SQL_ALCHEMY_CONN},
+            SECTION_TEAM_CORE: {OPTION_KEY_PARALLELISM: OPTION_VALUE_TEAM_PARALLELISM},
+        }
+
+        # Anything that is not one of the team scoped options under test keeps being read from the
+        # real configuration - the endpoint reads [api] expose_config through the same object.
+        real_get = conf.get
+
+        def _mock_has_option(section: str, option: str, **_) -> bool:
+            return option in team_config.get(section, {})
+
+        def _mock_get(section: str, option: str, *args, **kwargs) -> str:
+            if option in team_config.get(section, {}):
+                return team_config[section][option]
+            return real_get(section, option, *args, **kwargs)
+
+        with (
+            conf_vars(AIRFLOW_CONFIG_ENABLE_EXPOSE_CONFIG),
+            patch(
+                "airflow.api_fastapi.core_api.routes.public.config.conf.has_option",
+                new=_mock_has_option,
+            ),
+            patch(
+                "airflow.api_fastapi.core_api.routes.public.config.conf.get",
+                new=_mock_get,
+            ),
+        ):
+            yield
+
+    def test_get_config_value_masks_team_scoped_sensitive_option(self, test_client):
+        response = test_client.get(
+            f"/config/section/{SECTION_TEAM_DATABASE}/option/{OPTION_KEY_SQL_ALCHEMY_CONN}",
+            headers=HEADERS_JSON,
+        )
+        assert response.status_code == 200
+        assert response.json()["sections"][0]["options"][0]["value"] == OPTION_VALUE_SENSITIVE_HIDDEN
+
+    def test_get_config_value_keeps_team_scoped_non_sensitive_option(self, test_client):
+        response = test_client.get(
+            f"/config/section/{SECTION_TEAM_CORE}/option/{OPTION_KEY_PARALLELISM}",
+            headers=HEADERS_JSON,
+        )
+        assert response.status_code == 200
+        assert response.json()["sections"][0]["options"][0]["value"] == OPTION_VALUE_TEAM_PARALLELISM

shared/configuration/src/airflow_shared/configuration/parser.py

@@ -77,6 +77,34 @@ def _collect_kwarg_env_vars(prefix: str) -> dict[str, str]:
 ConfigSectionSourcesType = dict[str, str | tuple[str, str]]
 ConfigSourcesType = dict[str, ConfigSectionSourcesType]
 ENV_VAR_PREFIX = "AIRFLOW__"
+# Separates the team name from the base section name in a team scoped config file section.
+TEAM_SECTION_SEPARATOR = "="
+
+
+def team_section_name(team_name: str, section: str) -> str:
+    """
+    Build the config file section name that holds the team scoped overrides of ``section``.
+
+    :param team_name: name of the team the overrides belong to
+    :param section: base section name that is being overridden
+    :return: the team scoped section name, e.g. ``team_a=celery``
+    """
+    return f"{team_name}{TEAM_SECTION_SEPARATOR}{section}"
+
+
+def base_section_name(section: str) -> str:
+    """
+    Return the base section name of a possibly team scoped config file section.
+
+    Team scoped sections are built by :func:`team_section_name`. Base section names never contain
+    the separator, so the name is split on the last one - that way the base section is recovered
+    even for a team name that contains the separator itself.
+
+    :param section: section name, either a base one or a team scoped one
+    :return: the base section name, which is ``section`` itself when it is not team scoped
+    """
+    _, separator, base_section = section.rpartition(TEAM_SECTION_SEPARATOR)
+    return base_section if separator else section
 
 
 if TYPE_CHECKING:
@@ -562,6 +590,68 @@ def sensitive_config_values(self) -> set[tuple[str, str]]:
         sensitive.update(depr_section, depr_option)
         return sensitive
 
+    def _names_sensitive_team_env_var(self, env_var: str) -> bool:
+        """
+        Check whether an environment variable name is a team scoped override of a sensitive option.
+
+        Team scoped variables are named ``AIRFLOW__<TEAM>___<SECTION>__<KEY>`` (see
+        :meth:`_env_var_name`). A team name may contain underscores itself, so the team name is not
+        parsed out of the variable name; the name is matched against the tail that each option
+        registered as sensitive contributes instead. The ``_CMD`` / ``_SECRET`` fallbacks are
+        matched too, because - unlike for a base section - they are never resolved into their value.
+
+        :param env_var: environment variable name
+        :return: True if the variable holds a team scoped value of an option registered as sensitive
+        """
+        env_var = env_var.upper()
+        if not env_var.startswith(ENV_VAR_PREFIX):
+            return False
+        # Every tail matched below starts with the ``___`` separating the team name from the
+        # section, so a name not containing it cannot be a team scoped one.
+        if "___" not in env_var:
+            return False
+        for section, key in self.sensitive_config_values:
+            option_tail = self._env_var_name(section, key).removeprefix(ENV_VAR_PREFIX)
+            for tail in (f"___{option_tail}", f"___{option_tail}_CMD", f"___{option_tail}_SECRET"):
+                # The team name sits between the prefix and the tail, so it must not be empty.
+                if env_var.endswith(tail) and len(env_var) > len(ENV_VAR_PREFIX) + len(tail):
+                    return True
+        return False
+
+    def is_sensitive_option(self, section: str, key: str) -> bool:
+        """
+        Check whether the value of ``key`` in ``section`` is registered as sensitive.
+
+        Options are registered as sensitive under their base section name, while a team scoped
+        override of the very same option is held by a ``<team name>=<section>`` config file section
+        or by an ``AIRFLOW__<TEAM>___<SECTION>__<KEY>`` environment variable. Both of those
+        spellings are resolved back to the base option here, so that a team scoped value is treated
+        exactly like the base one. A name that does not resolve to a registered option is not
+        sensitive - so this only ever recognises more options as sensitive, never fewer.
+
+        :param section: section name, either a base one or a team scoped one
+        :param key: option name
+        :return: True if the value of the option should be treated as sensitive
+        """
+        section = section.lower()
+        key = key.lower()
+        if (section, key) in self.sensitive_config_values:
+            return True
+        base_section = base_section_name(section)
+        if base_section != section:
+            if (base_section, key) in self.sensitive_config_values:
+                return True
+            # A team scoped ``_cmd`` / ``_secret`` fallback is not resolved into its value, so it
+            # stays in the output as configured and has to be recognised on its own.
+            for fallback_suffix in ("_cmd", "_secret"):
+                if not key.endswith(fallback_suffix):
+                    continue
+                if (base_section, key.removesuffix(fallback_suffix)) in self.sensitive_config_values:
+                    return True
+        # A team scoped environment variable is reported under the section and key its name splits
+        # into, which is neither the base nor the team scoped section name.
+        return self._names_sensitive_team_env_var(self._env_var_name(section, key))
+
     def _update_defaults_from_string(self, config_string: str) -> None:
         """
         Update the defaults in _default_values based on values in config_string ("ini" format).
@@ -788,8 +878,11 @@ def _include_envs(
                 log.warning("Ignoring unknown env var '%s'", env_var)
                 continue
             if not display_sensitive and env_var != self._env_var_name("core", "unit_test_mode"):
+                if self._names_sensitive_team_env_var(env_var):
+                    # Covers the cmd/secret variants too; see is_sensitive_option.
+                    opt = "< hidden >"
                 # Don't hide cmd/secret values here
-                if not env_var.lower().endswith(("cmd", "secret")):
+                elif not env_var.lower().endswith(("cmd", "secret")):
                     if (section, key) in self.sensitive_config_values:
                         opt = "< hidden >"
             elif raw:
@@ -1121,7 +1214,7 @@ def _get_option_from_config_file(
     ) -> str | ValueNotFound:
         """Get config option from config file."""
         if team_name := kwargs.get("team_name", None):
-            section = f"{team_name}={section}"
+            section = team_section_name(team_name, section)
             # since this is the last lookup that supports team_name, pop it
             kwargs.pop("team_name")
         if super().has_option(section, key):
@@ -1792,15 +1885,19 @@ def as_dict(
         if not display_sensitive:
             # This ensures the ones from config file is hidden too
             # if they are not provided through env, cmd and secret
+            # The collected options are walked (rather than the registered sensitive ones) so that
+            # team scoped sections are covered as well - they are named after the team, not after
+            # the base section the option is registered under.
             hidden = "< hidden >"
-            for section, key in self.sensitive_config_values:
-                if config_sources.get(section):
-                    if config_sources[section].get(key, None):
-                        if display_source:
-                            source = config_sources[section][key][1]
-                            config_sources[section][key] = (hidden, source)
-                        else:
-                            config_sources[section][key] = hidden
+            for section, options in config_sources.items():
+                for key, value in list(options.items()):
+                    if not value or not self.is_sensitive_option(section, key):
+                        continue
+                    if display_source:
+                        source = value[1]
+                        options[key] = (hidden, source)
+                    else:
+                        options[key] = hidden
 
         return config_sources
 
@@ -1877,7 +1974,7 @@ def getsection(self, section: str, team_name: str | None = None) -> ConfigOption
         :param team_name: optional team name for team-specific configuration lookup
         """
         # Handle team-specific section lookup for config file
-        config_section = f"{team_name}={section}" if team_name else section
+        config_section = team_section_name(team_name, section) if team_name else section
 
         if not self.has_section(config_section) and not self._has_section_in_any_defaults(config_section):
             return None
@@ -2032,10 +2129,7 @@ def write(  # type: ignore[override]
                             section_to_write=section_to_write,
                             sources_dict=sources_dict,
                         )
-                        is_sensitive = (
-                            section_to_write.lower(),
-                            option.lower(),
-                        ) in self.sensitive_config_values
+                        is_sensitive = self.is_sensitive_option(section_to_write, option)
                         self._write_value(
                             file=file,
                             option=option,