Prevent unauthorized access to team-scoped secrets in SM and SSM

providers/amazon/docs/secrets-backends/aws-secrets-manager.rst

@@ -159,6 +159,24 @@ For example, if you want to only lookup connections starting by "m" in AWS Secre
       "profile_name": "default"
     }
 
+Multi-Team Support
+^^^^^^^^^^^^^^^^^^
+
+In multi-team mode, team-scoped secrets use ``--`` as a separator between the team name
+and the secret id. For example, a connection ``smtp_default`` owned by team ``marketing``
+should be stored at ``airflow/connections/marketing--smtp_default``, and a variable ``hello``
+owned by the same team should be stored at ``airflow/variables/marketing--hello``.
+
+Task authors request connections and variables by their normal ids. If a team-scoped
+secret is not found, the backend falls back to the global path (e.g.
+``airflow/connections/smtp_default``).
+
+.. note::
+
+    Connection ids and variable keys matching ``<team>--<name>`` are reserved for
+    team-scoped lookups. A request without team context for a key matching this pattern
+    will return ``None`` to prevent cross-team access.
+
 Example of storing Google Secrets in AWS Secrets Manager
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 For connecting to a google cloud connection, all the fields must be in the extra field. For example:

providers/amazon/docs/secrets-backends/aws-ssm-parameter-store.rst

@@ -134,3 +134,21 @@ If you have set ``variables_prefix`` as ``/airflow/variables``, then for an Vari
 you would want to store your Variable at ``/airflow/variables/hello``.
 
 Optionally you can supply a profile name to reference aws profile, e.g. defined in ``~/.aws/config``.
+
+Multi-Team Support
+""""""""""""""""""
+
+In multi-team mode, team-scoped secrets use ``--`` as a separator between the team name
+and the secret id. For example, a connection ``smtp_default`` owned by team ``marketing``
+should be stored at ``/airflow/connections/marketing--smtp_default``, and a variable ``hello``
+owned by the same team should be stored at ``/airflow/variables/marketing--hello``.
+
+Task authors request connections and variables by their normal ids. If a team-scoped
+secret is not found, the backend falls back to the global path (e.g.
+``/airflow/connections/smtp_default``).
+
+.. note::
+
+    Connection ids and variable keys matching ``<team>--<name>`` are reserved for
+    team-scoped lookups. A request without team context for a key matching this pattern
+    will return ``None`` to prevent cross-team access.

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

@@ -253,29 +253,15 @@ def get_config(self, key: str) -> str | None:
 
         return self._get_secret(self.config_prefix, key, self.config_lookup_pattern)
 
-    def _get_secret(
-        self, path_prefix, secret_id: str, lookup_pattern: str | None, team_name: str | None = None
-    ) -> str | None:
+    def _get_secret_value(self, secret_id: str, secrets_path: str) -> str | None:
         """
-        Get secret value from Secrets Manager.
+        Fetch a secret value from Secrets Manager.
 
-        :param path_prefix: Prefix for the Path to get Secret
-        :param secret_id: Secret Key
-        :param lookup_pattern: If provided, `secret_id` must match this pattern to look up the secret in
-            Secrets Manager
+        :param secret_id: Secret Key, used for logging on not-found.
+        :param secrets_path: Full path to look up in Secrets Manager.
+        :return: The secret value, or None if not found or on error.
         """
-        if lookup_pattern and not re.match(lookup_pattern, secret_id, re.IGNORECASE):
-            return None
-
         error_msg = "An error occurred when calling the get_secret_value operation"
