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

providers/yandex/src/airflow/providers/yandex/secrets/lockbox.py

@@ -18,7 +18,6 @@
 
 from __future__ import annotations
 
-import re
 from functools import cached_property
 from typing import Any
 
@@ -162,7 +161,8 @@ def get_conn_value(self, conn_id: str, team_name: str | None = None) -> str | No
         if conn_id == self.yc_connection_id:
             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_value(self.connections_prefix, conn_id, team_name=team_name)
@@ -178,7 +178,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_value(self.variables_prefix, key, team_name=team_name)
@@ -256,13 +257,38 @@ def _build_team_secret_name(self, prefix: str, team_name: str, key: str) -> str:
         team_prefix = self._build_secret_name(prefix, team_name)
         return f"{team_prefix}{self.sep * TEAM_SEP_MULTIPLIER}{key}"
 
-    def _is_team_specific_accessed_as_global(self, secret_id: str, team_name: str | None = None) -> bool:
-        team_sep = re.escape(self.sep * TEAM_SEP_MULTIPLIER)
-        return team_name is None and bool(re.fullmatch(rf"[^{re.escape(self.sep)}]+{team_sep}.+", 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><key>``, so an id that itself contains
+        the team separator makes the built name ambiguous: team ``a`` with key ``b//c`` and team
+        ``a//b`` with key ``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.
+        """
+        return self.sep * TEAM_SEP_MULTIPLIER in secret_id
+
+    def _log_refusal(self, kind: str, secret_id: str) -> None:
+        self.log.warning(
+            "%s id %r contains %r, which separates the team name from the key in a team scoped "
+            "secret name. Such an id is ambiguous and is not looked up. Returning None.",
+            kind.capitalize(),
+            secret_id,
+            self.sep * TEAM_SEP_MULTIPLIER,
+        )
 
     def _get_secret_value(self, prefix: str, key: str, team_name: str | None = None) -> str | None:
         secrets = self._get_secrets()
         secret: secret_pb.Secret | None = None
+        # The team scoped name is tried first. Keys that would make it name a namespace other
+        # than the caller's own are refused by the callers before reaching here.
         if team_name:
             secret = self._find_secret(secrets, prefix, self._build_team_secret_name("", team_name, key))
 

providers/yandex/tests/unit/yandex/secrets/test_lockbox.py

@@ -319,6 +319,79 @@ def test_get_conn_value_returns_none_for_team_scoped_id_without_team_name(self,
         assert backend.get_conn_value("teama//my_db") is None
         mock_get_secrets.assert_not_called()
 
+    @patch("airflow.providers.yandex.secrets.lockbox.LockboxSecretBackend._get_secrets")
+    @patch("airflow.providers.yandex.secrets.lockbox.LockboxSecretBackend._get_payload")
+    def test_another_teams_secret_is_not_reachable(self, mock_get_payload, mock_get_secrets):
+        """A caller scoped to one team must not reach another team's secret by naming it.
+
+        The target secret exists and is resolvable, so a backend that falls through to the
+        team agnostic name returns its value. Only refusing that fall-through returns None.
+        """
+        mock_get_secrets.return_value = [
+            secret_pb.Secret(id="123", name="airflow/connections/teama//my_db"),
+        ]
+        mock_get_payload.return_value = payload_pb.Payload(
+            entries=[payload_pb.Payload.Entry(text_value="teama-conn")]
+        )
+
+        backend = LockboxSecretBackend()
+
+        assert backend.get_conn_value("teama//my_db", team_name="teamb") is None
+        mock_get_payload.assert_not_called()
+
+    @patch("airflow.providers.yandex.secrets.lockbox.LockboxSecretBackend._get_secrets")
+    @patch("airflow.providers.yandex.secrets.lockbox.LockboxSecretBackend._get_payload")
+    def test_team_whose_name_extends_the_callers_is_not_reachable(self, mock_get_payload, mock_get_secrets):
+        """A prefix match on the caller's own namespace is not proof of ownership."""
+        mock_get_secrets.return_value = [
+            secret_pb.Secret(id="123", name="airflow/connections/teama//prod//my_db"),
+        ]
+        mock_get_payload.return_value = payload_pb.Payload(
+            entries=[payload_pb.Payload.Entry(text_value="teama-prod-conn")]
+        )
+
+        backend = LockboxSecretBackend()
+
+        assert backend.get_conn_value("teama//prod//my_db", team_name="teama") is None
+        mock_get_payload.assert_not_called()
+
+    @patch("airflow.providers.yandex.secrets.lockbox.LockboxSecretBackend._get_secrets")
+    @patch("airflow.providers.yandex.secrets.lockbox.LockboxSecretBackend._get_payload")
+    def test_team_scoped_lookup_cannot_reach_a_longer_teams_namespace(
+        self, mock_get_payload, mock_get_secrets
+    ):
+        """The team scoped name is not safe by construction -- the key 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 lookup *finds* another team's secret. Refusing
+        only the team agnostic fall-through leaves this open, because the fall-through is never
+        reached.
+        """
+        mock_get_secrets.return_value = [
+            secret_pb.Secret(id="123", name="airflow/connections/teama//prod//my_db"),
+        ]
+        mock_get_payload.return_value = payload_pb.Payload(
+            entries=[payload_pb.Payload.Entry(text_value="teama-prod-conn")]
+        )
+
+        backend = LockboxSecretBackend()
+
+        assert backend.get_conn_value("prod//my_db", team_name="teama") is None
+        mock_get_payload.assert_not_called()
+
+    @patch("airflow.providers.yandex.secrets.lockbox.LockboxSecretBackend._get_secrets")
+    def test_refusing_an_ambiguous_id_is_logged(self, mock_get_secrets, caplog):
+        """A silent ``None`` is indistinguishable from a missing secret, so the refusal is logged."""
+        backend = LockboxSecretBackend()
+
+        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_get_secrets.assert_not_called()
+
     @patch("airflow.providers.yandex.secrets.lockbox.LockboxSecretBackend._get_secrets")
     @patch("airflow.providers.yandex.secrets.lockbox.LockboxSecretBackend._get_payload")
     def test_yandex_lockbox_secret_backend__get_secret_value(self, mock_get_payload, mock_get_secrets):