Catch Permission Denied exception when getting secret from GCP Secret…

… Manager. (#10326)
apache · Aug 14, 2020 · 47387a6 · 47387a6
1 parent 2d4e44c
commit 47387a6
Show file tree

Hide file tree

Showing 2 changed files with 23 additions and 2 deletions.
diff --git a/airflow/providers/google/cloud/_internal_client/secret_manager_client.py b/airflow/providers/google/cloud/_internal_client/secret_manager_client.py
@@ -20,7 +20,7 @@
 
 import google
 from cached_property import cached_property
-from google.api_core.exceptions import NotFound
+from google.api_core.exceptions import NotFound, PermissionDenied
 from google.api_core.gapic_v1.client_info import ClientInfo
 from google.cloud.secretmanager_v1 import SecretManagerServiceClient
 
@@ -92,3 +92,9 @@ def get_secret(self,
                 "GCP API Call Error (NotFound): Secret ID %s not found.", secret_id
             )
             return None
+        except PermissionDenied:
+            self.log.error(
+                """GCP API Call Error (PermissionDenied): No access for Secret ID %s.
+                Did you add 'secretmanager.versions.access' permission?""", secret_id
+            )
+            return None
diff --git a/tests/providers/google/cloud/_internal_client/test_secret_manager_client.py b/tests/providers/google/cloud/_internal_client/test_secret_manager_client.py
@@ -18,7 +18,7 @@
 
 from unittest import TestCase, mock
 
-from google.api_core.exceptions import NotFound
+from google.api_core.exceptions import NotFound, PermissionDenied
 from google.cloud.secretmanager_v1.types import AccessSecretVersionResponse
 
 from airflow.providers.google.cloud._internal_client.secret_manager_client import _SecretManagerClient  # noqa
@@ -61,6 +61,21 @@ def test_get_non_existing_key(self, mock_client_info, mock_secrets_client):
         self.assertIsNone(secret)
         mock_client.access_secret_version.assert_called_once_with('full-path')
 
+    @mock.patch(INTERNAL_CLIENT_MODULE + ".SecretManagerServiceClient")
+    @mock.patch(INTERNAL_CLIENT_MODULE + ".ClientInfo")
+    def test_get_no_permissions(self, mock_client_info, mock_secrets_client):
+        mock_client = mock.MagicMock()
+        mock_client_info.return_value = mock.MagicMock()
+        mock_secrets_client.return_value = mock_client
+        mock_client.secret_version_path.return_value = "full-path"
+        # No permissions for requested secret id
+        mock_client.access_secret_version.side_effect = PermissionDenied('test-msg')
+        secrets_client = _SecretManagerClient(credentials="credentials")
+        secret = secrets_client.get_secret(secret_id="missing", project_id="project_id")
+        mock_client.secret_version_path.assert_called_once_with("project_id", 'missing', 'latest')
+        self.assertIsNone(secret)
+        mock_client.access_secret_version.assert_called_once_with('full-path')
+
     @mock.patch(INTERNAL_CLIENT_MODULE + ".SecretManagerServiceClient")
     @mock.patch(INTERNAL_CLIENT_MODULE + ".ClientInfo")
     def test_get_existing_key(self, mock_client_info, mock_secrets_client):