Only refuse team scoped like secret ids when multi_team is on

providers/amazon/src/airflow/providers/amazon/aws/secrets/secrets_manager.py

@@ -25,6 +25,7 @@
 from typing import Any
 
 from airflow.providers.amazon.aws.utils import trim_none_values
+from airflow.providers.common.compat.sdk import conf
 from airflow.secrets import BaseSecretsBackend
 from airflow.utils.log.logging_mixin import LoggingMixin
 
@@ -335,7 +336,12 @@ def _names_a_team_namespace(secret_id: str) -> bool:
         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.
+
+        Only checked in multi-team mode: ``team_name`` is never non-``None`` otherwise, so no
+        team scoped secret can exist to collide with.
         """
+        if not conf.getboolean("core", "multi_team", fallback=False):
+            return False
         return TEAM_SEP in secret_id
 
     def _log_refusal(self, kind: str, secret_id: str) -> None:

providers/amazon/src/airflow/providers/amazon/aws/secrets/systems_manager.py

@@ -23,6 +23,7 @@
 from functools import cached_property
 
 from airflow.providers.amazon.aws.utils import trim_none_values
+from airflow.providers.common.compat.sdk import conf
 from airflow.secrets import BaseSecretsBackend
 from airflow.utils.log.logging_mixin import LoggingMixin
 
@@ -214,7 +215,12 @@ def _names_a_team_namespace(secret_id: str) -> bool:
         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.
+
+        Only checked in multi-team mode: ``team_name`` is never non-``None`` otherwise, so no
+        team scoped secret can exist to collide with.
         """
+        if not conf.getboolean("core", "multi_team", fallback=False):
+            return False
         return TEAM_SEP in secret_id
 
     def _log_refusal(self, kind: str, secret_id: str) -> None:

providers/amazon/tests/unit/amazon/aws/secrets/test_secrets_manager.py

@@ -24,6 +24,10 @@
 
 from airflow.providers.amazon.aws.secrets.secrets_manager import TEAM_SEP, SecretsManagerBackend
 
+from tests_common.test_utils.config import conf_vars
+
+multi_team_enabled = conf_vars({("core", "multi_team"): "True"})
+
 
 class TestSecretsManagerBackend:
     @mock.patch("airflow.providers.amazon.aws.secrets.secrets_manager.SecretsManagerBackend.get_conn_value")
@@ -80,6 +84,7 @@ def test_get_conn_value_with_team_name(self):
         returned_uri = secrets_manager_backend.get_conn_value(conn_id="test_postgres", team_name="my_team")
         assert returned_uri == "postgresql://airflow:airflow@host:5432/airflow"
 
+    @multi_team_enabled
     @mock_aws
     def test_global_caller_cannot_access_team_scoped_connection(self):
         secret_id = "airflow/connections/my_team--test_postgres"
@@ -93,6 +98,7 @@ def test_global_caller_cannot_access_team_scoped_connection(self):
 
         assert secrets_manager_backend.get_conn_value(conn_id="my_team--test_postgres") is None
 
+    @multi_team_enabled
     @mock_aws
     def test_another_teams_secret_is_not_reachable(self):
         """A caller scoped to one team must not reach another team's secret by naming it."""
@@ -103,6 +109,7 @@ def test_another_teams_secret_is_not_reachable(self):
 
         assert backend.get_conn_value(conn_id="my_team--test_postgres", team_name="other_team") is None
 
+    @multi_team_enabled
     @mock_aws
     def test_team_whose_name_extends_the_callers_is_not_reachable(self):
         """A prefix match on the caller's own namespace is not proof of ownership."""
@@ -113,6 +120,7 @@ def test_team_whose_name_extends_the_callers_is_not_reachable(self):
 
         assert backend.get_conn_value(conn_id="my_team--prod--test_postgres", team_name="my_team") is None
 
