Refuse the team agnostic fall-through for a team scoped Key Vault secret name

providers/microsoft/azure/src/airflow/providers/microsoft/azure/secrets/key_vault.py

@@ -157,7 +157,8 @@ def get_conn_value(self, conn_id: str, team_name: str | None = None) -> str | No
         if self.connections_prefix is None:
             return None
 
-        if self._is_team_specific_accessed_as_global(conn_id, team_name):
+        if self._names_a_team_namespace(conn_id):
+            self._log_refusal("connection", conn_id)
             return None
 
         return self._get_secret(self.connections_prefix, conn_id, team_name=team_name)
@@ -173,7 +174,8 @@ def get_variable(self, key: str, team_name: str | None = None) -> str | None:
         if self.variables_prefix is None:
             return None
 
-        if self._is_team_specific_accessed_as_global(key, team_name):
+        if self._names_a_team_namespace(key):
+            self._log_refusal("variable", key)
             return None
 
         return self._get_secret(self.variables_prefix, key, team_name=team_name)
@@ -210,22 +212,59 @@ def build_path(path_prefix: str, secret_id: str, sep: str = "-") -> str:
         return path.replace("_", sep)
 
     def _build_team_secret_name(self, path_prefix: str, team_name: str, secret_id: str) -> str:
-        """Build a team-scoped secret name using a dedicated separator before the secret id."""
+        """
+        Build a team-scoped secret name using a dedicated separator before the secret id.
+
+        The secret id is normalised the same way :meth:`build_path` normalises every other name
+        in this backend. On its own that would let ``b__c`` manufacture the team separator; ids
+        whose normalised form contains it are refused by the callers before they get here.
+        """
         team_prefix = self.build_path(path_prefix, team_name, self.sep)
         normalized_secret_id = secret_id.replace("_", self.sep)
         return f"{team_prefix}{TEAM_SEP}{normalized_secret_id}"
 
-    def _is_team_specific_accessed_as_global(self, secret_id: str, team_name: str | None = None) -> bool:
-        normalized_secret_id = self.build_path("", secret_id, self.sep)
-        return team_name is None and bool(re.fullmatch(rf".+{re.escape(TEAM_SEP)}.+", normalized_secret_id))
+    def _names_a_team_namespace(self, secret_id: str) -> bool:
+        """
+        Whether ``secret_id`` spells out a team scoped secret name.
+
+        A team scoped secret is named ``<team>{TEAM_SEP}<secret id>``, so an id that itself
+        contains the team separator makes the built name ambiguous: team ``a`` with id ``b--c``
+        and team ``a--b`` with id ``c`` produce the same string. Such an id is refused for
+        *every* lookup -- team scoped as well as team agnostic -- because the ambiguity exists
+        in both directions and the caller's own namespace is not a safe harbour for it.
+
+        The id is never parsed to work out *which* team it names, because it cannot be: nothing
+        in the string distinguishes the two readings above. Comparing the id against the prefix
+        the caller's own team builds looks equivalent and is not -- a caller in team ``a`` would
+        match ``a--b``'s namespace on the prefix and read its secrets. Only the caller's own
+        namespace is ever constructed, never parsed.
+
+        The id is normalised first because :meth:`build_path` maps ``_`` onto the separator
+        everywhere in this backend, so ``b__c`` reaches Key Vault as ``b--c`` and would
+        otherwise manufacture the team separator from an id that does not visibly contain it.
+        """
+        return TEAM_SEP in self.build_path("", secret_id, self.sep)
+
+    def _log_refusal(self, kind: str, secret_id: str) -> None:
+        self.log.warning(
+            "%s id %r resolves to a name containing %r, which separates the team name from the "
+            "secret id in a team scoped secret name. Such an id is ambiguous and is not looked "
+            "up. Returning None.",
+            kind.capitalize(),
+            secret_id,
+            TEAM_SEP,
+        )
 
     def _get_secret(self, path_prefix: str, secret_id: str, team_name: str | None = None) -> str | None:
         """
         Get an Azure Key Vault secret value.
 
         :param path_prefix: Prefix for the Path to get Secret
         :param secret_id: Secret Key
+        :param team_name: Team the lookup is scoped to (if any)
         """
