GCP Secrets Backend Impersonation (#36072)

* Updated doc to include impersonation example * Providing support for `impersonation_chain`
apache · Dec 6, 2023 · ca20f07 · ca20f07
1 parent 4b89c8a
commit ca20f07
Show file tree

Hide file tree

Showing 2 changed files with 32 additions and 1 deletion.
diff --git a/airflow/providers/google/cloud/secrets/secret_manager.py b/airflow/providers/google/cloud/secrets/secret_manager.py
@@ -20,12 +20,16 @@
 import logging
 import re
 import warnings
+from typing import Sequence
 
 from google.auth.exceptions import DefaultCredentialsError
 
 from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
 from airflow.providers.google.cloud._internal_client.secret_manager_client import _SecretManagerClient
-from airflow.providers.google.cloud.utils.credentials_provider import get_credentials_and_project_id
+from airflow.providers.google.cloud.utils.credentials_provider import (
+ _get_target_principal_and_delegates,
+ get_credentials_and_project_id,
+)
 from airflow.secrets import BaseSecretsBackend
 from airflow.utils.log.logging_mixin import LoggingMixin
 
@@ -76,6 +80,14 @@ class CloudSecretManagerBackend(BaseSecretsBackend, LoggingMixin):
  :param project_id: Project ID to read the secrets from. If not passed, the project ID from credentials
  will be used.
  :param sep: Separator used to concatenate connections_prefix and conn_id. Default: "-"
+ :param impersonation_chain: Optional service account to impersonate using
+ short-term credentials, or chained list of accounts required to get the
+ access token of the last account in the list, which will be impersonated
+ in the request. If set as a string, the account must grant the
+ originating account the Service Account Token Creator IAM role. If set
+ as a sequence, the identities from the list must grant Service Account
+ Token Creator IAM role to the directly preceding identity, with first
+ account from the list granting this role to the originating account.
  """
 
  def __init__(
@@ -89,6 +101,7 @@ def __init__(
  gcp_scopes: str | None = None,
  project_id: str | None = None,
  sep: str = "-",
+ impersonation_chain: str | Sequence[str] | None = None,
  **kwargs,
  ) -> None:
  super().__init__(**kwargs)
@@ -103,11 +116,19 @@ def __init__(
  f"follows that pattern {SECRET_ID_PATTERN}"
  )
  try:
+ if impersonation_chain:
+ target_principal, delegates = _get_target_principal_and_delegates(impersonation_chain)
+ else:
+ target_principal = None
+ delegates = None
+
  self.credentials, self.project_id = get_credentials_and_project_id(
  keyfile_dict=gcp_keyfile_dict,
  key_path=gcp_key_path,
  credential_config_file=gcp_credential_config_file,
  scopes=gcp_scopes,
+ target_principal=target_principal,
+ delegates=delegates,
  )
  except (DefaultCredentialsError, FileNotFoundError):
  log.exception(

diff --git a/...rflow-providers-google/secrets-backends/google-cloud-secret-manager-backend.rst b/...rflow-providers-google/secrets-backends/google-cloud-secret-manager-backend.rst
@@ -77,6 +77,7 @@ the following parameters:
 * ``gcp_scopes``: Comma-separated string containing OAuth2 scopes.
 * ``sep``: Separator used to concatenate connections_prefix and conn_id. Default: ``"-"``
 * ``project_id``: Project ID to read the secrets from. If not passed, the project ID from credentials will be used.
+* ``impersonation_chain``: Optional service account to impersonate using short-term credentials, or chained list of accounts required to get the access token of the last account in the list, which will be impersonated in the request.
 
 All options should be passed as a JSON dictionary.
 
@@ -88,6 +89,15 @@ For example, if you want to set parameter ``connections_prefix`` to ``"example-c
  backend = airflow.providers.google.cloud.secrets.secret_manager.CloudSecretManagerBackend
  backend_kwargs = {"connections_prefix": "example-connections-prefix", "variables_prefix": "example-variables-prefix"}
 
+Also, if you are using Application Default Credentials (ADC) to read secrets from ``example-project`` but would like
+to impersonate a different service account, your configuration should look similar to this:
+
+.. code-block:: ini
+
+ [secrets]
+ backend = airflow.providers.google.cloud.secrets.secret_manager.CloudSecretManagerBackend
+ backend_kwargs = {"project_id": "example-project", "impersonation_chain": "impersonated_account@example_project.iam.gserviceaccount.com"}
+
 Set-up credentials
 """"""""""""""""""