+    @multi_team_enabled
     @mock_aws
     def test_team_scoped_lookup_cannot_reach_a_longer_teams_namespace(self):
         """The team scoped name is not safe by construction -- the id can extend it.
@@ -129,6 +137,7 @@ def test_team_scoped_lookup_cannot_reach_a_longer_teams_namespace(self):
 
         assert backend.get_conn_value(conn_id="prod--test_postgres", team_name="my_team") is None
 
+    @multi_team_enabled
     @mock_aws
     def test_refusing_an_ambiguous_id_is_logged(self, caplog):
         """A silent ``None`` is indistinguishable from a missing secret, so the refusal is logged.
@@ -158,6 +167,22 @@ def test_refusing_an_ambiguous_id_is_logged(self, caplog):
             assert sum(refused_id in r.getMessage() for r in refusals) == 1
         assert all(TEAM_SEP in r.getMessage() for r in refusals)
 
+    @mock_aws
+    def test_ambiguous_id_resolves_when_multi_team_is_disabled(self):
+        """No team scoped secret can exist without multi-team mode, so there is no ambiguity
+        to refuse -- an ordinary id containing the separator must resolve normally."""
+        secret_id = "airflow/connections/prod--test_postgres"
+        create_param = {
+            "Name": secret_id,
+            "SecretString": "postgresql://airflow:airflow@host:5432/airflow",
+        }
+        backend = SecretsManagerBackend()
+        backend.client.create_secret(**create_param)
+
+        assert backend.get_conn_value(conn_id="prod--test_postgres") == (
+            "postgresql://airflow:airflow@host:5432/airflow"
+        )
+
     @mock_aws
     def test_team_caller_falls_back_to_global_connection(self):
         secret_id = "airflow/connections/test_postgres"
@@ -209,6 +234,7 @@ def test_get_variable_with_team_name(self):
 
         assert secrets_manager_backend.get_variable(key="hello", team_name="my_team") == "world"
 
+    @multi_team_enabled
     @mock_aws
     def test_global_caller_cannot_access_team_scoped_variable(self):
         secret_id = "airflow/variables/my_team--hello"

providers/amazon/tests/unit/amazon/aws/secrets/test_systems_manager.py

@@ -116,6 +116,7 @@ def test_get_conn_value_with_team_name(self):
         returned_uri = ssm_backend.get_conn_value(conn_id="test_postgres", team_name="my_team")
         assert returned_uri == "postgresql://airflow:airflow@host:5432/airflow"
 
