Source code for airflow.contrib.secrets.gcp_secrets_manager

# 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.

"""
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.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




[docs]class CloudSecretsManagerBackend(BaseSecretsBackend, LoggingMixin): """ Retrieves Connection object from GCP Secrets Manager Configurable via ``airflow.cfg`` as follows: .. code-block:: ini [secrets] backend = airflow.contrib.secrets.gcp_secrets_manager.CloudSecretsManagerBackend 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 conn_id ``smtp_default``. If the Secrets Manager secret id is ``airflow-variables-hello``, this would be accessible if you provide ``{"variables_prefix": "airflow-variables", "sep": "-"}`` and request Variable Key ``hello``. 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 :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; use default credentials in the current environment if not provided. :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 variables_prefix="airflow-variables", # type: str gcp_key_path=None, # type: Optional[str] gcp_scopes=None, # type: Optional[str] sep="-", # type: str **kwargs ): super(CloudSecretsManagerBackend, self).__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 = None self.project_id = None if not self._is_valid_prefix_and_sep(): raise AirflowException( "`connections_prefix`, `variables_prefix` and `sep` should follows that pattern {}".format( SECRET_ID_PATTERN) )
[docs] def _is_valid_prefix_and_sep(self): prefix = self.connections_prefix + self.sep return bool(re.match(SECRET_ID_PATTERN, prefix))
@cached_property
[docs] def client(self): # type: (...) -> SecretManagerServiceClient """ Create an authenticated KMS 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
[docs] def get_conn_uri(self, conn_id): """ Get secret value from Secrets Manager. :param conn_id: connection id :type conn_id: str """ return self._get_secret(self.connections_prefix, conn_id)
[docs] def get_variable(self, key): # type: (str) -> Optional[str] """ Get Airflow Variable from Environment Variable :param key: Variable Key :return: Variable Value """ return self._get_secret(self.variables_prefix, key)
[docs] def _get_secret(self, path_prefix, secret_id): # type: (str, str) -> Optional[str] """ Get secret value from Parameter Store. :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

Was this entry helpful?