Only refuse team-scoped-looking env var ids when multi_team is on

airflow-core/src/airflow/secrets/environment_variables.py

@@ -21,6 +21,7 @@
 
 import os
 
+from airflow.configuration import conf
 from airflow.secrets import BaseSecretsBackend
 
 CONN_ENV_PREFIX = "AIRFLOW_CONN_"
@@ -34,10 +35,20 @@
 class EnvironmentVariablesBackend(BaseSecretsBackend):
     """Retrieves Connection object and Variable from environment variable."""
 
+    @staticmethod
+    def _names_a_team_namespace(secret_id: str) -> bool:
+        """
+        Whether ``secret_id`` spells out a team scoped secret name.
+
+        Only checked in multi-team mode: ``team_name`` is never non-``None`` otherwise, so no
+        team scoped variable can exist to collide with.
+        """
+        if not conf.getboolean("core", "multi_team", fallback=False):
+            return False
+        return TEAM_SEP in secret_id
+
     def get_conn_value(self, conn_id: str, team_name: str | None = None) -> str | None:
-        if TEAM_SEP in conn_id:
-            # An id containing the separator could collide with another team's namespace
-            # even on the scoped lookup below, so it must be refused before either runs.
+        if self._names_a_team_namespace(conn_id):
             return None
 
         if team_name and (
@@ -56,8 +67,7 @@ def get_variable(self, key: str, team_name: str | None = None) -> str | None:
         :param team_name: Team name associated to the task trying to access the variable (if any)
         :return: Variable Value
         """
-        if TEAM_SEP in key:
-            # Same collision risk as get_conn_value, see its code comment.
+        if self._names_a_team_namespace(key):
             return None
 
         if team_name and (

airflow-core/tests/unit/always/test_secrets.py

@@ -122,6 +122,7 @@ def test_backend_fallback_to_env_var(self, mock_get_connection):
         assert conn.get_uri() == "mysql://airflow:airflow@host:5432/airflow"
 
     @pytest.mark.db_test
+    @conf_vars({("core", "multi_team"): "True"})
     @mock.patch.dict(
         "os.environ",
         {
@@ -218,6 +219,7 @@ def test_backend_variable_order(self, mock_secret_get, mock_meta_get):
         mock_secret_get.return_value = "a_secret_value"
         assert Variable.get(key="not_myvar") == "a_secret_value"
 
+    @conf_vars({("core", "multi_team"): "True"})
     @mock.patch.dict(
         "os.environ",
         {

airflow-core/tests/unit/always/test_secrets_environment_variables.py

@@ -26,6 +26,8 @@
     EnvironmentVariablesBackend,
 )
 
+from tests_common.test_utils.config import conf_vars
+
 # A team specific secret is stored as ``<PREFIX>_<TEAM_NAME>___<SECRET_ID>``. Team names may contain
 # underscores (they are validated against ``^[a-zA-Z0-9_-]{3,50}$``), so both shapes are exercised.
 TEAM_NAMES = ["team_a", "teama"]
@@ -57,6 +59,11 @@ def lookup(env_prefix: str, method: str, secret_id: str, team_name: str | None)
 class TestEnvironmentVariablesBackendTeamScope:
     """A team specific secret must only be resolvable for the team it is stored for."""
 
+    @pytest.fixture(autouse=True)
+    def _multi_team_enabled(self):
+        with conf_vars({("core", "multi_team"): "True"}):
+            yield
+
     @pytest.mark.parametrize(("env_prefix", "method"), LOOKUPS)
     @pytest.mark.parametrize("team_name", TEAM_NAMES)
     def test_team_scoped_secret_is_not_resolved_without_a_team_scope(
@@ -186,3 +193,15 @@ def test_unset_secret_is_not_resolved(self, monkeypatch, env_prefix, method, tea
         monkeypatch.delenv(env_prefix + SECRET_ID.upper(), raising=False)
 
         assert lookup(env_prefix, method, SECRET_ID, team_name) is None
+
+
+class TestEnvironmentVariablesBackendMultiTeamDisabled:
+    """No team scoped variable can exist without multi-team mode, so there is no ambiguity
+    to refuse -- an ordinary id containing the separator must resolve normally."""
+
+    @pytest.mark.parametrize(("env_prefix", "method"), LOOKUPS)
+    def test_ambiguous_id_resolves_when_multi_team_is_disabled(self, monkeypatch, env_prefix, method):
+        secret_id = f"prod{TEAM_SEP}{SECRET_ID}"
+        monkeypatch.setenv(env_prefix + secret_id.upper(), GLOBAL_VALUE)
+
+        assert lookup(env_prefix, method, secret_id, None) == GLOBAL_VALUE