laminar_core/
gcs_credentials.rs

1//! Refreshable GCS workload-identity credentials.
2//!
3//! `object_store` 0.13 supports injecting a credential provider but cannot
4//! parse Google external-account ADC files. This module bridges Google's
5//! official refreshable provider into that existing hook without retaining or
6//! logging access tokens in LaminarDB.
7
8#![allow(clippy::disallowed_types)] // cold path: storage configuration
9
10use std::collections::HashMap;
11use std::fmt;
12use std::hash::BuildHasher;
13use std::io::Read;
14use std::sync::Arc;
15
16use google_cloud_auth::credentials::AccessTokenCredentials;
17use object_store::client::CredentialProvider;
18use object_store::gcp::{
19    GcpCredential, GcpCredentialProvider, GoogleCloudStorageBuilder, GoogleConfigKey,
20};
21use parking_lot::Mutex;
22use serde_json::Value;
23
24const MAX_APPLICATION_CREDENTIAL_BYTES: usize = 1024 * 1024;
25const GCS_READ_WRITE_SCOPE: &str = "https://www.googleapis.com/auth/devstorage.read_write";
26const APPLICATION_CREDENTIAL_KEYS: &[&str] =
27    &["google_application_credentials", "application_credentials"];
28const EXPLICIT_SERVICE_ACCOUNT_KEYS: &[&str] = &[
29    "google_service_account",
30    "service_account",
31    "google_service_account_path",
32    "service_account_path",
33    "google_service_account_key",
34    "service_account_key",
35];
36const AMBIENT_SERVICE_ACCOUNT_KEYS: &[&str] = &[
37    "SERVICE_ACCOUNT",
38    "GOOGLE_SERVICE_ACCOUNT",
39    "GOOGLE_SERVICE_ACCOUNT_PATH",
40    "GOOGLE_SERVICE_ACCOUNT_KEY",
41];
42
43/// A redacted failure while selecting Google application credentials.
44#[derive(Debug, thiserror::Error)]
45pub enum GcsCredentialError {
46    /// The selected credential file could not be read.
47    #[error("the selected GCS application credentials file could not be read")]
48    UnreadableApplicationCredentials,
49    /// The selected credential file exceeds the bounded configuration size.
50    #[error("the selected GCS application credentials file exceeds the 1 MiB limit")]
51    ApplicationCredentialsTooLarge,
52    /// The selected credential file is not valid JSON.
53    #[error("the selected GCS application credentials file is not valid JSON")]
54    InvalidApplicationCredentialsJson,
55    /// A required external-account field was missing or had the wrong type.
56    #[error("the selected GCS external-account credentials have an invalid '{0}' field")]
57    InvalidExternalAccountField(&'static str),
58    /// The selected ADC credential type is not supported by the pinned client.
59    #[error("the selected GCS application credential type is unsupported")]
60    UnsupportedApplicationCredentialType,
61    /// Google's credential provider rejected the external-account configuration.
62    #[error("the selected GCS external-account credentials are invalid")]
63    InvalidExternalAccountConfiguration,
64    /// Google's refreshable provider failed to acquire an access token.
65    #[error("the GCS workload-identity provider could not acquire an access token")]
66    TokenAcquisition,
67    /// Google's refreshable provider returned an unusable access token.
68    #[error("the GCS workload-identity provider returned an empty access token")]
69    EmptyAccessToken,
70}
71
72/// Select a refreshable credential provider when external-account ADC wins the
73/// existing explicit-over-ambient credential precedence.
74///
75/// Service-account key/path configuration remains owned by `object_store`.
76/// The returned provider lazily initializes Google's SDK on first I/O so a
77/// synchronously constructed object store is not tied to a temporary runtime.
78///
79/// # Errors
80///
81/// Returns a redacted error when the selected ADC file cannot be safely read,
82/// parsed, or classified.
83pub fn gcs_workload_identity_provider<F, S>(
84    explicit: &HashMap<String, String, S>,
85    env_lookup: &F,
86) -> Result<Option<GcpCredentialProvider>, GcsCredentialError>
87where
88    F: Fn(&str) -> Option<String>,
89    S: BuildHasher,
90{
91    if has_non_empty_option(explicit, EXPLICIT_SERVICE_ACCOUNT_KEYS) {
92        return Ok(None);
93    }
94    if let Some(path) = non_empty_option(explicit, APPLICATION_CREDENTIAL_KEYS) {
95        return load_workload_identity_provider(path);
96    }
97    if has_non_empty_environment(env_lookup, AMBIENT_SERVICE_ACCOUNT_KEYS) {
98        return Ok(None);
99    }
100    let Some(path) =
101        env_lookup("GOOGLE_APPLICATION_CREDENTIALS").filter(|value| !value.trim().is_empty())
102    else {
103        return Ok(None);
104    };
105    load_workload_identity_provider(&path)
106}
107
108/// Apply GCS options while keeping credential material owned by the injected
109/// workload-identity provider.
110///
111/// Options are applied in iterator order, matching `object_store::parse_url_opts`.
112/// Credential keys are deliberately omitted so ambient static credentials
113/// cannot override an explicitly selected external-account source.
114pub fn configure_gcs_workload_identity<I, K, V>(
115    builder: GoogleCloudStorageBuilder,
116    options: I,
117    credentials: GcpCredentialProvider,
118) -> GoogleCloudStorageBuilder
119where
120    I: IntoIterator<Item = (K, V)>,
121    K: AsRef<str>,
122    V: Into<String>,
123{
124    options
125        .into_iter()
126        .fold(builder, |builder, (key, value)| {
127            let Ok(key) = key.as_ref().to_ascii_lowercase().parse::<GoogleConfigKey>() else {
128                return builder;
129            };
130            if is_credential_key(key) {
131                builder
132            } else {
133                builder.with_config(key, value)
134            }
135        })
136        .with_credentials(credentials)
137}
138
139fn is_credential_key(key: GoogleConfigKey) -> bool {
140    matches!(
141        key,
142        GoogleConfigKey::ServiceAccount
143            | GoogleConfigKey::ServiceAccountKey
144            | GoogleConfigKey::ApplicationCredentials
145    )
146}
147
148fn non_empty_option<'a, S>(
149    options: &'a HashMap<String, String, S>,
150    keys: &[&str],
151) -> Option<&'a str>
152where
153    S: BuildHasher,
154{
155    keys.iter().find_map(|key| {
156        options.iter().find_map(|(candidate, value)| {
157            (candidate.eq_ignore_ascii_case(key) && !value.trim().is_empty())
158                .then_some(value.as_str())
159        })
160    })
161}
162
163fn has_non_empty_option<S>(options: &HashMap<String, String, S>, keys: &[&str]) -> bool
164where
165    S: BuildHasher,
166{
167    non_empty_option(options, keys).is_some()
168}
169
170fn has_non_empty_environment<F>(env_lookup: &F, keys: &[&str]) -> bool
171where
172    F: Fn(&str) -> Option<String>,
173{
174    keys.iter()
175        .any(|key| env_lookup(key).is_some_and(|value| !value.trim().is_empty()))
176}
177
178fn load_workload_identity_provider(
179    path: &str,
180) -> Result<Option<GcpCredentialProvider>, GcsCredentialError> {
181    let file = std::fs::File::open(path)
182        .map_err(|_| GcsCredentialError::UnreadableApplicationCredentials)?;
183    let mut bytes = Vec::new();
184    file.take((MAX_APPLICATION_CREDENTIAL_BYTES + 1) as u64)
185        .read_to_end(&mut bytes)
186        .map_err(|_| GcsCredentialError::UnreadableApplicationCredentials)?;
187    if bytes.len() > MAX_APPLICATION_CREDENTIAL_BYTES {
188        return Err(GcsCredentialError::ApplicationCredentialsTooLarge);
189    }
190    let document: Value = serde_json::from_slice(&bytes)
191        .map_err(|_| GcsCredentialError::InvalidApplicationCredentialsJson)?;
192    let credential_type = required_string(&document, "type")?;
193    match credential_type {
194        "service_account" | "authorized_user" => Ok(None),
195        "external_account" => {
196            validate_external_account_document(&document)?;
197            Ok(Some(Arc::new(WorkloadIdentityCredentialProvider {
198                document: Mutex::new(Some(document)),
199                credentials: tokio::sync::OnceCell::new(),
200            })))
201        }
202        _ => Err(GcsCredentialError::UnsupportedApplicationCredentialType),
203    }
204}
205
206fn validate_external_account_document(document: &Value) -> Result<(), GcsCredentialError> {
207    for field in ["audience", "subject_token_type", "token_url"] {
208        required_string(document, field)?;
209    }
210    if !document
211        .get("credential_source")
212        .is_some_and(Value::is_object)
213    {
214        return Err(GcsCredentialError::InvalidExternalAccountField(
215            "credential_source",
216        ));
217    }
218    Ok(())
219}
220
221fn required_string<'a>(
222    document: &'a Value,
223    field: &'static str,
224) -> Result<&'a str, GcsCredentialError> {
225    document
226        .get(field)
227        .and_then(Value::as_str)
228        .filter(|value| !value.trim().is_empty())
229        .ok_or(GcsCredentialError::InvalidExternalAccountField(field))
230}
231
232struct WorkloadIdentityCredentialProvider {
233    document: Mutex<Option<Value>>,
234    credentials: tokio::sync::OnceCell<AccessTokenCredentials>,
235}
236
237impl fmt::Debug for WorkloadIdentityCredentialProvider {
238    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
239        formatter.write_str("GcsWorkloadIdentityCredentialProvider")
240    }
241}
242
243#[async_trait::async_trait]
244impl CredentialProvider for WorkloadIdentityCredentialProvider {
245    type Credential = GcpCredential;
246
247    async fn get_credential(&self) -> object_store::Result<Arc<GcpCredential>> {
248        let credentials = self
249            .credentials
250            .get_or_try_init(|| async {
251                let document = self.document.lock().take().ok_or_else(|| {
252                    object_store_error(GcsCredentialError::InvalidExternalAccountConfiguration)
253                })?;
254                google_cloud_auth::credentials::external_account::Builder::new(document)
255                    .with_scopes([GCS_READ_WRITE_SCOPE])
256                    .build_access_token_credentials()
257                    .map_err(|_| {
258                        object_store_error(GcsCredentialError::InvalidExternalAccountConfiguration)
259                    })
260            })
261            .await?;
262        let token = credentials
263            .access_token()
264            .await
265            .map_err(|_| object_store_error(GcsCredentialError::TokenAcquisition))?;
266        if token.token.trim().is_empty() {
267            return Err(object_store_error(GcsCredentialError::EmptyAccessToken));
268        }
269        Ok(Arc::new(GcpCredential {
270            bearer: token.token,
271        }))
272    }
273}
274
275fn object_store_error(error: GcsCredentialError) -> object_store::Error {
276    object_store::Error::Generic {
277        store: "GCS authentication",
278        source: Box::new(error),
279    }
280}
281
282#[cfg(test)]
283mod tests {
284    use super::*;
285
286    fn external_account_document(subject_token_file: &str) -> Value {
287        serde_json::json!({
288            "type": "external_account",
289            "audience": "//iam.googleapis.com/projects/1/locations/global/workloadIdentityPools/test/providers/github",
290            "subject_token_type": "urn:ietf:params:oauth:token-type:jwt",
291            "token_url": "https://sts.googleapis.com/v1/token",
292            "credential_source": { "file": subject_token_file }
293        })
294    }
295
296    fn write_credentials(directory: &tempfile::TempDir, name: &str, document: &Value) -> String {
297        let path = directory.path().join(name);
298        std::fs::write(&path, serde_json::to_vec(document).unwrap()).unwrap();
299        path.to_string_lossy().into_owned()
300    }
301
302    #[test]
303    fn explicit_external_account_precedes_ambient_service_account() {
304        let directory = tempfile::tempdir().unwrap();
305        let path = write_credentials(
306            &directory,
307            "external.json",
308            &external_account_document("subject-token"),
309        );
310        let options = HashMap::from([("google_application_credentials".into(), path)]);
311        let provider = gcs_workload_identity_provider(&options, &|name| {
312            (name == "GOOGLE_SERVICE_ACCOUNT_KEY").then(|| "not-selected".into())
313        })
314        .unwrap()
315        .unwrap();
316        assert_eq!(
317            format!("{provider:?}"),
318            "GcsWorkloadIdentityCredentialProvider"
319        );
320    }
321
322    #[test]
323    fn explicit_service_account_retains_downstream_provider() {
324        let options =
325            HashMap::from([("google_service_account_key".into(), "not-inspected".into())]);
326        let selected = gcs_workload_identity_provider(&options, &|name| {
327            (name == "GOOGLE_APPLICATION_CREDENTIALS").then(|| "ambient.json".into())
328        })
329        .unwrap();
330        assert!(selected.is_none());
331    }
332
333    #[test]
334    fn supported_non_external_adc_retains_downstream_provider() {
335        let directory = tempfile::tempdir().unwrap();
336        let path = write_credentials(
337            &directory,
338            "authorized-user.json",
339            &serde_json::json!({ "type": "authorized_user" }),
340        );
341        let options = HashMap::from([("application_credentials".into(), path)]);
342        assert!(gcs_workload_identity_provider(&options, &|_| None)
343            .unwrap()
344            .is_none());
345    }
346
347    #[test]
348    fn errors_and_debug_output_do_not_disclose_credentials() {
349        let directory = tempfile::tempdir().unwrap();
350        let secret = "do-not-disclose";
351        let path = write_credentials(
352            &directory,
353            secret,
354            &serde_json::json!({
355                "type": "external_account",
356                "token_url": format!("https://{secret}.example/token"),
357                "credential_source": { "file": secret }
358            }),
359        );
360        let options = HashMap::from([("google_application_credentials".into(), path)]);
361        let error = gcs_workload_identity_provider(&options, &|_| None)
362            .unwrap_err()
363            .to_string();
364        assert!(error.contains("audience"), "{error}");
365        assert!(!error.contains(secret), "{error}");
366    }
367
368    #[test]
369    fn credential_file_read_is_bounded() {
370        let directory = tempfile::tempdir().unwrap();
371        let path = directory.path().join("oversized.json");
372        std::fs::write(&path, vec![b'x'; MAX_APPLICATION_CREDENTIAL_BYTES + 1]).unwrap();
373        let options = HashMap::from([(
374            "google_application_credentials".into(),
375            path.to_string_lossy().into_owned(),
376        )]);
377
378        let error = gcs_workload_identity_provider(&options, &|_| None).unwrap_err();
379        assert!(matches!(
380            error,
381            GcsCredentialError::ApplicationCredentialsTooLarge
382        ));
383    }
384
385    #[test]
386    fn credential_options_are_not_applied_to_the_object_store_builder() {
387        let directory = tempfile::tempdir().unwrap();
388        let path = write_credentials(
389            &directory,
390            "external.json",
391            &external_account_document("subject-token"),
392        );
393        let options = HashMap::from([("google_application_credentials".into(), path)]);
394        let provider = gcs_workload_identity_provider(&options, &|_| None)
395            .unwrap()
396            .unwrap();
397        let builder = configure_gcs_workload_identity(
398            GoogleCloudStorageBuilder::new().with_url("gs://test-bucket/prefix"),
399            options,
400            provider,
401        );
402        assert!(builder.build().is_ok());
403    }
404
405    #[tokio::test]
406    async fn external_account_provider_acquires_and_reuses_sdk_token() {
407        use tokio::io::{AsyncReadExt, AsyncWriteExt};
408
409        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
410        let token_url = format!("http://{}/token", listener.local_addr().unwrap());
411        let server = tokio::spawn(async move {
412            let (mut stream, _) = listener.accept().await.unwrap();
413            let mut request = [0_u8; 8 * 1024];
414            let length = stream.read(&mut request).await.unwrap();
415            assert!(String::from_utf8_lossy(&request[..length]).starts_with("POST /token "));
416            let body = r#"{"access_token":"test-access-token","issued_token_type":"urn:ietf:params:oauth:token-type:access_token","token_type":"Bearer","expires_in":3600}"#;
417            let response = format!(
418                "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
419                body.len()
420            );
421            stream.write_all(response.as_bytes()).await.unwrap();
422        });
423
424        let directory = tempfile::tempdir().unwrap();
425        let subject_token_path = directory.path().join("subject-token");
426        std::fs::write(&subject_token_path, "test-subject-token").unwrap();
427        let credential_path = write_credentials(
428            &directory,
429            "external.json",
430            &serde_json::json!({
431                "type": "external_account",
432                "audience": "test-audience",
433                "subject_token_type": "urn:ietf:params:oauth:token-type:jwt",
434                "token_url": token_url,
435                "credential_source": { "file": subject_token_path }
436            }),
437        );
438        let options = HashMap::from([("google_application_credentials".into(), credential_path)]);
439        let provider = gcs_workload_identity_provider(&options, &|_| None)
440            .unwrap()
441            .unwrap();
442
443        let first =
444            tokio::time::timeout(std::time::Duration::from_secs(5), provider.get_credential())
445                .await
446                .unwrap()
447                .unwrap();
448        let second = provider.get_credential().await.unwrap();
449        assert_eq!(first.bearer, "test-access-token");
450        assert_eq!(second.bearer, first.bearer);
451        server.await.unwrap();
452    }
453}
laminar_core/gcs_credentials.rs

laminar_core/
gcs_credentials.rs
