Adds GCP Secret Manager Hook (#9368)

* Adds GCP Secret Manager Hook
apache · Jun 18, 2020 · 4e09c64 · 4e09c64
1 parent 880b65a
commit 4e09c64
Show file tree

Hide file tree

Showing 17 changed files with 597 additions and 81 deletions.
diff --git a/airflow/contrib/secrets/gcp_secrets_manager.py b/airflow/contrib/secrets/gcp_secrets_manager.py
@@ -16,15 +16,30 @@
 # specific language governing permissions and limitations
 # under the License.
 
-"""This module is deprecated. Please use `airflow.providers.google.cloud.secrets.secrets_manager`."""
+"""This module is deprecated. Please use `airflow.providers.google.cloud.secrets.secret_manager`."""
 
 import warnings
 
 # pylint: disable=unused-import
-from airflow.providers.google.cloud.secrets.secrets_manager import CloudSecretsManagerBackend # noqa
+from airflow.providers.google.cloud.secrets.secret_manager import CloudSecretManagerBackend
 
 warnings.warn(
- "This module is deprecated. Please use `airflow.providers.google.cloud.secrets.secrets_manager`.",
+ "This module is deprecated. Please use `airflow.providers.google.cloud.secrets.secret_manager`.",
  DeprecationWarning,
  stacklevel=2,
 )
+
+
+class CloudSecretsManagerBackend(CloudSecretManagerBackend):
+ """
+ This class is deprecated.
+ Please use `airflow.providers.google.cloud.secrets.secret_manager.CloudSecretManagerBackend`.
+ """
+
+ def __init__(self, *args, **kwargs):
+ warnings.warn(
+ """This class is deprecated.
+ Please use `airflow.providers.google.cloud.secrets.secret_manager.CloudSecretManagerBackend`.""",
+ DeprecationWarning, stacklevel=2
+ )
+ super().__init__(*args, **kwargs)
diff --git a/airflow/providers/google/cloud/_internal_client/__init__.py b/airflow/providers/google/cloud/_internal_client/__init__.py
@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
diff --git a/airflow/providers/google/cloud/_internal_client/secret_manager_client.py b/airflow/providers/google/cloud/_internal_client/secret_manager_client.py
@@ -0,0 +1,95 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import re
+from typing import Optional
+
+import google
+from cached_property import cached_property
+from google.api_core.exceptions import NotFound
+from google.api_core.gapic_v1.client_info import ClientInfo
+from google.cloud.secretmanager_v1 import SecretManagerServiceClient
+
+from airflow.utils.log.logging_mixin import LoggingMixin
+from airflow.version import version
+
+SECRET_ID_PATTERN = r"^[a-zA-Z0-9-_]*$"
+
+
+class _SecretManagerClient(LoggingMixin):
+
+ """
+ Retrieves Secrets object from GCP Secrets Manager. This is a common class reused between SecretsManager
+ and Secrets Hook that provides the shared authentication and verification mechanisms. This class should
+ not be used directly, use SecretsManager or SecretsHook instead
+
+
+ :param credentials: Credentials used to authenticate to GCP
+ :type credentials: google.auth.credentials.Credentials
+ """
+ def __init__(
+ self,
+ credentials: google.auth.credentials.Credentials,
+ ):
+ super().__init__()
+ self.credentials = credentials
+
+ @staticmethod
+ def is_valid_secret_name(secret_name: str) -> bool:
+ """
+ Returns true if the secret name is valid.
+ :param secret_name: name of the secret
+ :type secret_name: str
+ :return:
+ """
+ return bool(re.match(SECRET_ID_PATTERN, secret_name))
+
+ @cached_property
+ def client(self) -> SecretManagerServiceClient:
+ """
+ Create an authenticated KMS client
+ """
+ _client = SecretManagerServiceClient(
+ credentials=self.credentials,
+ client_info=ClientInfo(client_library_version='airflow_v' + version)
+ )
+ return _client
+
+ def get_secret(self,
+ secret_id: str,
+ project_id: str,
+ secret_version: str = 'latest') -> Optional[str]:
+ """
+ Get secret value from the Secret Manager.
+
+ :param secret_id: Secret Key
+ :type secret_id: str
+ :param project_id: Project id to use
+ :type project_id: str
+ :param secret_version: version of the secret (default is 'latest')
+ :type secret_version: str
+ """
+ name = self.client.secret_version_path(project_id, secret_id, secret_version)
+ try:
+ response = self.client.access_secret_version(name)
+ value = response.payload.data.decode('UTF-8')
+ return value
+ except NotFound:
+ self.log.error(
+ "GCP API Call Error (NotFound): Secret ID %s not found.", secret_id
+ )
+ return None
diff --git a/airflow/providers/google/cloud/hooks/secret_manager.py b/airflow/providers/google/cloud/hooks/secret_manager.py
@@ -0,0 +1,74 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Hook for Secrets Manager service"""
+from typing import Optional
+
+from airflow.providers.google.cloud._internal_client.secret_manager_client import _SecretManagerClient # noqa
+from airflow.providers.google.common.hooks.base_google import GoogleBaseHook
+
+
+# noinspection PyAbstractClass
+class SecretsManagerHook(GoogleBaseHook):
+ """
+ Hook for the Google Secret Manager API.
+
+ See https://cloud.google.com/secret-manager
+
+ All the methods in the hook where project_id is used must be called with
+ keyword arguments rather than positional.
+
+ :param gcp_conn_id: The connection ID to use when fetching connection info.
+ :type gcp_conn_id: str
+ :param delegate_to: The account to impersonate, if any.
+ For this to work, the service account making the request must have
+ domain-wide delegation enabled.
+ :type delegate_to: str
+ """
+ def __init__(
+ self,
+ gcp_conn_id: str = "google_cloud_default",
+ delegate_to: Optional[str] = None
+ ) -> None:
+ super().__init__(gcp_conn_id, delegate_to)
+ self.client = _SecretManagerClient(credentials=self._get_credentials())
+
+ def get_conn(self) -> _SecretManagerClient:
+ """
+ Retrieves the connection to Secret Manager.
+
+ :return: Secret Manager client.
+ :rtype: airflow.providers.google.cloud._internal_client.secret_manager_client._SecretManagerClient
+ """
+ return self.client
+
+ @GoogleBaseHook.fallback_to_default_project_id
+ def get_secret(self, secret_id: str,
+ secret_version: str = 'latest',
+ project_id: Optional[str] = None) -> Optional[str]:
+ """
+ Get secret value from the Secret Manager.
+
+ :param secret_id: Secret Key
+ :type secret_id: str
+ :param secret_version: version of the secret (default is 'latest')
+ :type secret_version: str
+ :param project_id: Project id (if you want to override the project_id from credentials)
+ :type project_id: str
+ """
+ return self.get_conn().get_secret(secret_id=secret_id, secret_version=secret_version,
+ project_id=project_id) # type: ignore
diff --git a/...s/google/cloud/secrets/secrets_manager.py → ...rs/google/cloud/secrets/secret_manager.py b/...s/google/cloud/secrets/secrets_manager.py → ...rs/google/cloud/secrets/secret_manager.py
@@ -18,26 +18,20 @@
 """
 Objects relating to sourcing connections from GCP Secrets Manager
 """
-import re
 from typing import Optional
 
 from cached_property import cached_property
-from google.api_core.exceptions import NotFound
-from google.api_core.gapic_v1.client_info import ClientInfo
-from google.cloud.secretmanager_v1 import SecretManagerServiceClient
 
-from airflow import version
 from airflow.exceptions import AirflowException
-from airflow.providers.google.cloud.utils.credentials_provider import (
- _get_scopes, get_credentials_and_project_id,
-)
+from airflow.providers.google.cloud._internal_client.secret_manager_client import _SecretManagerClient # noqa
+from airflow.providers.google.cloud.utils.credentials_provider import 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):
+class CloudSecretManagerBackend(BaseSecretsBackend, LoggingMixin):
  """
  Retrieves Connection object from GCP Secrets Manager
 
@@ -46,11 +40,11 @@ class CloudSecretsManagerBackend(BaseSecretsBackend, LoggingMixin):
  .. code-block:: ini
 
  [secrets]
- backend = airflow.providers.google.cloud.secrets.secrets_manager.CloudSecretsManagerBackend
+ backend = airflow.providers.google.cloud.secrets.secret_manager.CloudSecretManagerBackend
  backend_kwargs = {"connections_prefix": "airflow-connections", "sep": "-"}
 
  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
+ accessible if you provide ``{"connections_prefix": "airflow-connections", "sep": "-"}`` and request
  conn_id ``smtp_default``.
 
  If the Secrets Manager secret id is ``airflow-variables-hello``, this would be
@@ -63,60 +57,63 @@ class CloudSecretsManagerBackend(BaseSecretsBackend, LoggingMixin):
  :type connections_prefix: str
  :param variables_prefix: Specifies the prefix of the secret to read to get Variables.
  :type variables_prefix: str
- :param gcp_key_path: Path to GCP Credential JSON file;
+ :param gcp_key_path: Path to GCP Credential JSON file. Mutually exclusive with gcp_keyfile_dict.
  use default credentials in the current environment if not provided.
  :type gcp_key_path: str
+ :param gcp_keyfile_dict: Dictionary of keyfile parameters. Mutually exclusive with gcp_key_path.
+ :type gcp_keyfile_dict: dict
  :param gcp_scopes: Comma-separated string containing GCP scopes
  :type gcp_scopes: str
+ :param project_id: Project id (if you want to override the project_id from credentials)
+ :type project_id: str
  :param sep: separator used to concatenate connections_prefix and conn_id. Default: "-"
  :type sep: str
  """
  def __init__(
  self,
  connections_prefix: str = "airflow-connections",
  variables_prefix: str = "airflow-variables",
+ gcp_keyfile_dict: Optional[dict] = None,
  gcp_key_path: Optional[str] = None,
  gcp_scopes: Optional[str] = None,
+ project_id: Optional[str] = None,
  sep: str = "-",
  **kwargs
  ):
  super().__init__(**kwargs)
  self.connections_prefix = connections_prefix
  self.variables_prefix = variables_prefix
- self.gcp_key_path = gcp_key_path
- self.gcp_scopes = gcp_scopes
  self.sep = sep
- self.credentials: Optional[str] = None
- self.project_id: Optional[str] = None
  if not self._is_valid_prefix_and_sep():
  raise AirflowException(
  "`connections_prefix`, `variables_prefix` and `sep` should "
  f"follows that pattern {SECRET_ID_PATTERN}"
  )
-
- def _is_valid_prefix_and_sep(self) -> bool:
- prefix = self.connections_prefix + self.sep
- return bool(re.match(SECRET_ID_PATTERN, prefix))
+ self.credentials, self.project_id = get_credentials_and_project_id(
+ keyfile_dict=gcp_keyfile_dict,
+ key_path=gcp_key_path,
+ scopes=gcp_scopes
+ )
+ # In case project id provided
+ if project_id:
+ self.project_id = project_id
 
  @cached_property
- def client(self) -> SecretManagerServiceClient:
+ def client(self) -> _SecretManagerClient:
  """
- Create an authenticated KMS client
+ Cached property returning secret client.
+
+ :return: Secrets client
  """
- scopes = _get_scopes(self.gcp_scopes)
- self.credentials, self.project_id = get_credentials_and_project_id(
- key_path=self.gcp_key_path,
- scopes=scopes
- )
- _client = SecretManagerServiceClient(
- credentials=self.credentials,
- client_info=ClientInfo(client_library_version='airflow_v' + version.version)
- )
- return _client
+ return _SecretManagerClient(credentials=self.credentials)
+
+ def _is_valid_prefix_and_sep(self) -> bool:
+ prefix = self.connections_prefix + self.sep
+ return _SecretManagerClient.is_valid_secret_name(prefix)
 
  def get_conn_uri(self, conn_id: str) -> Optional[str]:
  """
- Get secret value from Secrets Manager.
+ Get secret value from the SecretManager.
 
  :param conn_id: connection id
  :type conn_id: str
@@ -134,23 +131,12 @@ def get_variable(self, key: str) -> Optional[str]:
 
  def _get_secret(self, path_prefix: str, secret_id: str) -> Optional[str]:
  """
- Get secret value from Parameter Store.
+ Get secret value from the SecretManager based on prefix.
 
  :param path_prefix: Prefix for the Path to get Secret
  :type path_prefix: str
  :param secret_id: Secret Key
  :type secret_id: str
  """
  secret_id = self.build_path(path_prefix, secret_id, 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)
- try:
- response = self.client.access_secret_version(name)
- value = response.payload.data.decode('UTF-8')
- return value
- except NotFound:
- self.log.error(
- "GCP API Call Error (NotFound): Secret ID %s not found.", secret_id
- )
- return None
+ return self.client.get_secret(secret_id=secret_id, project_id=self.project_id)
diff --git a/docs/conf.py b/docs/conf.py
@@ -203,8 +203,10 @@
  "_api/airflow/providers/cncf/index.rst",
  # Utils for internal use
  '_api/airflow/providers/google/cloud/utils',
- # Internal client for hashicorp
+ # Internal client for Hashicorp Vault
  '_api/airflow/providers/hashicorp/_internal_client',
+ # Internal client for GCP Secret Manager
+ '_api/airflow/providers/google/cloud/_internal_client',
  # Templates or partials
  'autoapi_templates',
  'howto/operator/gcp/_partials',