airbyte.secrets.google_gsm

  1# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
  2"""Secret manager that retrieves secrets from Google Secrets Manager (GSM).
  3
  4Usage Example:
  5
  6```python
  7import ab
  8from airbyte import secrets
  9
 10# Initialize the Google GSM secret manager
 11gsm = secrets.GoogleGSMSecretManager(
 12    project="my-project",
 13    credentials_json=ab.get_secret("GCP_GSM_CREDENTIALS"),
 14)
 15# Get secrets with the label 'connector=source-github'
 16secrets = gsm.fetch_connector_secrets("source-github")
 17
 18# Get the first secret and parse the JSON value
 19config: dict = next(secrets, None).parse_json()
 20
 21# Pass the config to your source
 22source = ab.get_source(
 23    "source-github",
 24    config=config,
 25    streams="*",
 26)
 27read_result = source.read()
 28```
 29
 30"""
 31
 32from __future__ import annotations
 33
 34import json
 35import os
 36from pathlib import Path
 37from typing import TYPE_CHECKING
 38
 39from google.cloud import secretmanager_v1 as secretmanager
 40
 41from airbyte import exceptions as exc
 42from airbyte.secrets.base import SecretHandle, SecretSourceEnum, SecretString
 43from airbyte.secrets.custom import CustomSecretManager
 44
 45
 46if TYPE_CHECKING:
 47    from collections.abc import Iterable, MutableMapping
 48
 49    from google.cloud.secretmanager_v1.services.secret_manager_service.pagers import (
 50        ListSecretsPager,
 51    )
 52
 53
 54class GSMSecretHandle(SecretHandle):
 55    """A handle for a secret stored in Google Secrets Manager (GSM).
 56
 57    This class inherits from `SecretHandle` and adds a `labels` attribute for inspecting GSM
 58    labels.
 59    """
 60
 61    parent: GoogleGSMSecretManager
 62
 63    def _get_gsm_secret_object(self) -> secretmanager.Secret:
 64        """Get the `Secret` object from GSM."""
 65        return self.parent.secret_client.get_secret(
 66            name=self.secret_name,
 67        )
 68
 69    @property
 70    def labels(self) -> MutableMapping[str, str]:
 71        """Get the labels of the secret."""
 72        return self._get_gsm_secret_object().labels
 73
 74
 75class GoogleGSMSecretManager(CustomSecretManager):
 76    """Secret manager that retrieves secrets from Google Secrets Manager (GSM).
 77
 78    This class inherits from `CustomSecretManager` and also adds methods
 79    that are specific to this implementation: `fetch_secrets()`,
 80    `fetch_secrets_by_label()` and `fetch_connector_secrets()`.
 81
 82    This secret manager is not enabled by default. To use it, you must provide the project ID and
 83    the credentials for a service account with the necessary permissions to access the secrets.
 84
 85    The `fetch_connector_secret()` method assumes a label name of `connector`
 86    matches the name of the connector (`source-github`, `destination-snowflake`, etc.)
 87    """
 88
 89    name = SecretSourceEnum.GOOGLE_GSM.value
 90    auto_register = False
 91    as_backup = False
 92    replace_existing = False
 93
 94    CONNECTOR_LABEL = "connector"
 95    """The label key used to filter secrets by connector name."""
 96
 97    def __init__(
 98        self,
 99        project: str,
100        *,
101        credentials_path: str | None = None,
102        credentials_json: str | SecretString | None = None,
103        auto_register: bool = False,
104        as_backup: bool = False,
105    ) -> None:
106        """Instantiate a new Google GSM secret manager instance.
107
108        You can provide either the path to the credentials file or the JSON contents of the
109        credentials file. If both are provided, a `PyAirbyteInputError` will be raised.
110        """
111        if credentials_path and credentials_json:
112            raise exc.PyAirbyteInputError(
113                guidance=("You can provide `credentials_path` or `credentials_json` but not both."),
114            )
115
116        self.project = project
117
118        if credentials_json is not None and not isinstance(credentials_json, SecretString):
119            credentials_json = SecretString(credentials_json)
120
121        if not credentials_json and not credentials_path:
122            if "GOOGLE_APPLICATION_CREDENTIALS" in os.environ:
123                credentials_path = os.environ["GOOGLE_APPLICATION_CREDENTIALS"]
124
125            elif "GCP_GSM_CREDENTIALS" in os.environ:
126                credentials_json = SecretString(os.environ["GCP_GSM_CREDENTIALS"])
127
128        if credentials_path:
129            credentials_json = SecretString(Path(credentials_path).read_text(encoding="utf-8"))
130
131        if not credentials_json:
132            raise exc.PyAirbyteInputError(
133                guidance=(
134                    "No Google Cloud credentials found. You can provide the path to the "
135                    "credentials file using the `credentials_path` argument, or provide the JSON "
136                    "contents of the credentials file using the `credentials_json` argument."
137                ),
138            )
139
140        self.secret_client = secretmanager.SecretManagerServiceClient.from_service_account_info(
141            json.loads(credentials_json)
142        )
143
144        if auto_register:
145            self.auto_register = auto_register
146
147        if as_backup:
148            self.as_backup = as_backup
149
150        super().__init__()  # Handles the registration if needed
151
152    def _fully_qualified_secret_name(self, secret_name: str) -> str:
153        """Get the fully qualified secret name."""
154        full_name = secret_name
155        if "projects/" not in full_name:
156            # This is not yet fully qualified
157            full_name = f"projects/{self.project}/secrets/{secret_name}/versions/latest"
158
159        if "/versions/" not in full_name:
160            full_name += "/versions/latest"
161
162        return full_name
163
164    def get_secret(self, secret_name: str) -> SecretString | None:
165        """Get a named secret from Google Colab user secrets."""
166        return SecretString(
167            self.secret_client.access_secret_version(
168                name=self._fully_qualified_secret_name(secret_name)
169            ).payload.data.decode("UTF-8")
170        )
171
172    def get_secret_handle(
173        self,
174        secret_name: str,
175    ) -> GSMSecretHandle:
176        """Fetch secret in the secret manager, using the secret name.
177
178        Unlike `get_secret`, this method returns a `GSMSecretHandle` object, which can be used to
179        inspect the secret's labels and other metadata.
180
181        Args:
182            secret_name (str): The name of the connector to filter by.
183
184        Returns:
185            GSMSecretHandle: A handle for the matching secret.
186        """
187        return GSMSecretHandle(
188            parent=self,
189            secret_name=self._fully_qualified_secret_name(secret_name),
190        )
191
192    def fetch_secrets(
193        self,
194        *,
195        filter_string: str,
196    ) -> Iterable[GSMSecretHandle]:
197        """List all available secrets in the secret manager.
198
199        Example filter strings:
200        - `labels.connector=source-bigquery`: Filter for secrets with the labe 'source-bigquery'.
201
202        Args:
203            filter_string (str): A filter string to apply to the list of secrets, following the
204                format described in the Google Secret Manager documentation:
205                https://cloud.google.com/secret-manager/docs/filtering
206
207        Returns:
208            Iterable[GSMSecretHandle]: An iterable of `GSMSecretHandle` objects for the matching
209            secrets.
210        """
211        gsm_secrets: ListSecretsPager = self.secret_client.list_secrets(
212            request=secretmanager.ListSecretsRequest(
213                filter=filter_string,
214                parent=f"projects/{self.project}",
215            ),
216        )
217
218        return [
219            GSMSecretHandle(
220                parent=self,
221                secret_name=secret.name,
222            )
223            for secret in gsm_secrets
224        ]
225
226    def fetch_secrets_by_label(
227        self,
228        label_key: str,
229        label_value: str,
230    ) -> Iterable[GSMSecretHandle]:
231        """List all available secrets in the secret manager.
232
233        Args:
234            label_key (str): The key of the label to filter by.
235            label_value (str): The value of the label to filter by.
236
237        Returns:
238            Iterable[GSMSecretHandle]: An iterable of `GSMSecretHandle` objects for the matching
239            secrets.
240        """
241        return self.fetch_secrets(filter_string=f"labels.{label_key}={label_value}")
242
243    def fetch_connector_secrets(
244        self,
245        connector_name: str,
246    ) -> Iterable[GSMSecretHandle]:
247        """Fetch secrets in the secret manager, using the connector name as a filter for the label.
248
249        The label key used to filter the secrets is defined by the `CONNECTOR_LABEL` attribute,
250        which defaults to 'connector'.
251
252        Args:
253            connector_name (str): The name of the connector to filter by.
254
255        Returns:
256            Iterable[GSMSecretHandle]: An iterable of `GSMSecretHandle` objects for the matching
257            secrets.
258        """
259        return self.fetch_secrets_by_label(
260            label_key=self.CONNECTOR_LABEL,
261            label_value=connector_name,
262        )
263
264    def fetch_connector_secret(
265        self,
266        connector_name: str,
267    ) -> GSMSecretHandle:
268        """Fetch secret in the secret manager, using the connector name as a filter for the label.
269
270        This method is a convenience method that returns the first secret found for the connector.
271
272        The label key used to filter the secrets is defined by the `CONNECTOR_LABEL` attribute,
273        which defaults to 'connector'.
274
275        Args:
276            connector_name (str): The name of the connector to filter by.
277
278        Returns:
279            GSMSecretHandle: A handle for the matching secret.
280        """
281        results: Iterable[GSMSecretHandle] = self.fetch_connector_secrets(connector_name)
282        try:
283            result = next(iter(results))
284        except StopIteration:
285            raise exc.PyAirbyteError(
286                message="No secrets found for connector.",
287                guidance=(
288                    "Please check that the connector name is correct "
289                    "and that the secret is correctly labeled."
290                ),
291                context={
292                    "project": self.project,
293                    "connector_name": connector_name,
294                    "label_key": self.CONNECTOR_LABEL,
295                },
296            ) from None
297
298        return result