+        # The team scoped name is tried first. Ids that would make it name a namespace other
+        # than the caller's own are refused by the callers before reaching here.
         if team_name:
             team_secret = self._get_secret_value(
                 path_prefix, self._build_team_secret_name("", team_name, secret_id)

providers/microsoft/azure/tests/unit/microsoft/azure/secrets/test_key_vault.py

@@ -113,6 +113,98 @@ def test_get_variable_returns_none_for_team_scoped_key_without_team_name(self, m
         assert backend.get_variable("teama--hello") is None
         mock_client.get_secret.assert_not_called()
 
+    @mock.patch(f"{KEY_VAULT_MODULE}.AzureKeyVaultBackend.client")
+    def test_another_teams_secret_is_not_reachable(self, mock_client):
+        """A caller scoped to one team must not reach another team's secret by naming it.
+
+        The target secret is resolvable, so a backend that falls through to the team agnostic
+        name returns it. Only a backend that refuses that fall-through returns None.
+        """
+
+        def only_the_target_exists(name):
+            if name == "airflow-connections-teama--my-db":
+                return mock.Mock(value="teama-secret")
+            raise ResourceNotFoundError
+
+        mock_client.get_secret.side_effect = only_the_target_exists
+
+        backend = AzureKeyVaultBackend()
+
+        assert backend.get_conn_value("teama--my_db", team_name="teamb") is None
+
+    @mock.patch(f"{KEY_VAULT_MODULE}.AzureKeyVaultBackend.client")
+    def test_team_whose_name_extends_the_callers_is_not_reachable(self, mock_client):
+        """A prefix match on the caller's own namespace is not proof of ownership.
+
+        Team names may contain the separator, so ``teama--prod`` is a distinct team whose
+        namespace starts with ``teama``'s. Treating that prefix as ownership would hand one
+        team the secrets of every team whose name extends it.
+        """
+
+        def only_the_target_exists(name):
+            if name == "airflow-connections-teama--prod--my-db":
+                return mock.Mock(value="teama-prod-secret")
+            raise ResourceNotFoundError
+
+        mock_client.get_secret.side_effect = only_the_target_exists
+
+        backend = AzureKeyVaultBackend()
+
+        assert backend.get_conn_value("teama--prod--my_db", team_name="teama") is None
+
+    @mock.patch(f"{KEY_VAULT_MODULE}.AzureKeyVaultBackend.client")
+    def test_team_scoped_lookup_cannot_reach_a_longer_teams_namespace(self, mock_client):
+        """The team scoped name is not safe by construction -- the id can extend it.
+
+        Team ``teama`` asking for ``prod--my_db`` builds exactly the name team ``teama--prod``
+        builds for ``my_db``, so the team scoped probe *hits* another team's secret. Refusing
+        only the team agnostic fall-through leaves this open, because the fall-through is
+        never reached.
+        """
+
+        def only_the_target_exists(name):
+            if name == "airflow-connections-teama--prod--my-db":
+                return mock.Mock(value="teama-prod-secret")
+            raise ResourceNotFoundError
+
+        mock_client.get_secret.side_effect = only_the_target_exists
+
+        backend = AzureKeyVaultBackend()
+
+        assert backend.get_conn_value("prod--my_db", team_name="teama") is None
+
+    @mock.patch(f"{KEY_VAULT_MODULE}.AzureKeyVaultBackend.client")
+    def test_underscores_cannot_manufacture_a_team_namespace(self, mock_client):
+        """``build_path`` maps ``_`` onto the separator, so ``__`` becomes the team separator.
+
+        An id that contains no separator as written still names another team's namespace once
+        normalised, so the guard has to run on the normalised form.
+        """
+
+        def only_the_target_exists(name):
+            if name == "airflow-connections-teama--prod--my-db":
+                return mock.Mock(value="teama-prod-secret")
+            raise ResourceNotFoundError
+
+        mock_client.get_secret.side_effect = only_the_target_exists
+
+        backend = AzureKeyVaultBackend()
+
+        assert backend.get_conn_value("prod__my_db", team_name="teama") is None
+
+    @mock.patch(f"{KEY_VAULT_MODULE}.AzureKeyVaultBackend.client")
+    def test_refusing_an_ambiguous_id_is_logged(self, mock_client, caplog):
+        """A silent ``None`` is indistinguishable from a missing secret, so the refusal is logged."""
+        backend = AzureKeyVaultBackend()
+
+        assert backend.get_conn_value("prod--my_db") is None
+        assert backend.get_variable("prod__hello") is None
+
+        refusals = [r for r in caplog.records if "is ambiguous and is not looked up" in r.getMessage()]
+        assert len(refusals) == 2
+        assert all(r.levelname == "WARNING" for r in refusals)
+        mock_client.get_secret.assert_not_called()
+
     @mock.patch(f"{KEY_VAULT_MODULE}.AzureKeyVaultBackend._get_secret")
     def test_variable_prefix_none_value(self, mock_get_secret):
         """