objstore_r/
token.rs

1// Author: spicyzhug@gmail.com
2// Copyright: lituus-io, all rights reserved
3//
4//! OAuth access tokens for Google APIs (feature `gcs`).
5//!
6//! [`Adc::discover`] follows Application Default Credentials the way `gcloud`
7//! does for the cases that need no private-key signing: the metadata server on
8//! Cloud Run / GCE / GKE, and user credentials from
9//! `gcloud auth application-default login`. Service-account key files are not
10//! supported on purpose (use workload identity or impersonation instead of
11//! long-lived keys). Applications with their own credential stack implement
12//! [`TokenProvider`] directly.
13//!
14//! Tokens are cached until a minute before they expire and are never printed:
15//! [`AccessToken`]'s `Debug` is redacted.
16//!
17//! A process that must act only as its caller passes the caller's token with
18//! [`Adc::from_access_token`], or discovers with [`MetadataPolicy::Forbid`],
19//! under which the metadata server is never contacted: no user credential
20//! means [`Error::NoCredentials`], not the instance's own identity.
21
22use crate::{Error, Result};
23use graph_r::fsio::now_ms;
24use std::future::Future;
25use std::sync::Mutex;
26use std::time::Duration;
27
28const METADATA_HOST: &str = "metadata.google.internal";
29const METADATA_TOKEN_PATH: &str = "/computeMetadata/v1/instance/service-accounts/default/token";
30const TOKEN_URI: &str = "https://oauth2.googleapis.com/token";
31/// Refresh this long before expiry.
32const SKEW_MS: u64 = 60_000;
33
34/// A bearer token.
35#[derive(Clone)]
36pub struct AccessToken {
37    secret: String,
38    /// Expiry, Unix milliseconds (`u64::MAX`: never).
39    pub expires_at_ms: u64,
40}
41
42impl AccessToken {
43    /// A token that never expires (tests, emulators, externally managed tokens).
44    #[must_use]
45    pub fn new(secret: impl Into<String>, expires_at_ms: u64) -> Self {
46        Self { secret: secret.into(), expires_at_ms }
47    }
48
49    /// The token text, for an `Authorization: Bearer` header.
50    #[must_use]
51    pub fn secret(&self) -> &str {
52        &self.secret
53    }
54
55    fn fresh(&self, now: u64) -> bool {
56        self.expires_at_ms > now.saturating_add(SKEW_MS)
57    }
58}
59
60impl std::fmt::Debug for AccessToken {
61    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
62        f.debug_struct("AccessToken")
63            .field("secret", &"<redacted>")
64            .field("expires_at_ms", &self.expires_at_ms)
65            .finish()
66    }
67}
68
69/// A source of access tokens.
70pub trait TokenProvider {
71    /// Future of [`token`](TokenProvider::token).
72    type Token<'a>: Future<Output = Result<AccessToken>> + 'a
73    where
74        Self: 'a;
75
76    /// A token valid for at least a minute (an empty secret sends no header).
77    fn token(&self) -> Self::Token<'_>;
78
79    /// The project billed for API quota (user credentials only).
80    fn quota_project(&self) -> Option<&str> {
81        None
82    }
83}
84
85/// A fixed token.
86#[derive(Debug, Clone)]
87pub struct Static(AccessToken);
88
89impl Static {
90    /// A fixed bearer token.
91    #[must_use]
92    pub fn new(secret: impl Into<String>) -> Self {
93        Self(AccessToken::new(secret, u64::MAX))
94    }
95
96    /// No token at all (emulators such as fake-gcs-server).
97    #[must_use]
98    pub fn none() -> Self {
99        Self::new(String::new())
100    }
101}
102
103impl TokenProvider for Static {
104    type Token<'a> = std::future::Ready<Result<AccessToken>>;
105    fn token(&self) -> Self::Token<'_> {
106        std::future::ready(Ok(self.0.clone()))
107    }
108}
109
110#[derive(Debug, Default)]
111struct Cache(Mutex<Option<AccessToken>>);
112
113impl Cache {
114    fn get(&self, now: u64) -> Option<AccessToken> {
115        self.0
116            .lock()
117            .unwrap_or_else(std::sync::PoisonError::into_inner)
118            .as_ref()
119            .filter(|t| t.fresh(now))
120            .cloned()
121    }
122    fn put(&self, t: &AccessToken) {
123        *self.0.lock().unwrap_or_else(std::sync::PoisonError::into_inner) = Some(t.clone());
124    }
125}
126
127#[derive(serde::Deserialize)]
128struct TokenResponse {
129    access_token: String,
130    expires_in: Option<u64>,
131}
132
133fn http_client() -> Result<reqwest::Client> {
134    reqwest::Client::builder()
135        .timeout(Duration::from_secs(10))
136        .redirect(reqwest::redirect::Policy::none())
137        .build()
138        .map_err(|e| Error::unavailable(format!("http client: {e}"), false))
139}
140
141async fn parse_token(resp: reqwest::Response) -> Result<AccessToken> {
142    let status = resp.status();
143    if !status.is_success() {
144        return Err(Error::Auth { message: format!("token endpoint answered {status}") });
145    }
146    let body =
147        resp.bytes().await.map_err(|_| Error::unavailable("token response truncated", true))?;
148    let t: TokenResponse = serde_json::from_slice(&body)
149        .map_err(|_| Error::Auth { message: "malformed token response".into() })?;
150    Ok(AccessToken::new(
151        t.access_token,
152        now_ms().saturating_add(t.expires_in.unwrap_or(300).saturating_mul(1000)),
153    ))
154}
155
156/// The metadata server (Cloud Run, GCE, GKE workload identity).
157#[derive(Debug)]
158pub struct Metadata {
159    http: reqwest::Client,
160    url: String,
161    cache: Cache,
162}
163
164impl Metadata {
165    /// The default service account of the instance, at
166    /// `metadata.google.internal` or the `host[:port]` in `GCE_METADATA_HOST`
167    /// (as Google's client libraries honour it).
168    pub fn new() -> Result<Self> {
169        let host = std::env::var("GCE_METADATA_HOST").ok().filter(|h| !h.is_empty());
170        Self::with_url(&format!(
171            "http://{}{METADATA_TOKEN_PATH}",
172            host.as_deref().unwrap_or(METADATA_HOST)
173        ))
174    }
175
176    /// Another token URL (tests).
177    pub fn with_url(url: &str) -> Result<Self> {
178        Ok(Self { http: http_client()?, url: url.to_owned(), cache: Cache::default() })
179    }
180}
181
182impl TokenProvider for Metadata {
183    type Token<'a> = impl Future<Output = Result<AccessToken>> + 'a;
184    fn token(&self) -> Self::Token<'_> {
185        async move {
186            if let Some(t) = self.cache.get(now_ms()) {
187                return Ok(t);
188            }
189            let resp = self
190                .http
191                .get(&self.url)
192                .header("Metadata-Flavor", "Google")
193                .send()
194                .await
195                .map_err(|_| Error::unavailable("metadata server unreachable", true))?;
196            let t = parse_token(resp).await?;
197            self.cache.put(&t);
198            Ok(t)
199        }
200    }
201}
202
203/// User credentials from `gcloud auth application-default login`.
204pub struct GcloudUser {
205    http: reqwest::Client,
206    client_id: String,
207    client_secret: String,
208    refresh_token: String,
209    quota_project: Option<String>,
210    token_uri: String,
211    cache: Cache,
212}
213
214impl std::fmt::Debug for GcloudUser {
215    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
216        f.debug_struct("GcloudUser")
217            .field("quota_project", &self.quota_project)
218            .finish_non_exhaustive()
219    }
220}
221
222#[derive(serde::Deserialize)]
223struct UserFile {
224    #[serde(rename = "type")]
225    kind: String,
226    client_id: Option<String>,
227    client_secret: Option<String>,
228    refresh_token: Option<String>,
229    quota_project_id: Option<String>,
230}
231
232impl GcloudUser {
233    /// Parse an `authorized_user` credentials file.
234    pub fn from_json(json: &[u8]) -> Result<Self> {
235        let f: UserFile = serde_json::from_slice(json)
236            .map_err(|_| Error::Auth { message: "unreadable credentials file".into() })?;
237        if f.kind != "authorized_user" {
238            return Err(Error::Auth {
239                message: format!("credentials of type {:?} are not supported", f.kind),
240            });
241        }
242        let (Some(client_id), Some(client_secret), Some(refresh_token)) =
243            (f.client_id, f.client_secret, f.refresh_token)
244        else {
245            return Err(Error::Auth { message: "incomplete user credentials".into() });
246        };
247        Ok(Self {
248            http: http_client()?,
249            client_id,
250            client_secret,
251            refresh_token,
252            quota_project: f.quota_project_id,
253            token_uri: TOKEN_URI.to_owned(),
254            cache: Cache::default(),
255        })
256    }
257
258    /// Another token endpoint (tests).
259    #[must_use]
260    pub fn with_token_uri(mut self, uri: &str) -> Self {
261        uri.clone_into(&mut self.token_uri);
262        self
263    }
264}
265
266impl TokenProvider for GcloudUser {
267    type Token<'a> = impl Future<Output = Result<AccessToken>> + 'a;
268    fn token(&self) -> Self::Token<'_> {
269        async move {
270            if let Some(t) = self.cache.get(now_ms()) {
271                return Ok(t);
272            }
273            let form = [
274                ("grant_type", "refresh_token"),
275                ("client_id", self.client_id.as_str()),
276                ("client_secret", self.client_secret.as_str()),
277                ("refresh_token", self.refresh_token.as_str()),
278            ];
279            let body = form
280                .iter()
281                .map(|(k, v)| format!("{k}={}", form_encode(v)))
282                .collect::<Vec<_>>()
283                .join("&");
284            let resp = self
285                .http
286                .post(&self.token_uri)
287                .header("content-type", "application/x-www-form-urlencoded")
288                .body(body)
289                .send()
290                .await
291                .map_err(|_| Error::unavailable("token endpoint unreachable", true))?;
292            let t = parse_token(resp).await?;
293            self.cache.put(&t);
294            Ok(t)
295        }
296    }
297
298    fn quota_project(&self) -> Option<&str> {
299        self.quota_project.as_deref()
300    }
301}
302
303fn form_encode(s: &str) -> String {
304    let mut out = String::with_capacity(s.len());
305    crate::key::encode_component(s, &mut out);
306    out
307}
308
309/// Whether credential discovery may fall through to the metadata server.
310#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
311pub enum MetadataPolicy {
312    /// After user credentials, use the instance's own identity (Cloud Run,
313    /// GCE, GKE). The historical behaviour.
314    #[default]
315    Allow,
316    /// Never contact the metadata server: without user credentials,
317    /// discovery is [`Error::NoCredentials`].
318    Forbid,
319}
320
321/// Application Default Credentials (no private-key signing).
322#[derive(Debug)]
323pub enum Adc {
324    /// The instance metadata server.
325    Metadata(Metadata),
326    /// gcloud user credentials.
327    User(GcloudUser),
328    /// A fixed token.
329    Static(Static),
330}
331
332impl Adc {
333    /// `GOOGLE_APPLICATION_CREDENTIALS` (user credentials), then gcloud's
334    /// well-known file, then the metadata server
335    /// ([`discover_with`](Self::discover_with) under [`MetadataPolicy::Allow`]).
336    pub fn discover() -> Result<Self> {
337        Self::discover_with(MetadataPolicy::Allow)
338    }
339
340    /// Discovery as [`discover`](Self::discover), with the metadata server
341    /// allowed or forbidden.
342    pub fn discover_with(metadata: MetadataPolicy) -> Result<Self> {
343        Self::discover_from(
344            |name: &str| std::env::var_os(name),
345            |p: &std::path::Path| std::fs::read(p).ok(),
346            cfg!(windows),
347            metadata,
348        )
349    }
350
351    fn discover_from<F, R>(var: F, read: R, windows: bool, metadata: MetadataPolicy) -> Result<Self>
352    where
353        F: Fn(&str) -> Option<std::ffi::OsString>,
354        R: Fn(&std::path::Path) -> Option<Vec<u8>>,
355    {
356        match user_credentials(var, read, windows)? {
357            Some(user) => Ok(Adc::User(user)),
358            None => match metadata {
359                MetadataPolicy::Allow => Metadata::new().map(Adc::Metadata),
360                MetadataPolicy::Forbid => Err(Error::NoCredentials {
361                    reason: "no user credentials and the metadata server is forbidden",
362                }),
363            },
364        }
365    }
366
367    /// A caller-supplied access token, used as is and never refreshed
368    /// ([`Adc::Static`]). An empty or blank token is
369    /// [`Error::NoCredentials`]: it would send unauthenticated requests.
370    pub fn from_access_token(secret: impl Into<String>) -> Result<Self> {
371        let secret = secret.into();
372        if secret.trim().is_empty() {
373            return Err(Error::NoCredentials { reason: "empty access token" });
374        }
375        Ok(Adc::Static(Static::new(secret)))
376    }
377}
378
379/// User credentials: the file named by `GOOGLE_APPLICATION_CREDENTIALS`, else
380/// gcloud's well-known file; `None` when neither exists. `var` and `read`
381/// are the environment and the file system, passed in for tests.
382fn user_credentials<F, R>(var: F, read: R, windows: bool) -> Result<Option<GcloudUser>>
383where
384    F: Fn(&str) -> Option<std::ffi::OsString>,
385    R: Fn(&std::path::Path) -> Option<Vec<u8>>,
386{
387    if let Some(path) = var("GOOGLE_APPLICATION_CREDENTIALS") {
388        let json = read(std::path::Path::new(&path))
389            .ok_or_else(|| Error::Auth { message: "unreadable credentials file".into() })?;
390        return GcloudUser::from_json(&json).map(Some);
391    }
392    if let Some(dir) = gcloud_config_dir(&var, windows) {
393        if let Some(json) = read(&dir.join("application_default_credentials.json")) {
394            return GcloudUser::from_json(&json).map(Some);
395        }
396    }
397    Ok(None)
398}
399
400/// gcloud's configuration directory, where `gcloud auth
401/// application-default login` writes the well-known credentials file:
402/// `CLOUDSDK_CONFIG` when set, else `%APPDATA%\gcloud` on Windows and
403/// `$HOME/.config/gcloud` elsewhere.
404///
405/// Windows never consults `HOME`: OpenSSH, Git Bash and MSYS set it to the
406/// profile directory, and `%HOME%\gcloud` is not where gcloud writes, so a
407/// HOME-first lookup missed the file and fell through to the metadata server --
408/// on a GCE VM that silently authenticated as the VM's own service account.
409/// `var` is the environment, passed in so the rule is tested without touching
410/// the process's own.
411fn gcloud_config_dir<F>(var: F, windows: bool) -> Option<std::path::PathBuf>
412where
413    F: Fn(&str) -> Option<std::ffi::OsString>,
414{
415    let set = |name: &str| var(name).filter(|v| !v.is_empty());
416    if let Some(dir) = set("CLOUDSDK_CONFIG") {
417        return Some(std::path::PathBuf::from(dir));
418    }
419    if windows {
420        return set("APPDATA").map(|a| std::path::PathBuf::from(a).join("gcloud"));
421    }
422    set("HOME").map(|h| std::path::PathBuf::from(h).join(".config").join("gcloud"))
423}
424
425impl TokenProvider for Adc {
426    type Token<'a> = impl Future<Output = Result<AccessToken>> + 'a;
427    fn token(&self) -> Self::Token<'_> {
428        async move {
429            match self {
430                Adc::Metadata(m) => m.token().await,
431                Adc::User(u) => u.token().await,
432                Adc::Static(s) => s.token().await,
433            }
434        }
435    }
436
437    fn quota_project(&self) -> Option<&str> {
438        match self {
439            Adc::User(u) => u.quota_project(),
440            _ => None,
441        }
442    }
443}
444
445#[cfg(test)]
446mod tests {
447    use super::*;
448
449    fn env<'a>(
450        pairs: &'a [(&'a str, &'a str)],
451    ) -> impl Fn(&str) -> Option<std::ffi::OsString> + 'a {
452        move |name| pairs.iter().find(|(k, _)| *k == name).map(|(_, v)| std::ffi::OsString::from(v))
453    }
454
455    #[test]
456    fn windows_reads_appdata_even_when_home_is_set() {
457        let e = [("HOME", r"C:\Users\a"), ("APPDATA", r"C:\Users\a\AppData\Roaming")];
458        assert_eq!(
459            gcloud_config_dir(env(&e), true),
460            Some(std::path::PathBuf::from(r"C:\Users\a\AppData\Roaming").join("gcloud"))
461        );
462    }
463
464    #[test]
465    fn posix_reads_home_dot_config() {
466        let e = [("HOME", "/home/a"), ("APPDATA", "/ignored")];
467        assert_eq!(
468            gcloud_config_dir(env(&e), false),
469            Some(std::path::PathBuf::from("/home/a").join(".config").join("gcloud"))
470        );
471    }
472
473    #[test]
474    fn cloudsdk_config_wins_on_every_os() {
475        let e = [("CLOUDSDK_CONFIG", "/custom"), ("HOME", "/home/a"), ("APPDATA", "/appdata")];
476        for windows in [true, false] {
477            assert_eq!(
478                gcloud_config_dir(env(&e), windows),
479                Some(std::path::PathBuf::from("/custom"))
480            );
481        }
482    }
483
484    #[test]
485    fn empty_values_count_as_unset() {
486        let e = [("CLOUDSDK_CONFIG", ""), ("APPDATA", "")];
487        assert_eq!(gcloud_config_dir(env(&e), true), None);
488        assert_eq!(gcloud_config_dir(env(&[("HOME", "")]), false), None);
489    }
490
491    #[test]
492    fn tokens_are_never_printed() {
493        let t = AccessToken::new("ya29.secret", 1);
494        assert!(!format!("{t:?}").contains("ya29"));
495        let u = GcloudUser::from_json(
496            br#"{"type":"authorized_user","client_id":"id","client_secret":"cs","refresh_token":"rt","quota_project_id":"p"}"#,
497        )
498        .unwrap();
499        let dbg = format!("{u:?}");
500        assert!(!dbg.contains("cs") && !dbg.contains("rt"), "{dbg}");
501        assert_eq!(u.quota_project(), Some("p"));
502        assert!(GcloudUser::from_json(br#"{"type":"service_account"}"#).is_err());
503    }
504
505    #[test]
506    fn from_access_token_is_static_and_refuses_blank() {
507        let a = Adc::from_access_token("ya29.caller").unwrap();
508        assert!(matches!(&a, Adc::Static(_)));
509        let t = futures::executor::block_on(a.token()).unwrap();
510        assert_eq!((t.secret(), t.expires_at_ms), ("ya29.caller", u64::MAX));
511        assert_eq!(a.quota_project(), None);
512        for blank in ["", "  ", "\n"] {
513            assert!(matches!(
514                Adc::from_access_token(blank),
515                Err(Error::NoCredentials { reason: "empty access token" })
516            ));
517        }
518    }
519
520    #[test]
521    fn static_tokens_are_fixed_and_none_sends_nothing() {
522        let t = futures::executor::block_on(Static::new("abc").token()).unwrap();
523        assert_eq!(t.secret(), "abc");
524        assert!(futures::executor::block_on(Static::none().token()).unwrap().secret().is_empty());
525    }
526
527    #[test]
528    fn user_credentials_come_from_the_env_file_then_the_well_known_file() {
529        let user = br#"{"type":"authorized_user","client_id":"i","client_secret":"s","refresh_token":"r","quota_project_id":"q"}"#;
530        let files = |want: &'static str| {
531            move |p: &std::path::Path| (p == std::path::Path::new(want)).then(|| user.to_vec())
532        };
533        let e = [("GOOGLE_APPLICATION_CREDENTIALS", "/k.json")];
534        let u = user_credentials(env(&e), files("/k.json"), false).unwrap().unwrap();
535        assert_eq!(u.quota_project(), Some("q"));
536        assert!(user_credentials(env(&e), |_: &std::path::Path| None, false).is_err());
537        let wk = "/h/.config/gcloud/application_default_credentials.json";
538        assert!(user_credentials(env(&[("HOME", "/h")]), files(wk), false).unwrap().is_some());
539        assert!(user_credentials(env(&[("HOME", "/h")]), files("/x"), false).unwrap().is_none());
540    }
541
542    #[test]
543    fn forbid_never_builds_the_metadata_provider() {
544        let none = |_: &std::path::Path| None;
545        let e = [("HOME", "/nowhere")];
546        assert!(matches!(
547            Adc::discover_from(env(&e), none, false, MetadataPolicy::Forbid),
548            Err(Error::NoCredentials { .. })
549        ));
550        // The same environment under Allow falls through, as it always has.
551        assert!(matches!(
552            Adc::discover_from(env(&e), none, false, MetadataPolicy::Allow),
553            Ok(Adc::Metadata(_))
554        ));
555        assert_eq!(MetadataPolicy::default(), MetadataPolicy::Allow, "compatible default");
556        // User credentials win under either policy.
557        let user = br#"{"type":"authorized_user","client_id":"i","client_secret":"s","refresh_token":"r"}"#;
558        let e = [("GOOGLE_APPLICATION_CREDENTIALS", "/k.json")];
559        let read = |_: &std::path::Path| Some(user.to_vec());
560        assert!(matches!(
561            Adc::discover_from(env(&e), read, false, MetadataPolicy::Forbid),
562            Ok(Adc::User(_))
563        ));
564    }
565
566    #[test]
567    fn freshness_has_skew() {
568        let now = 1_000_000;
569        assert!(!AccessToken::new("x", now + 30_000).fresh(now));
570        assert!(AccessToken::new("x", now + 120_000).fresh(now));
571    }
572}
objstore_r/token.rs

objstore_r/
token.rs