-        if path_prefix and team_name:
-            secrets_path = self.build_path(path_prefix, team_name, self.sep)
-            secrets_path = self.build_path(secrets_path, secret_id, self.sep)
-        elif path_prefix:
-            secrets_path = self.build_path(path_prefix, secret_id, self.sep)
-        else:
-            secrets_path = secret_id
-
         try:
             response = self.client.get_secret_value(
                 SecretId=secrets_path,
@@ -316,3 +302,32 @@ def _get_secret(
                 exc_info=True,
             )
             return None
+
+    def _get_secret(
+        self, path_prefix, secret_id: str, lookup_pattern: str | None, team_name: str | None = None
+    ) -> str | None:
+        """
+        Get secret value from Secrets Manager.
+
+        :param path_prefix: Prefix for the Path to get Secret
+        :param secret_id: Secret Key
+        :param lookup_pattern: If provided, `secret_id` must match this pattern to look up the secret in
+            Secrets Manager
+        :param team_name: Team name associated to the task trying to access the variable (if any)
+        """
+        if lookup_pattern and not re.match(lookup_pattern, secret_id, re.IGNORECASE):
+            return None
+        if team_name is None and re.fullmatch(r"[^-]+--.+", secret_id):
+            return None
+        if path_prefix and team_name:
+            secrets_path = self.build_path(path_prefix, f"{team_name}--{secret_id}", self.sep)
+            value = self._get_secret_value(secret_id, secrets_path)
+            if value is not None:
+                return value
+
+        if path_prefix:
+            secrets_path = self.build_path(path_prefix, secret_id, self.sep)
+        else:
+            secrets_path = secret_id
+
+        return self._get_secret_value(secret_id, secrets_path)

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

@@ -169,6 +169,19 @@ def get_config(self, key: str) -> str | None:
 
         return self._get_secret(self.config_prefix, key, self.config_lookup_pattern)
 
+    def _get_parameter_value(self, ssm_path: str) -> str | None:
+        """
+        Fetch a parameter value from SSM, returning None if not found.
+
+        :param ssm_path: SSM parameter path
+        """
+        try:
+            response = self.client.get_parameter(Name=ssm_path, WithDecryption=True)
+            return response["Parameter"]["Value"]
+        except self.client.exceptions.ParameterNotFound:
+            self.log.debug("Parameter %s not found.", ssm_path)
+            return None
+
     def _get_secret(
         self, path_prefix: str, secret_id: str, lookup_pattern: str | None, team_name: str | None = None
     ) -> str | None:
@@ -183,19 +196,19 @@ def _get_secret(
         """
         if lookup_pattern and not re.match(lookup_pattern, secret_id, re.IGNORECASE):
             return None
+        if team_name is None and re.fullmatch(r"[^-]+--.+", secret_id):
+            return None
         if team_name:
-            ssm_path = self.build_path(path_prefix, team_name)
-            ssm_path = self.build_path(ssm_path, secret_id)
-        else:
-            ssm_path = self.build_path(path_prefix, secret_id)
+            ssm_path = self.build_path(path_prefix, f"{team_name}--{secret_id}")
+            ssm_path = self._ensure_leading_slash(ssm_path)
+            value = self._get_parameter_value(ssm_path)
+            if value is not None:
+                return value
+
+        ssm_path = self.build_path(path_prefix, secret_id)
         ssm_path = self._ensure_leading_slash(ssm_path)
 
-        try:
-            response = self.client.get_parameter(Name=ssm_path, WithDecryption=True)
-            return response["Parameter"]["Value"]
-        except self.client.exceptions.ParameterNotFound:
-            self.log.debug("Parameter %s not found.", ssm_path)
-            return None
+        return self._get_parameter_value(ssm_path=ssm_path)
 
     def _ensure_leading_slash(self, ssm_path: str):
         """

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

@@ -67,7 +67,7 @@ def test_get_conn_value_non_existent_key(self):
 
     @mock_aws
     def test_get_conn_value_with_team_name(self):
-        secret_id = "airflow/connections/my_team/test_postgres"
+        secret_id = "airflow/connections/my_team--test_postgres"
         create_param = {
             "Name": secret_id,
             "SecretString": "postgresql://airflow:airflow@host:5432/airflow",
@@ -79,6 +79,35 @@ 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"
 
+    @mock_aws
+    def test_global_caller_cannot_access_team_scoped_connection(self):
+        secret_id = "airflow/connections/my_team--test_postgres"
+        create_param = {
+            "Name": secret_id,
+            "SecretString": "postgresql://airflow:airflow@host:5432/airflow",
+        }
+
+        secrets_manager_backend = SecretsManagerBackend()
+        secrets_manager_backend.client.create_secret(**create_param)
+
+        assert secrets_manager_backend.get_conn_value(conn_id="my_team--test_postgres") is None
+
+    @mock_aws
+    def test_team_caller_falls_back_to_global_connection(self):
+        secret_id = "airflow/connections/test_postgres"
+        create_param = {
+            "Name": secret_id,
+            "SecretString": "postgresql://airflow:airflow@host:5432/airflow",
+        }
+
+        secrets_manager_backend = SecretsManagerBackend()
+        secrets_manager_backend.client.create_secret(**create_param)
+
+        returned_uri = secrets_manager_backend.get_conn_value(
+            conn_id="test_postgres", team_name="non_existent_team"
+        )
+        assert returned_uri == "postgresql://airflow:airflow@host:5432/airflow"
+
     @mock_aws
     def test_get_variable(self):
         secret_id = "airflow/variables/hello"
@@ -106,14 +135,24 @@ def test_get_variable_non_existent_key(self):
 
     @mock_aws
     def test_get_variable_with_team_name(self):
-        secret_id = "airflow/variables/my_team/hello"
+        secret_id = "airflow/variables/my_team--hello"
         create_param = {"Name": secret_id, "SecretString": "world"}
 
         secrets_manager_backend = SecretsManagerBackend()
         secrets_manager_backend.client.create_secret(**create_param)
 
         assert secrets_manager_backend.get_variable(key="hello", team_name="my_team") == "world"
 
+    @mock_aws
+    def test_global_caller_cannot_access_team_scoped_variable(self):
+        secret_id = "airflow/variables/my_team--hello"
+        create_param = {"Name": secret_id, "SecretString": "world"}
+
+        secrets_manager_backend = SecretsManagerBackend()
+        secrets_manager_backend.client.create_secret(**create_param)
+
+        assert secrets_manager_backend.get_variable(key="my_team--hello") is None
+
     @mock_aws
     def test_get_config_non_existent_key(self):
         """

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

@@ -103,7 +103,7 @@ def test_get_conn_value_non_existent_key(self):
     @mock_aws
     def test_get_conn_value_with_team_name(self):
         param = {
-            "Name": "/airflow/connections/my_team/test_postgres",
+            "Name": "/airflow/connections/my_team--test_postgres",
             "Type": "String",
             "Value": "postgresql://airflow:airflow@host:5432/airflow",
         }
@@ -112,6 +112,29 @@ 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"
 
+    @mock_aws
+    def test_global_caller_cannot_access_team_scoped_connection(self):
+        param = {
+            "Name": "/airflow/connections/my_team--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="my_team--test_postgres") is None
+
+    @mock_aws
+    def test_team_caller_falls_back_to_global_connection(self):
+        param = {
+            "Name": "/airflow/connections/test_postgres",
+            "Type": "String",
+            "Value": "postgresql://airflow:airflow@host:5432/airflow",
+        }
+        ssm_backend = SystemsManagerParameterStoreBackend()
+        ssm_backend.client.put_parameter(**param)
+        returned_uri = ssm_backend.get_conn_value(conn_id="test_postgres", team_name="non_existent_team")
+        assert returned_uri == "postgresql://airflow:airflow@host:5432/airflow"
+
     @mock_aws
     def test_get_variable(self):
         param = {"Name": "/airflow/variables/hello", "Type": "String", "Value": "world"}
@@ -159,13 +182,22 @@ def test_get_variable_non_existent_key(self):
 
     @mock_aws
     def test_get_variable_with_team_name(self):
-        param = {"Name": "/airflow/variables/my_team/hello", "Type": "String", "Value": "world"}
+        param = {"Name": "/airflow/variables/my_team--hello", "Type": "String", "Value": "world"}
 
         ssm_backend = SystemsManagerParameterStoreBackend()
         ssm_backend.client.put_parameter(**param)
 
         assert ssm_backend.get_variable(key="hello", team_name="my_team") == "world"
 
+    @mock_aws
+    def test_global_caller_cannot_access_team_scoped_variable(self):
+        param = {"Name": "/airflow/variables/my_team--hello", "Type": "String", "Value": "world"}
+
+        ssm_backend = SystemsManagerParameterStoreBackend()
+        ssm_backend.client.put_parameter(**param)
+
+        assert ssm_backend.get_variable(key="my_team--hello") is None
+
     @conf_vars(
         {
             ("secrets", "backend"): "airflow.providers.amazon.aws.secrets.systems_manager."