+    @conf_vars({("core", "multi_team"): "True"})
     @mock_aws
     def test_global_caller_cannot_access_team_scoped_connection(self):
         param = {
@@ -127,6 +128,7 @@ def test_global_caller_cannot_access_team_scoped_connection(self):
         ssm_backend.client.put_parameter(**param)
         assert ssm_backend.get_conn_value(conn_id="my_team--test_postgres") is None
 
+    @conf_vars({("core", "multi_team"): "True"})
     @mock_aws
     def test_another_teams_secret_is_not_reachable(self):
         """A caller scoped to one team must not reach another team's parameter by naming it."""
@@ -140,6 +142,7 @@ def test_another_teams_secret_is_not_reachable(self):
 
         assert ssm_backend.get_conn_value(conn_id="my_team--test_postgres", team_name="other_team") is None
 
+    @conf_vars({("core", "multi_team"): "True"})
     @mock_aws
     def test_team_whose_name_extends_the_callers_is_not_reachable(self):
         """A prefix match on the caller's own namespace is not proof of ownership."""
@@ -153,6 +156,7 @@ def test_team_whose_name_extends_the_callers_is_not_reachable(self):
 
         assert ssm_backend.get_conn_value(conn_id="my_team--prod--test_postgres", team_name="my_team") is None
 
+    @conf_vars({("core", "multi_team"): "True"})
     @mock_aws
     def test_team_scoped_lookup_cannot_reach_a_longer_teams_namespace(self):
         """The team scoped name is not safe by construction -- the id can extend it.
@@ -172,6 +176,7 @@ def test_team_scoped_lookup_cannot_reach_a_longer_teams_namespace(self):
 
         assert ssm_backend.get_conn_value(conn_id="prod--test_postgres", team_name="my_team") is None
 
+    @conf_vars({("core", "multi_team"): "True"})
     @mock_aws
     def test_refusing_an_ambiguous_id_is_logged(self, caplog):
         """A silent ``None`` is indistinguishable from a missing secret, so the refusal is logged.
@@ -201,6 +206,22 @@ def test_refusing_an_ambiguous_id_is_logged(self, caplog):
             assert sum(refused_id in r.getMessage() for r in refusals) == 1
         assert all(TEAM_SEP in r.getMessage() for r in refusals)
 
+    @mock_aws
+    def test_ambiguous_id_resolves_when_multi_team_is_disabled(self):
+        """No team scoped parameter can exist without multi-team mode, so there is no ambiguity
+        to refuse -- an ordinary id containing the separator must resolve normally."""
+        param = {
+            "Name": "/airflow/connections/prod--test_postgres",
+            "Type": "String",
+            "Value": "postgresql://airflow:airflow@host:5432/airflow",
+        }
+        ssm_backend = SystemsManagerParameterStoreBackend()
+        ssm_backend.client.put_parameter(**param)
+
+        assert ssm_backend.get_conn_value(conn_id="prod--test_postgres") == (
+            "postgresql://airflow:airflow@host:5432/airflow"
+        )
+
     @mock_aws
     def test_team_caller_falls_back_to_global_connection(self):
         param = {
@@ -267,6 +288,7 @@ def test_get_variable_with_team_name(self):
 
         assert ssm_backend.get_variable(key="hello", team_name="my_team") == "world"
 
+    @conf_vars({("core", "multi_team"): "True"})
     @mock_aws
     def test_global_caller_cannot_access_team_scoped_variable(self):
         param = {"Name": "/airflow/variables/my_team--hello", "Type": "String", "Value": "world"}

providers/google/src/airflow/providers/google/cloud/secrets/secret_manager.py

@@ -23,7 +23,7 @@
 
 from google.auth.exceptions import DefaultCredentialsError
 
-from airflow.providers.common.compat.sdk import AirflowException
+from airflow.providers.common.compat.sdk import AirflowException, conf
 from airflow.providers.google.cloud._internal_client.secret_manager_client import _SecretManagerClient
 from airflow.providers.google.cloud.utils.credentials_provider import (
     _get_target_principal_and_delegates,
@@ -248,7 +248,12 @@ def _names_a_team_namespace(self, secret_id: str) -> bool:
         backend does, is wrong here: the inherited implementation prepends a separator to an
         empty prefix (``'' -> '-smtp_default'``) and normalizes nothing, so the guard would
         both mis-anchor and miss ids whose separator only appears after normalization.
+
+        Only checked in multi-team mode: ``team_name`` is never non-``None`` otherwise, so no
+        team scoped secret can exist to collide with.
         """
+        if not conf.getboolean("core", "multi_team", fallback=False):
+            return False
         return TEAM_SEP in secret_id
 
     def _get_secret(self, path_prefix: str, secret_id: str, team_name: str | None = None) -> str | None:

providers/google/tests/unit/google/cloud/secrets/test_secret_manager.py

@@ -28,6 +28,10 @@
 from airflow.providers.common.compat.sdk import AirflowException
 from airflow.providers.google.cloud.secrets.secret_manager import TEAM_SEP, CloudSecretManagerBackend
 
+from tests_common.test_utils.config import conf_vars
+
+multi_team_enabled = conf_vars({("core", "multi_team"): "True"})
+
 CREDENTIALS = "test-creds"
 KEY_FILE = "test-file.json"
 PROJECT_ID = "test-project-id"
@@ -318,6 +322,7 @@ def test_team_scoped_secret_is_resolved_for_its_own_team(self, mock_client, mock
 
         assert backend.get_conn_value(conn_id=CONN_ID, team_name=self.TEAM) == CONN_URI
 
+    @multi_team_enabled
     @mock.patch(MODULE_NAME + ".get_credentials_and_project_id")
     @mock.patch(MODULE_NAME + "._SecretManagerClient")
     def test_team_scoped_secret_is_not_resolved_for_another_team(self, mock_client, mock_get_creds):
@@ -329,6 +334,7 @@ def test_team_scoped_secret_is_not_resolved_for_another_team(self, mock_client,
 
         assert backend.get_conn_value(conn_id=encoded, team_name=self.OTHER_TEAM) is None
 
+    @multi_team_enabled
     @mock.patch(MODULE_NAME + ".get_credentials_and_project_id")
     @mock.patch(MODULE_NAME + "._SecretManagerClient")
     def test_team_scoped_secret_is_not_resolved_without_a_team_scope(self, mock_client, mock_get_creds):
@@ -339,6 +345,7 @@ def test_team_scoped_secret_is_not_resolved_without_a_team_scope(self, mock_clie
 
         assert backend.get_conn_value(conn_id=encoded) is None
 
+    @multi_team_enabled
     @mock.patch(MODULE_NAME + ".get_credentials_and_project_id")
     @mock.patch(MODULE_NAME + "._SecretManagerClient")
     def test_team_whose_name_extends_the_callers_is_not_readable(self, mock_client, mock_get_creds):
@@ -362,6 +369,7 @@ def test_team_agnostic_secret_is_resolved_for_any_team_scope(self, mock_client,
         assert backend.get_conn_value(conn_id=CONN_ID) == CONN_URI
         assert backend.get_conn_value(conn_id=CONN_ID, team_name=self.TEAM) == CONN_URI
 
+    @multi_team_enabled
     @mock.patch(MODULE_NAME + ".get_credentials_and_project_id")
     @mock.patch(MODULE_NAME + "._SecretManagerClient")
     def test_team_scoped_variable_is_not_resolved_for_another_team(self, mock_client, mock_get_creds):
@@ -416,6 +424,7 @@ def test_underscores_cannot_manufacture_a_team_namespace(self, mock_client, mock
         # the two names must not coincide in the first place
         assert backend._build_team_secret_name(CONNECTIONS_PREFIX, self.TEAM, f"prod__{CONN_ID}") != victim
 
+    @multi_team_enabled
     @mock.patch(MODULE_NAME + ".get_credentials_and_project_id")
     @mock.patch(MODULE_NAME + "._SecretManagerClient")
     def test_ambiguous_id_is_refused_even_for_its_own_team(self, mock_client, mock_get_creds):
@@ -432,6 +441,7 @@ def test_ambiguous_id_is_refused_even_for_its_own_team(self, mock_client, mock_g
 
         assert backend.get_conn_value(conn_id=ambiguous, team_name=self.TEAM) is None
 
+    @multi_team_enabled
     @mock.patch(MODULE_NAME + ".get_credentials_and_project_id")
     @mock.patch(MODULE_NAME + "._SecretManagerClient")
     def test_refusing_an_ambiguous_id_is_logged(self, mock_client, mock_get_creds, caplog):
@@ -448,6 +458,18 @@ def test_refusing_an_ambiguous_id_is_logged(self, mock_client, mock_get_creds, c
         assert all(r.levelname == "WARNING" for r in refusals)
         assert all(ambiguous in r.getMessage() for r in refusals)
 
+    @mock.patch(MODULE_NAME + ".get_credentials_and_project_id")
+    @mock.patch(MODULE_NAME + "._SecretManagerClient")
+    def test_ambiguous_id_resolves_when_multi_team_is_disabled(self, mock_client, mock_get_creds):
+        """No team scoped secret can exist without multi-team mode, so there is no ambiguity
+        to refuse -- an ordinary id containing the separator must resolve normally."""
+        mock_get_creds.return_value = CREDENTIALS, PROJECT_ID
+        backend, store = self._backend(mock_client)
+        ambiguous = f"prod{TEAM_SEP}{CONN_ID}"
+        store[backend.build_path(CONNECTIONS_PREFIX, ambiguous, SEP)] = CONN_URI
+
+        assert backend.get_conn_value(conn_id=ambiguous) == CONN_URI
+
     @mock.patch(MODULE_NAME + ".get_credentials_and_project_id")
     @mock.patch(MODULE_NAME + "._SecretManagerClient")
     def test_config_lookup_is_not_team_scoped(self, mock_client, mock_get_creds):

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

@@ -33,6 +33,7 @@
 from azure.identity import ClientSecretCredential, DefaultAzureCredential
 from azure.keyvault.secrets import SecretClient
 
+from airflow.providers.common.compat.sdk import conf
 from airflow.providers.microsoft.azure.utils import get_sync_default_azure_credential
 from airflow.secrets import BaseSecretsBackend
 from airflow.utils.log.logging_mixin import LoggingMixin
@@ -247,7 +248,12 @@ def _names_a_team_namespace(self, secret_id: str) -> bool:
         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.
+
+        Only checked in multi-team mode: ``team_name`` is never non-``None`` otherwise, so no
+        team scoped secret can exist to collide with.
         """
+        if not conf.getboolean("core", "multi_team", fallback=False):
+            return False
         return TEAM_SEP in self.build_path("", secret_id, self.sep)
 
     def _log_refusal(self, kind: str, secret_id: str) -> None:

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

@@ -24,7 +24,10 @@
 
 from airflow.providers.microsoft.azure.secrets.key_vault import AzureKeyVaultBackend
 
+from tests_common.test_utils.config import conf_vars
+
 KEY_VAULT_MODULE = "airflow.providers.microsoft.azure.secrets.key_vault"
+multi_team_enabled = conf_vars({("core", "multi_team"): "True"})
 
 
 class TestAzureKeyVaultBackend:
@@ -107,13 +110,15 @@ def test_get_variable_uses_team_secret_with_custom_prefix(self, mock_client):
         assert secret_val == "team-value"
         mock_client.get_secret.assert_called_once_with(name="custom-variables-team-a--hello")
 
+    @multi_team_enabled
     @mock.patch(f"{KEY_VAULT_MODULE}.AzureKeyVaultBackend.client")
     def test_get_variable_returns_none_for_team_scoped_key_without_team_name(self, mock_client):
         backend = AzureKeyVaultBackend()
 
         assert backend.get_variable("teama--hello") is None
         mock_client.get_secret.assert_not_called()
 
+    @multi_team_enabled
     @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.
@@ -133,6 +138,7 @@ def only_the_target_exists(name):
 
         assert backend.get_conn_value("teama--my_db", team_name="teamb") is None
 
+    @multi_team_enabled
     @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.
@@ -153,6 +159,7 @@ def only_the_target_exists(name):
 
         assert backend.get_conn_value("teama--prod--my_db", team_name="teama") is None
 
+    @multi_team_enabled
     @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.
@@ -174,6 +181,7 @@ def only_the_target_exists(name):
 
         assert backend.get_conn_value("prod--my_db", team_name="teama") is None
 
+    @multi_team_enabled
     @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.
@@ -193,6 +201,7 @@ def only_the_target_exists(name):
 
         assert backend.get_conn_value("prod__my_db", team_name="teama") is None
 
+    @multi_team_enabled
     @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."""
@@ -215,6 +224,17 @@ def test_refusing_an_ambiguous_id_is_logged(self, mock_client, caplog):
             assert sum(refused_id in r.getMessage() for r in refusals) == 1
         mock_client.get_secret.assert_not_called()
 
+    @mock.patch(f"{KEY_VAULT_MODULE}.AzureKeyVaultBackend.client")
+    def test_ambiguous_id_resolves_when_multi_team_is_disabled(self, mock_client):
+        """No team scoped secret can exist without multi-team mode, so there is no ambiguity
+        to refuse -- an ordinary id containing the separator must resolve normally."""
+        mock_client.get_secret.return_value = mock.Mock(value="world")
+        backend = AzureKeyVaultBackend()
+
+        assert backend.get_conn_value("prod--my_db") == "world"
+        assert backend.get_variable("prod__hello") == "world"
+        assert backend.get_config("prod--sql_alchemy_conn") == "world"
+
     @mock.patch(f"{KEY_VAULT_MODULE}.AzureKeyVaultBackend._get_secret")
     def test_variable_prefix_none_value(self, mock_get_secret):
         """