Apply the team scope in the Google Secret Manager backend

providers/google/docs/changelog.rst

@@ -27,6 +27,20 @@
 Changelog
 ---------
 
+.. note::
+    ``CloudSecretManagerBackend`` now applies the team scope it is given. Previously
+    ``get_conn_value`` and ``get_variable`` accepted a ``team_name`` and dropped it, so every
+    lookup resolved the team-agnostic secret name regardless of the team.
+
+    In multi-team deployments a team-scoped secret must now be stored under
+    ``{prefix}{sep}{team_name}--{secret_id}``; the team-agnostic name is still used as a
+    fallback when no team-scoped secret exists. Secrets that were relied on to resolve for a
+    specific team have to be re-stored under the team-scoped name.
+
+    Connection ids and variable keys containing ``--`` are no longer resolved in either mode,
+    because a team name may itself contain ``--`` and the resulting name would be ambiguous.
+    See :doc:`/secrets-backends/google-cloud-secret-manager-backend` for the full convention.
+
 22.3.0
 ......
 

providers/google/docs/secrets-backends/google-cloud-secret-manager-backend.rst

@@ -196,6 +196,48 @@ To check the variables is correctly read from the backend secret, you can use ``
     $ airflow variables get first-variable
     secret_content
 
+Multi-team lookup
+=================
+
+In multi-team mode this backend looks for a team-scoped secret first and falls back to the
+team-agnostic name when no team-scoped secret exists.
+
+For connections:
+
+* Team-scoped: ``{connections_prefix}{sep}{team_name}--{conn_id}``
+* Team-agnostic fallback: ``{connections_prefix}{sep}{conn_id}``
+
+For variables:
+
+* Team-scoped: ``{variables_prefix}{sep}{team_name}--{key}``
+* Team-agnostic fallback: ``{variables_prefix}{sep}{key}``
+
+The team name and the secret id are separated by ``--``, while ``sep`` (default ``-``) separates
+the prefix from what follows. So with ``connections_prefix="airflow-connections"``,
+``team_name="team_a"`` and ``conn_id="smtp_default"``, the backend looks up
+``airflow-connections-team_a--smtp_default`` before falling back to
+``airflow-connections-smtp_default``.
+
+.. note::
+
+    Unlike the Azure Key Vault backend, this backend does **not** normalize underscores to the
+    separator. The secret id keeps the exact form you request it under, in the team-scoped name
+    as well as the team-agnostic one.
+
+Because a team name may itself contain ``--``, an id that contains ``--`` is ambiguous — team
+``team_a`` with conn_id ``prod--db`` and team ``team_a--prod`` with conn_id ``db`` would name the
+same secret. Such an id is therefore never resolved, for team-scoped and team-agnostic lookups
+alike, and the backend never parses an id to infer which team it belongs to. A warning is logged
+whenever an id is refused for this reason, so the resulting ``None`` is not mistaken for a missing
+secret. Avoid ``--`` in connection ids and variable keys.
+
+.. warning::
+
+    This refusal applies whether or not you use teams. If you already store a connection or
+    variable whose id contains ``--``, it stops resolving after upgrading and you must rename it.
+
+``get_config`` is not team-scoped and is unaffected by any of the above.
+
 Clean up
 ========
 

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

@@ -37,6 +37,10 @@
 
 SECRET_ID_PATTERN = r"^[a-zA-Z0-9-_]*$"
 
+# Separator between the team name and the secret id in a team scoped secret name.
+# Matches the convention the other secrets backends use.
+TEAM_SEP = "--"
+
 
 class CloudSecretManagerBackend(BaseSecretsBackend, LoggingMixin):
     """
@@ -166,7 +170,11 @@ def get_conn_value(self, conn_id: str, team_name: str | None = None) -> str | No
         if self.connections_prefix is None:
             return None
 
-        return self._get_secret(self.connections_prefix, conn_id)
+        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)
 
     def get_variable(self, key: str, team_name: str | None = None) -> str | None:
         """
@@ -179,7 +187,11 @@ def get_variable(self, key: str, team_name: str | None = None) -> str | None:
         if self.variables_prefix is None:
             return None
 
