Fix CloudSecretsManagerBackend invalid connections_prefix (#7861)

(cherry picked from commit f9c2263)

airflow/contrib/secrets/gcp_secrets_manager.py

@@ -18,6 +18,7 @@
 """
 Objects relating to sourcing connections from GCP Secrets Manager
 """
+import re
 from typing import Optional
 
 from cached_property import cached_property
@@ -27,12 +28,15 @@
 
 from airflow import version
 
+from airflow.exceptions import AirflowException
 from airflow.contrib.utils.gcp_credentials_provider import (
     _get_scopes, get_credentials_and_project_id,
 )
 from airflow.secrets import BaseSecretsBackend
 from airflow.utils.log.logging_mixin import LoggingMixin
 
+SECRET_ID_PATTERN = r"^[a-zA-Z0-9-_]*$"
+
 
 class CloudSecretsManagerBackend(BaseSecretsBackend, LoggingMixin):
     """
@@ -44,10 +48,11 @@ class CloudSecretsManagerBackend(BaseSecretsBackend, LoggingMixin):
 
         [secrets]
         backend = airflow.contrib.secrets.gcp_secrets_manager.CloudSecretsManagerBackend
-        backend_kwargs = {"connections_prefix": "airflow/connections"}
+        backend_kwargs = {"connections_prefix": "airflow-connections", "sep": "-"}
 
-    For example, if secret id is ``airflow/connections/smtp_default``, this would be accessible
-    if you provide ``{"connections_prefix": "airflow/connections"}`` and request conn_id ``smtp_default``.
+    For example, if the Secrets Manager secret id is ``airflow-connections-smtp_default``, this would be
+    accessiblen if you provide ``{"connections_prefix": "airflow-connections", "sep": "-"}`` and request
+    conn_id ``smtp_default``. The full secret id should follow the pattern "[a-zA-Z0-9-_]".
 
     :param connections_prefix: Specifies the prefix of the secret to read to get Connections.
     :type connections_prefix: str
@@ -56,20 +61,33 @@ class CloudSecretsManagerBackend(BaseSecretsBackend, LoggingMixin):
     :type gcp_key_path: str
     :param gcp_scopes: Comma-separated string containing GCP scopes
     :type gcp_scopes: str
+    :param sep: separator used to concatenate connections_prefix and conn_id. Default: "-"
+    :type sep: str
     """
     def __init__(
         self,
-        connections_prefix="airflow/connections",  # type: str
+        connections_prefix="airflow-connections",  # type: str
         gcp_key_path=None,  # type: Optional[str]
         gcp_scopes=None,  # type: Optional[str]
+        sep="-",  # type: str
         **kwargs
     ):
-        self.connections_prefix = connections_prefix.rstrip("/")
+        super(CloudSecretsManagerBackend, self).__init__(**kwargs)
+        self.connections_prefix = connections_prefix
         self.gcp_key_path = gcp_key_path
         self.gcp_scopes = gcp_scopes
+        self.sep = sep
         self.credentials = None
         self.project_id = None
-        super(CloudSecretsManagerBackend, self).__init__(**kwargs)
+        if not self._is_valid_prefix_and_sep():
+            raise AirflowException(
+                "`connections_prefix` and `sep` should follows that pattern {}".format(
+                    SECRET_ID_PATTERN)
+            )
+
+    def _is_valid_prefix_and_sep(self):
+        prefix = self.connections_prefix + self.sep
+        return bool(re.match(SECRET_ID_PATTERN, prefix))
 
     @cached_property
     def client(self):
@@ -95,7 +113,11 @@ def get_conn_uri(self, conn_id):
         :param conn_id: connection id
         :type conn_id: str
         """
-        secret_id = self.build_path(connections_prefix=self.connections_prefix, conn_id=conn_id)
+        secret_id = self.build_path(
+            connections_prefix=self.connections_prefix,
+            conn_id=conn_id,
+            sep=self.sep
+        )
         # always return the latest version of the secret
         secret_version = "latest"
         name = self.client.secret_version_path(self.project_id, secret_id, secret_version)

airflow/secrets/base_secrets.py

@@ -31,17 +31,19 @@ def __init__(self, **kwargs):
         pass
 
     @staticmethod
-    def build_path(connections_prefix, conn_id):
-        # type: (str, str) -> str
+    def build_path(connections_prefix, conn_id, sep="/"):
+        # type: (str, str, str) -> str
         """
         Given conn_id, build path for Secrets Backend
 
         :param connections_prefix: prefix of the secret to read to get Connections
         :type connections_prefix: str
         :param conn_id: connection id
         :type conn_id: str
+        :param sep: separator used to concatenate connections_prefix and conn_id. Default: "/"
+        :type sep: str
         """
-        return "{}/{}".format(connections_prefix, conn_id)
+        return "{}{}{}".format(connections_prefix, sep, conn_id)
 
     def get_conn_uri(self, conn_id):
         # type: (str) -> Optional[str]

docs/howto/use-alternative-secrets-backend.rst

@@ -155,14 +155,17 @@ Available parameters to ``backend_kwargs``:
 * ``connections_prefix``: Specifies the prefix of the secret to read to get Connections.
 * ``gcp_key_path``: Path to GCP Credential JSON file
 * ``gcp_scopes``: Comma-separated string containing GCP scopes
+* ``sep``: separator used to concatenate connections_prefix and conn_id. Default: "-"
+
+Note: The full GCP Secrets Manager secret id should follow the pattern "[a-zA-Z0-9-_]".
 
 Here is a sample configuration:
 
 .. code-block:: ini
 
     [secrets]
     backend = airflow.contrib.secrets.gcp_secrets_manager.CloudSecretsManagerBackend
-    backend_kwargs = {"connections_prefix": "airflow/connections"}
+    backend_kwargs = {"connections_prefix": "airflow-connections", "sep": "-"}
 
 When ``gcp_key_path`` is not provided, it will use the Application Default Credentials in the current environment. You can set up the credentials with:
 

tests/contrib/secrets/test_gcp_secrets_manager.py

@@ -22,21 +22,52 @@
 from parameterized import parameterized
 
 from airflow.contrib.secrets.gcp_secrets_manager import CloudSecretsManagerBackend
+from airflow.exceptions import AirflowException
 from airflow.models import Connection
 from tests.compat import mock
 
 CREDENTIALS = 'test-creds'
 KEY_FILE = 'test-file.json'
 PROJECT_ID = 'test-project-id'
+CONNECTIONS_PREFIX = "test-connections"
+SEP = '-'
 CONN_ID = 'test-postgres'
 CONN_URI = 'postgresql://airflow:airflow@host:5432/airflow'
 
 MODULE_NAME = "airflow.contrib.secrets.gcp_secrets_manager"
 
 
 class TestCloudSecretsManagerBackend(TestCase):
+    def test_default_valid_and_sep(self):
+        backend = CloudSecretsManagerBackend()
+        self.assertTrue(backend._is_valid_prefix_and_sep())
+
     @parameterized.expand([
-        "airflow/connections",
+        ("colon:", "not:valid", ":"),
+        ("slash/", "not/valid", "/"),
+        ("space_with_char", "a b", ""),
+        ("space_only", "", " ")
+    ])
+    def test_raise_exception_with_invalid_prefix_sep(self, _, prefix, sep):
+        with self.assertRaises(AirflowException):
+            CloudSecretsManagerBackend(connections_prefix=prefix, sep=sep)
+
+    @parameterized.expand([
+        ("dash-", "valid1", "-", True),
+        ("underscore_", "isValid", "_", True),
+        ("empty_string", "", "", True),
+        ("space_prefix", " ", "", False),
+        ("space_sep", "", " ", False),
+        ("colon:", "not:valid", ":", False)
+    ])
+    def test_is_valid_prefix_and_sep(self, _, prefix, sep, is_valid):
+        backend = CloudSecretsManagerBackend()
+        backend.connections_prefix = prefix
+        backend.sep = sep
+        self.assertEqual(backend._is_valid_prefix_and_sep(), is_valid)
+
+    @parameterized.expand([
+        "airflow-connections",
         "connections",
         "airflow"
     ])
@@ -52,10 +83,11 @@ def test_get_conn_uri(self, connections_prefix, mock_client_callable, mock_get_c
         mock_client.access_secret_version.return_value = test_response
 
         secrets_manager_backend = CloudSecretsManagerBackend(connections_prefix=connections_prefix)
+        secret_id = secrets_manager_backend.build_path(connections_prefix, CONN_ID, SEP)
         returned_uri = secrets_manager_backend.get_conn_uri(conn_id=CONN_ID)
         self.assertEqual(CONN_URI, returned_uri)
         mock_client.secret_version_path.assert_called_once_with(
-            PROJECT_ID, "{}/{}".format(connections_prefix, CONN_ID), "latest"
+            PROJECT_ID, secret_id, "latest"
         )
 
     @mock.patch(MODULE_NAME + ".get_credentials_and_project_id")
@@ -76,9 +108,7 @@ def test_get_conn_uri_non_existent_key(self, mock_client_callable, mock_get_cred
         # The requested secret id or secret version does not exist
         mock_client.access_secret_version.side_effect = NotFound('test-msg')
 
-        connections_prefix = "airflow/connections"
-
-        secrets_manager_backend = CloudSecretsManagerBackend(connections_prefix=connections_prefix)
+        secrets_manager_backend = CloudSecretsManagerBackend(connections_prefix=CONNECTIONS_PREFIX)
 
         self.assertIsNone(secrets_manager_backend.get_conn_uri(conn_id=CONN_ID))
         self.assertEqual([], secrets_manager_backend.get_connections(conn_id=CONN_ID))

tests/secrets/test_secrets_backends.py

@@ -19,7 +19,10 @@
 import os
 import unittest
 
+from parameterized import parameterized
+
 from airflow.models import Connection
+from airflow.secrets.base_secrets import BaseSecretsBackend
 from airflow.secrets.environment_variables import EnvironmentVariablesBackend
 from airflow.secrets.metastore import MetastoreBackend
 from airflow.utils.db import create_session
@@ -37,6 +40,15 @@ def __init__(self, conn_id, variation):
 
 
 class TestBaseSecretsBackend(unittest.TestCase):
+
+    @parameterized.expand([
+        ('default', {"connections_prefix": "PREFIX", "conn_id": "ID"}, "PREFIX/ID"),
+        ('with_sep', {"connections_prefix": "PREFIX", "conn_id": "ID", "sep": "-"}, "PREFIX-ID")
+    ])
+    def test_build_path(self, _, kwargs, output):
+        build_path = BaseSecretsBackend.build_path
+        self.assertEqual(build_path(**kwargs), output)
+
     def test_env_secrets_backend(self):
         sample_conn_1 = SampleConn("sample_1", "A")
         env_secrets_backend = EnvironmentVariablesBackend()