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 GSM.
166
167        Returns 'None' if the secret is not found.
168        """
169        try:
170            return SecretString(
171                self.secret_client.access_secret_version(
172                    name=self._fully_qualified_secret_name(secret_name)
173                ).payload.data.decode("UTF-8")
174            )
175        except Exception:
176            return None
177
178    def get_secret_handle(
179        self,
180        secret_name: str,
181    ) -> GSMSecretHandle:
182        """Fetch secret in the secret manager, using the secret name.
183
184        Unlike `get_secret`, this method returns a `GSMSecretHandle` object, which can be used to
185        inspect the secret's labels and other metadata.
186
187        Args:
188            secret_name (str): The name of the connector to filter by.
189
190        Returns:
191            GSMSecretHandle: A handle for the matching secret.
192        """
193        return GSMSecretHandle(
194            parent=self,
195            secret_name=self._fully_qualified_secret_name(secret_name),
196        )
197
198    def fetch_secrets(
199        self,
200        *,
201        filter_string: str,
202    ) -> Iterable[GSMSecretHandle]:
203        """List all available secrets in the secret manager.
204
205        Example filter strings:
206        - `labels.connector=source-bigquery`: Filter for secrets with the labe 'source-bigquery'.
207
208        Args:
209            filter_string (str): A filter string to apply to the list of secrets, following the
210                format described in the Google Secret Manager documentation:
211                https://cloud.google.com/secret-manager/docs/filtering
212
213        Returns:
214            Iterable[GSMSecretHandle]: An iterable of `GSMSecretHandle` objects for the matching
215            secrets.
216        """
217        gsm_secrets: ListSecretsPager = self.secret_client.list_secrets(
218            request=secretmanager.ListSecretsRequest(
219                filter=filter_string,
220                parent=f"projects/{self.project}",
221            ),
222        )
223
224        return [
225            GSMSecretHandle(
226                parent=self,
227                secret_name=secret.name,
228            )
229            for secret in gsm_secrets
230        ]
231
232    def fetch_secrets_by_label(
233        self,
234        label_key: str,
235        label_value: str,
236    ) -> Iterable[GSMSecretHandle]:
237        """List all available secrets in the secret manager.
238
239        Args:
240            label_key (str): The key of the label to filter by.
241            label_value (str): The value of the label to filter by.
242
243        Returns:
244            Iterable[GSMSecretHandle]: An iterable of `GSMSecretHandle` objects for the matching
245            secrets.
246        """
247        return self.fetch_secrets(filter_string=f"labels.{label_key}={label_value}")
248
249    def fetch_connector_secrets(
250        self,
251        connector_name: str,
252    ) -> Iterable[GSMSecretHandle]:
253        """Fetch secrets in the secret manager, using the connector name as a filter for the label.
254
255        The label key used to filter the secrets is defined by the `CONNECTOR_LABEL` attribute,
256        which defaults to 'connector'.
257
258        Args:
259            connector_name (str): The name of the connector to filter by.
260
261        Returns:
262            Iterable[GSMSecretHandle]: An iterable of `GSMSecretHandle` objects for the matching
263            secrets.
264        """
265        return self.fetch_secrets_by_label(
266            label_key=self.CONNECTOR_LABEL,
267            label_value=connector_name,
268        )
269
270    def fetch_connector_secret(
271        self,
272        connector_name: str,
273    ) -> GSMSecretHandle:
274        """Fetch secret in the secret manager, using the connector name as a filter for the label.
275
276        This method is a convenience method that returns the first secret found for the connector.
277
278        The label key used to filter the secrets is defined by the `CONNECTOR_LABEL` attribute,
279        which defaults to 'connector'.
280
281        Args:
282            connector_name (str): The name of the connector to filter by.
283
284        Returns:
285            GSMSecretHandle: A handle for the matching secret.
286        """
287        results: Iterable[GSMSecretHandle] = self.fetch_connector_secrets(connector_name)
288        try:
289            result = next(iter(results))
290        except StopIteration:
291            raise exc.PyAirbyteError(
292                message="No secrets found for connector.",
293                guidance=(
294                    "Please check that the connector name is correct "
295                    "and that the secret is correctly labeled."
296                ),
297                context={
298                    "project": self.project,
299                    "connector_name": connector_name,
300                    "label_key": self.CONNECTOR_LABEL,
301                },
302            ) from None
303
304        return result