gcp_bigquery_client/
auth.rs

Help
1//! Helpers to manage GCP authentication.
2use std::path::Path;
3use std::path::PathBuf;
4use std::sync::Arc;
5
6use async_trait::async_trait;
7use dyn_clone::{clone_trait_object, DynClone};
8use hyper_util::client::legacy::connect::HttpConnector;
9use yup_oauth2::authenticator::ApplicationDefaultCredentialsTypes;
10use yup_oauth2::authenticator::Authenticator as YupAuthenticator;
11use yup_oauth2::authorized_user::AuthorizedUserSecret;
12use yup_oauth2::hyper_rustls::HttpsConnector;
13use yup_oauth2::ApplicationDefaultCredentialsAuthenticator as YupApplicationDefaultCredentialsAuthenticator;
14use yup_oauth2::ApplicationDefaultCredentialsFlowOpts;
15use yup_oauth2::AuthorizedUserAuthenticator as YupAuthorizedUserAuthenticator;
16use yup_oauth2::{ApplicationSecret, ServiceAccountKey};
17use yup_oauth2::{InstalledFlowAuthenticator as YupInstalledFlowAuthenticator, InstalledFlowReturnMethod};
18
19use crate::error::BQError;
20
21#[async_trait]
22pub trait Authenticator: DynClone + Send + Sync {
23    async fn access_token(&self) -> Result<String, BQError>;
24}
25
26clone_trait_object!(Authenticator);
27
28/// A service account authenticator.
29#[derive(Clone)]
30pub struct ServiceAccountAuthenticator {
31    auth: Option<YupAuthenticator<HttpsConnector<HttpConnector>>>,
32    scopes: Vec<String>,
33    is_using_workload_identity: bool,
34}
35
36#[async_trait]
37impl Authenticator for ServiceAccountAuthenticator {
38    /// Returns an access token.
39    async fn access_token(&self) -> Result<String, BQError> {
40        let token = if self.is_using_workload_identity {
41            get_access_token_with_workload_identity().await?.access_token
42        } else {
43            self.auth
44                .clone()
45                .unwrap()
46                .token(self.scopes.as_ref())
47                .await?
48                .token()
49                .ok_or(BQError::NoToken)?
50                .to_string()
51        };
52        Ok(token)
53    }
54}
55
56impl ServiceAccountAuthenticator {
57    pub(crate) async fn from_service_account_key(
58        sa_key: ServiceAccountKey,
59        scopes: &[&str],
60    ) -> Result<Arc<dyn Authenticator>, BQError> {
61        let auth = yup_oauth2::ServiceAccountAuthenticator::builder(sa_key).build().await;
62
63        match auth {
64            Err(err) => Err(BQError::InvalidServiceAccountAuthenticator(err)),
65            Ok(auth) => Ok(Arc::new(ServiceAccountAuthenticator {
66                auth: Some(auth),
67                scopes: scopes.iter().map(|scope| scope.to_string()).collect(),
68                is_using_workload_identity: false,
69            })),
70        }
71    }
72
73    pub(crate) async fn with_workload_identity(scopes: &[&str]) -> Result<Arc<dyn Authenticator>, BQError> {
74        Ok(Arc::new(ServiceAccountAuthenticator {
75            auth: None,
76            scopes: scopes.iter().map(|scope| scope.to_string()).collect(),
77            is_using_workload_identity: true,
78        }))
79    }
80}
81
82pub(crate) async fn service_account_authenticator(
83    scopes: Vec<&str>,
84    sa_key_file: &str,
85) -> Result<Arc<dyn Authenticator>, BQError> {
86    let sa_key = yup_oauth2::read_service_account_key(sa_key_file).await?;
87    ServiceAccountAuthenticator::from_service_account_key(sa_key, &scopes).await
88}
89
90#[derive(Deserialize)]
91pub struct WorkloadIdentityAccessToken {
92    pub access_token: String,
93    pub expires_in: i32,
94    pub token_type: String,
95}
96
97pub(crate) async fn get_access_token_with_workload_identity() -> Result<WorkloadIdentityAccessToken, BQError> {
98    let client = reqwest::Client::new();
99    let resp = client
100        .get("http://metadata/computeMetadata/v1/instance/service-accounts/default/token")
101        .header("Metadata-Flavor", "Google")
102        .send()
103        .await?;
104
105    let content: WorkloadIdentityAccessToken = resp.json().await?;
106
107    Ok(content)
108}
109
110#[derive(Clone)]
111pub struct InstalledFlowAuthenticator {
112    auth: Option<YupAuthenticator<HttpsConnector<HttpConnector>>>,
113    scopes: Vec<String>,
114}
115
116impl InstalledFlowAuthenticator {
117    pub(crate) async fn from_token_file_path<P: Into<PathBuf>>(
118        app_secret: ApplicationSecret,
119        scopes: &[&str],
120        persistant_file_path: P,
121    ) -> Result<Arc<dyn Authenticator>, BQError> {
122        let auth = YupInstalledFlowAuthenticator::builder(app_secret, InstalledFlowReturnMethod::HTTPRedirect)
123            .persist_tokens_to_disk(persistant_file_path)
124            .build()
125            .await;
126
127        match auth {
128            Err(err) => Err(BQError::InvalidInstalledFlowAuthenticator(err)),
129            Ok(auth) => {
130                // For InstalledFlowAuthenticator, we need to call token(), because it is more natural to execute the authorization code flow before returning `InstalledFlowAuthenticator` rather than before the first API call.
131                match auth.token(scopes).await {
132                    Err(token_err) => Err(BQError::YupAuthError(token_err)),
133                    Ok(_) => Ok(Arc::new(InstalledFlowAuthenticator {
134                        auth: Some(auth),
135                        scopes: scopes.iter().map(|scope| scope.to_string()).collect(),
136                    })),
137                }
138            }
139        }
140    }
141}
142
143#[async_trait]
144impl Authenticator for InstalledFlowAuthenticator {
145    async fn access_token(&self) -> Result<String, BQError> {
146        Ok(self
147            .auth
148            .clone()
149            .unwrap()
150            .token(self.scopes.as_ref())
151            .await?
152            .token()
153            .ok_or(BQError::NoToken)?
154            .to_string())
155    }
156}
157
158/// Send a request to Google's OAuth 2.0 server and get an access token.
159/// See [Gooogle OAuth2.0 Documentation](https://developers.google.com/identity/protocols/oauth2/native-app).
160pub(crate) async fn installed_flow_authenticator<S: AsRef<[u8]>, P: Into<PathBuf>>(
161    secret: S,
162    scopes: &[&str],
163    persistant_file_path: P,
164) -> Result<Arc<dyn Authenticator>, BQError> {
165    let app_secret = yup_oauth2::parse_application_secret(secret)?;
166    InstalledFlowAuthenticator::from_token_file_path(app_secret, scopes, persistant_file_path).await
167}
168
169#[derive(Clone)]
170pub struct ApplicationDefaultCredentialsAuthenticator {
171    auth: Option<YupAuthenticator<HttpsConnector<HttpConnector>>>,
172    scopes: Vec<String>,
173}
174
175impl ApplicationDefaultCredentialsAuthenticator {
176    pub(crate) async fn from_scopes(scopes: &[&str]) -> Result<Arc<dyn Authenticator>, BQError> {
177        let opts = ApplicationDefaultCredentialsFlowOpts::default();
178        let auth = match YupApplicationDefaultCredentialsAuthenticator::builder(opts).await {
179            ApplicationDefaultCredentialsTypes::InstanceMetadata(auth) => auth.build().await,
180            ApplicationDefaultCredentialsTypes::ServiceAccount(auth) => auth.build().await,
181        };
182
183        match auth {
184            Err(err) => Err(BQError::InvalidApplicationDefaultCredentialsAuthenticator(err)),
185            Ok(auth) => Ok(Arc::new(ApplicationDefaultCredentialsAuthenticator {
186                auth: Some(auth),
187                scopes: scopes.iter().map(|scope| scope.to_string()).collect(),
188            })),
189        }
190    }
191}
192
193#[async_trait]
194impl Authenticator for ApplicationDefaultCredentialsAuthenticator {
195    async fn access_token(&self) -> Result<String, BQError> {
196        Ok(self
197            .auth
198            .clone()
199            .unwrap()
200            .token(self.scopes.as_ref())
201            .await?
202            .token()
203            .ok_or(BQError::NoToken)?
204            .to_string())
205    }
206}
207
208pub(crate) async fn application_default_credentials_authenticator(
209    scopes: &[&str],
210) -> Result<Arc<dyn Authenticator>, BQError> {
211    ApplicationDefaultCredentialsAuthenticator::from_scopes(scopes).await
212}
213
214#[derive(Clone)]
215pub struct AuthorizedUserAuthenticator {
216    auth: Option<YupAuthenticator<HttpsConnector<HttpConnector>>>,
217    scopes: Vec<String>,
218}
219
220impl AuthorizedUserAuthenticator {
221    pub(crate) async fn from_authorized_user_secret(
222        authorized_user_secret: AuthorizedUserSecret,
223        scopes: &[&str],
224    ) -> Result<Arc<dyn Authenticator>, BQError> {
225        let auth = YupAuthorizedUserAuthenticator::builder(authorized_user_secret)
226            .build()
227            .await;
228
229        match auth {
230            Err(err) => Err(BQError::InvalidAuthorizedUserAuthenticator(err)),
231            Ok(auth) => Ok(Arc::new(AuthorizedUserAuthenticator {
232                auth: Some(auth),
233                scopes: scopes.iter().map(|scope| scope.to_string()).collect(),
234            })),
235        }
236    }
237}
238
239#[async_trait]
240impl Authenticator for AuthorizedUserAuthenticator {
241    async fn access_token(&self) -> Result<String, BQError> {
242        Ok(self
243            .auth
244            .clone()
245            .unwrap()
246            .token(self.scopes.as_ref())
247            .await?
248            .token()
249            .ok_or(BQError::NoToken)?
250            .to_string())
251    }
252}
253
254pub(crate) async fn authorized_user_authenticator<S: AsRef<Path>>(
255    secret: S,
256    scopes: &[&str],
257) -> Result<Arc<dyn Authenticator>, BQError> {
258    let authorized_user_secret = yup_oauth2::read_authorized_user_secret(secret).await?;
259    AuthorizedUserAuthenticator::from_authorized_user_secret(authorized_user_secret, scopes).await
260}
gcp_bigquery_client/auth.rs

gcp_bigquery_client/
auth.rs