-        return self._get_secret(self.variables_prefix, key)
+        if self._names_a_team_namespace(key):
+            self._log_refusal("variable", key)
+            return None
+
+        return self._get_secret(self.variables_prefix, key, team_name)
 
     def get_config(self, key: str) -> str | None:
         """
@@ -193,12 +205,75 @@ def get_config(self, key: str) -> str | None:
 
         return self._get_secret(self.config_prefix, key)
 
-    def _get_secret(self, path_prefix: str, secret_id: str) -> str | None:
+    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 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 _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.
+
+        The secret id is used verbatim. Normalizing it (``_`` -> ``sep``, as the Azure Key
+        Vault backend does) would let a plain id manufacture ``TEAM_SEP``: team ``a`` asking
+        for ``b__c`` would build the same name as team ``a--b`` asking for ``c``. It would
+        also contradict this backend's naming, which keeps underscores everywhere else --
+        unlike Azure, this backend does not override :meth:`build_path`.
+        """
+        team_prefix = self.build_path(path_prefix, team_name, self.sep)
+        return f"{team_prefix}{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 ``<prefix><sep><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
+        harbor 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 raw id is matched. Routing it through :meth:`build_path` first, as the Azure
+        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.
+        """
+        return TEAM_SEP in secret_id
+
+    def _get_secret(self, path_prefix: str, secret_id: str, team_name: str | None = None) -> str | None:
         """
         Get secret value from the SecretManager based on prefix.
 
         :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)
         """
-        secret_id = self.build_path(path_prefix, secret_id, self.sep)
-        return self.client.get_secret(secret_id=secret_id, project_id=self.project_id)
+        # ``self.client`` builds a new ``_SecretManagerClient`` -- and with it a real gRPC
+        # client -- on every access, so it is bound once for both lookups below.
+        client = self.client
+
+        # The team scoped name is tried first and is safe by construction: it can only ever
+        # build the caller's own namespace. Ids that would make that name ambiguous are
+        # refused by the callers before reaching here.
+        if team_name:
+            team_secret = client.get_secret(
+                secret_id=self._build_team_secret_name(path_prefix, team_name, secret_id),
+                project_id=self.project_id,
+            )
+            if team_secret is not None:
+                return team_secret
+
+        return client.get_secret(
+            secret_id=self.build_path(path_prefix, secret_id, self.sep), project_id=self.project_id
+        )

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

@@ -26,14 +26,15 @@
 
 from airflow.models import Connection
 from airflow.providers.common.compat.sdk import AirflowException
-from airflow.providers.google.cloud.secrets.secret_manager import CloudSecretManagerBackend
+from airflow.providers.google.cloud.secrets.secret_manager import TEAM_SEP, CloudSecretManagerBackend
 
 CREDENTIALS = "test-creds"
 KEY_FILE = "test-file.json"
 PROJECT_ID = "test-project-id"
 OVERRIDDEN_PROJECT_ID = "overridden-test-project-id"
 CONNECTIONS_PREFIX = "test-connections"
 VARIABLES_PREFIX = "test-variables"
+CONFIG_PREFIX = "test-config"
 SEP = "-"
 CONN_ID = "test-postgres"
 CONN_URI = "postgresql://airflow:airflow@host:5432/airflow"
@@ -272,3 +273,193 @@ def test_init_succeeds_with_explicit_project_id(self, mock_client_callable, mock
 
         backend = CloudSecretManagerBackend(project_id="explicit-project")
         assert backend.project_id == "explicit-project"
+
+
+class TestCloudSecretManagerBackendTeamScope:
+    """A team scoped secret must only be resolvable for the team it is stored for."""
+
+    TEAM = "team_a"
+    OTHER_TEAM = "team_b"
+    # A team name may itself contain the team separator, so one team's namespace can start
+    # with another's. Treating a prefix match as ownership is what this guards against.
+    EXTENDING_TEAM = "team_a--prod"
+
+    @staticmethod
+    def _backend(mock_client_callable):
+        """
+        Build a backend over an in-memory secret store keyed by full secret name.
+
+        The store is returned rather than taken as an argument so tests can populate it after
+        building the backend -- the lookup closes over it by reference. Names are derived from
+        the backend, so it has to exist first.
+        """
+        store: dict[str, str] = {}
+        client = mock.MagicMock()
+        client.get_secret.side_effect = lambda secret_id, project_id, **kwargs: store.get(secret_id)
+        mock_client_callable.return_value = client
+        backend = CloudSecretManagerBackend(
+            connections_prefix=CONNECTIONS_PREFIX,
+            variables_prefix=VARIABLES_PREFIX,
+            config_prefix=CONFIG_PREFIX,
+            project_id=PROJECT_ID,
+        )
+        return backend, store
+
+    @staticmethod
+    def _team_name(backend, team, secret_id, prefix=CONNECTIONS_PREFIX):
+        return backend._build_team_secret_name(prefix, team, secret_id)
+
+    @mock.patch(MODULE_NAME + ".get_credentials_and_project_id")
+    @mock.patch(MODULE_NAME + "._SecretManagerClient")
+    def test_team_scoped_secret_is_resolved_for_its_own_team(self, mock_client, mock_get_creds):
+        mock_get_creds.return_value = CREDENTIALS, PROJECT_ID
+        backend, store = self._backend(mock_client)
+        store[self._team_name(backend, self.TEAM, CONN_ID)] = CONN_URI
+
+        assert backend.get_conn_value(conn_id=CONN_ID, team_name=self.TEAM) == CONN_URI
+
+    @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):
+        """The whole point: team B must not reach team A's secret by spelling out its name."""
+        mock_get_creds.return_value = CREDENTIALS, PROJECT_ID
+        backend, store = self._backend(mock_client)
+        encoded = f"{self.TEAM}--{CONN_ID}"
+        store[self._team_name(backend, self.TEAM, CONN_ID)] = CONN_URI
+
+        assert backend.get_conn_value(conn_id=encoded, team_name=self.OTHER_TEAM) is None
+
+    @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):
+        mock_get_creds.return_value = CREDENTIALS, PROJECT_ID
+        backend, store = self._backend(mock_client)
+        encoded = f"{self.TEAM}--{CONN_ID}"
+        store[self._team_name(backend, self.TEAM, CONN_ID)] = CONN_URI
+
+        assert backend.get_conn_value(conn_id=encoded) is None
+
+    @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):
+        """A prefix match on the caller's namespace is not proof of ownership."""
+        mock_get_creds.return_value = CREDENTIALS, PROJECT_ID
+        backend, store = self._backend(mock_client)
+        store[self._team_name(backend, self.EXTENDING_TEAM, CONN_ID)] = CONN_URI
+        encoded = f"{self.EXTENDING_TEAM}--{CONN_ID}"
+
+        assert backend.get_conn_value(conn_id=encoded, team_name=self.TEAM) is None
+        # and the owning team still reaches it normally, with the bare id
+        assert backend.get_conn_value(conn_id=CONN_ID, team_name=self.EXTENDING_TEAM) == CONN_URI
+
+    @mock.patch(MODULE_NAME + ".get_credentials_and_project_id")
+    @mock.patch(MODULE_NAME + "._SecretManagerClient")
+    def test_team_agnostic_secret_is_resolved_for_any_team_scope(self, mock_client, mock_get_creds):
+        mock_get_creds.return_value = CREDENTIALS, PROJECT_ID
+        backend, store = self._backend(mock_client)
+        store[backend.build_path(CONNECTIONS_PREFIX, CONN_ID, SEP)] = CONN_URI
+
+        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
+
+    @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):
+        """Variables share the lookup, so they must be scoped identically."""
+        mock_get_creds.return_value = CREDENTIALS, PROJECT_ID
+        backend, store = self._backend(mock_client)
+        encoded = f"{self.TEAM}--{VAR_KEY}"
+        store[self._team_name(backend, self.TEAM, VAR_KEY, VARIABLES_PREFIX)] = VAR_VALUE
+
+        assert backend.get_variable(key=VAR_KEY, team_name=self.TEAM) == VAR_VALUE
+        assert backend.get_variable(key=encoded, team_name=self.OTHER_TEAM) is None
+
+    @mock.patch(MODULE_NAME + ".get_credentials_and_project_id")
+    @mock.patch(MODULE_NAME + "._SecretManagerClient")
+    def test_team_secret_name_is_built_literally(self, mock_client, mock_get_creds):
+        """
+        Pin the literal name rather than deriving it from the method under test.
+
+        Every other test in this class keys its store with ``_build_team_secret_name``, so it
+        agrees with the implementation by construction and cannot see a naming defect.
+        """
+        mock_get_creds.return_value = CREDENTIALS, PROJECT_ID
+        backend, _ = self._backend(mock_client)
+
+        assert (
+            backend._build_team_secret_name(CONNECTIONS_PREFIX, "team_a", "test_postgres")
+            == f"{CONNECTIONS_PREFIX}{SEP}team_a{TEAM_SEP}test_postgres"
+        )
+        # Underscores survive in the secret id, matching the team agnostic name this backend
+        # documents -- unlike Azure, nothing here normalises ``_`` to the separator.
+        assert (
+            backend._build_team_secret_name(CONNECTIONS_PREFIX, "team_a", "smtp_default")
+            == f"{CONNECTIONS_PREFIX}{SEP}team_a{TEAM_SEP}smtp_default"
+        )
+
+    @mock.patch(MODULE_NAME + ".get_credentials_and_project_id")
+    @mock.patch(MODULE_NAME + "._SecretManagerClient")
+    def test_underscores_cannot_manufacture_a_team_namespace(self, mock_client, mock_get_creds):
+        """
+        An id containing ``__`` must not resolve into a team whose name ends with that segment.
+
+        Normalising ``_`` to the separator when building the team scoped name (as the Azure
+        backend does) made team ``team_a`` asking for ``prod__conn`` build exactly the name
+        team ``team_a--prod`` stores ``conn`` under.
+        """
+        mock_get_creds.return_value = CREDENTIALS, PROJECT_ID
+        backend, store = self._backend(mock_client)
+        victim = backend._build_team_secret_name(CONNECTIONS_PREFIX, self.EXTENDING_TEAM, CONN_ID)
+        store[victim] = CONN_URI
+
+        assert backend.get_conn_value(conn_id=f"prod__{CONN_ID}", team_name=self.TEAM) is None
+        # 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
+
+    @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):
+        """
+        An id containing the separator is ambiguous in both directions, so it is never resolved.
+
+        ``team_a`` with id ``b--c`` and ``team_a--b`` with id ``c`` name the same secret, so
+        honouring the id for its own team would still hand one team the other's secret.
+        """
+        mock_get_creds.return_value = CREDENTIALS, PROJECT_ID
+        backend, store = self._backend(mock_client)
+        ambiguous = f"prod{TEAM_SEP}{CONN_ID}"
+        store[backend._build_team_secret_name(CONNECTIONS_PREFIX, self.TEAM, ambiguous)] = CONN_URI
+
+        assert backend.get_conn_value(conn_id=ambiguous, team_name=self.TEAM) is None
+
+    @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):
+        """A silent ``None`` is indistinguishable from a missing secret, so the refusal is logged."""
+        mock_get_creds.return_value = CREDENTIALS, PROJECT_ID
+        backend, _ = self._backend(mock_client)
+        ambiguous = f"prod{TEAM_SEP}{CONN_ID}"
+
+        assert backend.get_conn_value(conn_id=ambiguous) is None
+        assert backend.get_variable(key=ambiguous) 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)
+        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_config_lookup_is_not_team_scoped(self, mock_client, mock_get_creds):
+        """
+        ``get_config`` has no team scope, so the namespace guard must not apply to it.
+
+        Refusing config keys containing the separator would be stricter than the Azure backend
+        this convention comes from, and there is no team namespace for a config key to invade.
+        """
+        mock_get_creds.return_value = CREDENTIALS, PROJECT_ID
+        backend, store = self._backend(mock_client)
+        key = f"some{TEAM_SEP}option"
+        store[backend.build_path(CONFIG_PREFIX, key, SEP)] = "config-value"
+
+        assert backend.get_config(key) == "config-value"