diff --git a/airflow/providers/google/cloud/secrets/secret_manager.py b/airflow/providers/google/cloud/secrets/secret_manager.py index a40c6bfbe5fb4..699f35812f147 100644 --- 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(