Resolve team scoped spelling of per-key secrets-backend-kwarg options as sensitive

airflow-core/newsfragments/71160.bugfix.rst

@@ -0,0 +1 @@
+Redact team scoped spellings of the per-key secrets-backend-kwarg options (``[<team>=secrets]`` / ``[<team>=workers]`` config sections, and the corresponding ``AIRFLOW__<TEAM>___...`` environment variables) in ``GET /config`` and ``GET /config/section/{section}/option/{option}``, the same way the non-team-scoped spelling is already redacted.

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

@@ -21,6 +21,7 @@
 from fastapi import HTTPException, status
 from fastapi.responses import Response
 
+from airflow._shared.configuration.parser import base_section_name
 from airflow.api_fastapi.common.types import Mimetype
 from airflow.api_fastapi.core_api.datamodels.config import Config
 from airflow.configuration import conf
@@ -31,29 +32,44 @@
 # (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
+# A team scoped spelling of the same option -- the ``[<team>=secrets]`` config-file section, or
+# ``AIRFLOW__<TEAM>___SECRETS__BACKEND_KWARG__*``, which is reported under a section named after
+# the team -- is resolved back to its base section via ``base_section_name`` before matching, so
+# it is masked the same way. See https://github.com/apache/airflow/issues/71037
 _PER_KEY_SENSITIVE_PREFIXES: dict[str, str] = {
     "secrets": "backend_kwarg__",
     "workers": "secrets_backend_kwarg__",
 }
 
 
 def _is_per_key_sensitive_option(section: str, option: str) -> bool:
-    """Return True for synthetic per-key secrets-backend-kwarg options."""
-    prefix = _PER_KEY_SENSITIVE_PREFIXES.get(section)
+    """
+    Return True for synthetic per-key secrets-backend-kwarg options.
+
+    Resolves a team scoped section (e.g. ``myteam=secrets``) back to its base
+    section (``secrets``) before matching, so a team scoped spelling of the
+    same option is treated exactly like the base one.
+    """
+    prefix = _PER_KEY_SENSITIVE_PREFIXES.get(base_section_name(section))
     return prefix is not None and option.startswith(prefix)
 
 
 def _mask_per_key_sensitive_options(conf_dict: dict) -> None:
-    """Mask synthetic per-key secrets-backend-kwarg options in-place."""
-    for section, prefix in _PER_KEY_SENSITIVE_PREFIXES.items():
-        options = conf_dict.get(section)
+    """
+    Mask synthetic per-key secrets-backend-kwarg options in-place.
+
+    Iterates every section actually present in ``conf_dict`` -- rather than
+    only the literal ``secrets`` / ``workers`` keys -- and resolves each one
+    to its base section, so a team scoped section (config-file spelling) or a
+    team scoped env var (which ``conf.as_dict`` reports under a
+    team-derived section) is masked as well.
+    """
+    for section, options in conf_dict.items():
         if not options:
             continue
+        prefix = _PER_KEY_SENSITIVE_PREFIXES.get(base_section_name(section))
+        if prefix is None:
+            continue
         for option in list(options):
             if option.startswith(prefix):
                 current = options[option]

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

@@ -623,3 +623,73 @@ def test_get_config_value_keeps_team_scoped_non_sensitive_option(self, test_clie
         )
         assert response.status_code == 200
         assert response.json()["sections"][0]["options"][0]["value"] == OPTION_VALUE_TEAM_PARALLELISM
+
+
+SECTION_TEAM_SECRETS = f"{TEAM_NAME}={SECTION_SECRETS}"
+SECTION_TEAM_WORKERS = f"{TEAM_NAME}={SECTION_WORKERS}"
+
+
+class TestTeamScopedPerKeyBackendKwargMasking(TestConfigEndpoint):
+    """A team scoped spelling of a per-key secrets-backend-kwarg option -- the
+    ``[<team>=secrets]`` config-file section, or an
+    ``AIRFLOW__<TEAM>___SECRETS__BACKEND_KWARG__*`` env var -- is reported by
+    ``conf.as_dict`` under a section named after the team (``team_a=secrets``)
+    rather than under the literal ``secrets`` / ``workers`` section. Both
+    spellings collapse to that same shape once ``conf.as_dict`` has parsed
+    them, so a single team scoped section covers both. The Config API must
+    resolve that section back to its base section and redact the option the
+    same way it redacts the non-team-scoped one. See
+    https://github.com/apache/airflow/issues/71037"""
+
+    @pytest.fixture(autouse=True)
+    def setup_team_scoped_per_key(self) -> Generator[None, None, None]:
+        per_key_dict = {
+            SECTION_CORE: {OPTION_KEY_PARALLELISM: OPTION_VALUE_PARALLELISM},
+            SECTION_TEAM_SECRETS: {PER_KEY_OPTION_SECRETS: PER_KEY_VALUE},
+            SECTION_TEAM_WORKERS: {PER_KEY_OPTION_WORKERS: PER_KEY_VALUE},
+        }
+
+        def _mock_conf_as_dict(display_sensitive: bool, **_):
+            return {section: options.copy() for section, options in per_key_dict.items()}
+
+        def _mock_has_option(section: str, option: str) -> bool:
+            return option in per_key_dict.get(section, {})
+
+        with (
+            conf_vars(AIRFLOW_CONFIG_ENABLE_EXPOSE_CONFIG),
+            patch(
+                "airflow.api_fastapi.core_api.routes.public.config.conf.as_dict",
+                new=_mock_conf_as_dict,
+            ),
+            patch(
+                "airflow.api_fastapi.core_api.routes.public.config.conf.has_option",
+                new=_mock_has_option,
+            ),
+        ):
+            yield
+
+    def test_get_config_masks_team_scoped_per_key_secrets_backend_kwargs(self, test_client):
+        """``GET /config`` must redact a team scoped per-key option under
+        both the ``secrets`` and ``workers`` team scoped sections when
+        ``display_sensitive=False`` (the API-server default)."""
+        response = test_client.get("/config", headers=HEADERS_JSON)
+        assert response.status_code == 200
+
+        sections = {
+            s["name"]: {o["key"]: o["value"] for o in s["options"]} for s in response.json()["sections"]
+        }
+        assert sections[SECTION_TEAM_SECRETS][PER_KEY_OPTION_SECRETS] == OPTION_VALUE_SENSITIVE_HIDDEN
+        assert sections[SECTION_TEAM_WORKERS][PER_KEY_OPTION_WORKERS] == OPTION_VALUE_SENSITIVE_HIDDEN
+        # Non-sensitive option in the same response must remain untouched.
+        assert sections[SECTION_CORE][OPTION_KEY_PARALLELISM] == OPTION_VALUE_PARALLELISM
+
+    def test_get_config_value_masks_team_scoped_per_key_secrets_backend_kwarg(self, test_client):
+        """``GET /config/section/{section}/option/{option}`` must redact a
+        team scoped per-key synthetic option the same way it redacts the
+        non-team-scoped one."""
+        response = test_client.get(
+            f"/config/section/{SECTION_TEAM_SECRETS}/option/{PER_KEY_OPTION_SECRETS}",
+            headers=HEADERS_JSON,
+        )
+        assert response.status_code == 200
+        assert response.json()["sections"][0]["options"][0]["value"] == OPTION_VALUE_SENSITIVE_HIDDEN