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
17use crate::{Error, Result};
18use graph_r::fsio::now_ms;
19use std::future::Future;
20use std::sync::Mutex;
21use std::time::Duration;
22
23const METADATA_TOKEN: &str =
24    "http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/token";
25const TOKEN_URI: &str = "https://oauth2.googleapis.com/token";
26/// Refresh this long before expiry.
27const SKEW_MS: u64 = 60_000;
28
29/// A bearer token.
30#[derive(Clone)]
31pub struct AccessToken {
32    secret: String,
33    /// Expiry, Unix milliseconds (`u64::MAX`: never).
34    pub expires_at_ms: u64,
35}
36
37impl AccessToken {
38    /// A token that never expires (tests, emulators, externally managed tokens).
39    #[must_use]
40    pub fn new(secret: impl Into<String>, expires_at_ms: u64) -> Self {
41        Self { secret: secret.into(), expires_at_ms }
42    }
43
44    /// The token text, for an `Authorization: Bearer` header.
45    #[must_use]
46    pub fn secret(&self) -> &str {
47        &self.secret
48    }
49
50    fn fresh(&self, now: u64) -> bool {
51        self.expires_at_ms > now.saturating_add(SKEW_MS)
52    }
53}
54
55impl std::fmt::Debug for AccessToken {
56    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
57        f.debug_struct("AccessToken")
58            .field("secret", &"<redacted>")
59            .field("expires_at_ms", &self.expires_at_ms)
60            .finish()
61    }
62}
63
64/// A source of access tokens.
65pub trait TokenProvider {
66    /// Future of [`token`](TokenProvider::token).
67    type Token<'a>: Future<Output = Result<AccessToken>> + 'a
68    where
69        Self: 'a;
70
71    /// A token valid for at least a minute (an empty secret sends no header).
72    fn token(&self) -> Self::Token<'_>;
73
74    /// The project billed for API quota (user credentials only).
75    fn quota_project(&self) -> Option<&str> {
76        None
77    }
78}
79
80/// A fixed token.
81#[derive(Debug, Clone)]
82pub struct Static(AccessToken);
83
84impl Static {
85    /// A fixed bearer token.
86    #[must_use]
87    pub fn new(secret: impl Into<String>) -> Self {
88        Self(AccessToken::new(secret, u64::MAX))
89    }
90
91    /// No token at all (emulators such as fake-gcs-server).
92    #[must_use]
93    pub fn none() -> Self {
94        Self::new(String::new())
95    }
96}
97
98impl TokenProvider for Static {
99    type Token<'a> = std::future::Ready<Result<AccessToken>>;
100    fn token(&self) -> Self::Token<'_> {
101        std::future::ready(Ok(self.0.clone()))
102    }
103}
104
105#[derive(Debug, Default)]
106struct Cache(Mutex<Option<AccessToken>>);
107
108impl Cache {
109    fn get(&self, now: u64) -> Option<AccessToken> {
110        self.0
111            .lock()
112            .unwrap_or_else(std::sync::PoisonError::into_inner)
113            .as_ref()
114            .filter(|t| t.fresh(now))
115            .cloned()
116    }
117    fn put(&self, t: &AccessToken) {
118        *self.0.lock().unwrap_or_else(std::sync::PoisonError::into_inner) = Some(t.clone());
119    }
120}
121
122#[derive(serde::Deserialize)]
123struct TokenResponse {
124    access_token: String,
125    expires_in: Option<u64>,
126}
127
128fn http_client() -> Result<reqwest::Client> {
129    reqwest::Client::builder()
130        .timeout(Duration::from_secs(10))
131        .redirect(reqwest::redirect::Policy::none())
132        .build()
133        .map_err(|e| Error::unavailable(format!("http client: {e}"), false))
134}
135
136async fn parse_token(resp: reqwest::Response) -> Result<AccessToken> {
137    let status = resp.status();
138    if !status.is_success() {
139        return Err(Error::Auth { message: format!("token endpoint answered {status}") });
140    }
141    let body =
142        resp.bytes().await.map_err(|_| Error::unavailable("token response truncated", true))?;
143    let t: TokenResponse = serde_json::from_slice(&body)
144        .map_err(|_| Error::Auth { message: "malformed token response".into() })?;
145    Ok(AccessToken::new(
146        t.access_token,
147        now_ms().saturating_add(t.expires_in.unwrap_or(300).saturating_mul(1000)),
148    ))
149}
150
151/// The metadata server (Cloud Run, GCE, GKE workload identity).
152#[derive(Debug)]
153pub struct Metadata {
154    http: reqwest::Client,
155    url: String,
156    cache: Cache,
157}
158
159impl Metadata {
160    /// The default service account of the instance.
161    pub fn new() -> Result<Self> {
162        Self::with_url(METADATA_TOKEN)
163    }
164
165    /// Another token URL (tests).
166    pub fn with_url(url: &str) -> Result<Self> {
167        Ok(Self { http: http_client()?, url: url.to_owned(), cache: Cache::default() })
168    }
169}
170
171impl TokenProvider for Metadata {
172    type Token<'a> = impl Future<Output = Result<AccessToken>> + 'a;
173    fn token(&self) -> Self::Token<'_> {
174        async move {
175            if let Some(t) = self.cache.get(now_ms()) {
176                return Ok(t);
177            }
178            let resp = self
179                .http
180                .get(&self.url)
181                .header("Metadata-Flavor", "Google")
182                .send()
183                .await
184                .map_err(|_| Error::unavailable("metadata server unreachable", true))?;
185            let t = parse_token(resp).await?;
186            self.cache.put(&t);
187            Ok(t)
188        }
189    }
190}
191
192/// User credentials from `gcloud auth application-default login`.
193pub struct GcloudUser {
194    http: reqwest::Client,
195    client_id: String,
196    client_secret: String,
197    refresh_token: String,
198    quota_project: Option<String>,
199    token_uri: String,
200    cache: Cache,
201}
202
203impl std::fmt::Debug for GcloudUser {
204    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
205        f.debug_struct("GcloudUser")
206            .field("quota_project", &self.quota_project)
207            .finish_non_exhaustive()
208    }
209}
210
211#[derive(serde::Deserialize)]
212struct UserFile {
213    #[serde(rename = "type")]
214    kind: String,
215    client_id: Option<String>,
216    client_secret: Option<String>,
217    refresh_token: Option<String>,
218    quota_project_id: Option<String>,
219}
220
221impl GcloudUser {
222    /// Parse an `authorized_user` credentials file.
223    pub fn from_json(json: &[u8]) -> Result<Self> {
224        let f: UserFile = serde_json::from_slice(json)
225            .map_err(|_| Error::Auth { message: "unreadable credentials file".into() })?;
226        if f.kind != "authorized_user" {
227            return Err(Error::Auth {
228                message: format!("credentials of type {:?} are not supported", f.kind),
229            });
230        }
231        let (Some(client_id), Some(client_secret), Some(refresh_token)) =
232            (f.client_id, f.client_secret, f.refresh_token)
233        else {
234            return Err(Error::Auth { message: "incomplete user credentials".into() });
235        };
236        Ok(Self {
237            http: http_client()?,
238            client_id,
239            client_secret,
240            refresh_token,
241            quota_project: f.quota_project_id,
242            token_uri: TOKEN_URI.to_owned(),
243            cache: Cache::default(),
244        })
245    }
246
247    /// Another token endpoint (tests).
248    #[must_use]
249    pub fn with_token_uri(mut self, uri: &str) -> Self {
250        uri.clone_into(&mut self.token_uri);
251        self
252    }
253}
254
255impl TokenProvider for GcloudUser {
256    type Token<'a> = impl Future<Output = Result<AccessToken>> + 'a;
257    fn token(&self) -> Self::Token<'_> {
258        async move {
259            if let Some(t) = self.cache.get(now_ms()) {
260                return Ok(t);
261            }
262            let form = [
263                ("grant_type", "refresh_token"),
264                ("client_id", self.client_id.as_str()),
265                ("client_secret", self.client_secret.as_str()),
266                ("refresh_token", self.refresh_token.as_str()),
267            ];
268            let body = form
269                .iter()
270                .map(|(k, v)| format!("{k}={}", form_encode(v)))
271                .collect::<Vec<_>>()
272                .join("&");
273            let resp = self
274                .http
275                .post(&self.token_uri)
276                .header("content-type", "application/x-www-form-urlencoded")
277                .body(body)
278                .send()
279                .await
280                .map_err(|_| Error::unavailable("token endpoint unreachable", true))?;
281            let t = parse_token(resp).await?;
282            self.cache.put(&t);
283            Ok(t)
284        }
285    }
286
287    fn quota_project(&self) -> Option<&str> {
288        self.quota_project.as_deref()
289    }
290}
291
292fn form_encode(s: &str) -> String {
293    let mut out = String::with_capacity(s.len());
294    crate::key::encode_component(s, &mut out);
295    out
296}
297
298/// Application Default Credentials (no private-key signing).
299#[derive(Debug)]
300pub enum Adc {
301    /// The instance metadata server.
302    Metadata(Metadata),
303    /// gcloud user credentials.
304    User(GcloudUser),
305    /// A fixed token.
306    Static(Static),
307}
308
309impl Adc {
310    /// `GOOGLE_APPLICATION_CREDENTIALS` (user credentials), then gcloud's
311    /// well-known file, then the metadata server.
312    pub fn discover() -> Result<Self> {
313        if let Some(path) = std::env::var_os("GOOGLE_APPLICATION_CREDENTIALS") {
314            let json = std::fs::read(path)
315                .map_err(|_| Error::Auth { message: "unreadable credentials file".into() })?;
316            return GcloudUser::from_json(&json).map(Adc::User);
317        }
318        if let Some(dir) = gcloud_config_dir(|name: &str| std::env::var_os(name), cfg!(windows)) {
319            if let Ok(json) = std::fs::read(dir.join("application_default_credentials.json")) {
320                return GcloudUser::from_json(&json).map(Adc::User);
321            }
322        }
323        Metadata::new().map(Adc::Metadata)
324    }
325}
326
327/// gcloud's configuration directory, where `gcloud auth
328/// application-default login` writes the well-known credentials file:
329/// `CLOUDSDK_CONFIG` when set, else `%APPDATA%\gcloud` on Windows and
330/// `$HOME/.config/gcloud` elsewhere.
331///
332/// Windows never consults `HOME`: OpenSSH, Git Bash and MSYS set it to the
333/// profile directory, and `%HOME%\gcloud` is not where gcloud writes, so a
334/// HOME-first lookup missed the file and fell through to the metadata server --
335/// on a GCE VM that silently authenticated as the VM's own service account.
336/// `var` is the environment, passed in so the rule is tested without touching
337/// the process's own.
338fn gcloud_config_dir<F>(var: F, windows: bool) -> Option<std::path::PathBuf>
339where
340    F: Fn(&str) -> Option<std::ffi::OsString>,
341{
342    let set = |name: &str| var(name).filter(|v| !v.is_empty());
343    if let Some(dir) = set("CLOUDSDK_CONFIG") {
344        return Some(std::path::PathBuf::from(dir));
345    }
346    if windows {
347        return set("APPDATA").map(|a| std::path::PathBuf::from(a).join("gcloud"));
348    }
349    set("HOME").map(|h| std::path::PathBuf::from(h).join(".config").join("gcloud"))
350}
351
352impl TokenProvider for Adc {
353    type Token<'a> = impl Future<Output = Result<AccessToken>> + 'a;
354    fn token(&self) -> Self::Token<'_> {
355        async move {
356            match self {
357                Adc::Metadata(m) => m.token().await,
358                Adc::User(u) => u.token().await,
359                Adc::Static(s) => s.token().await,
360            }
361        }
362    }
363
364    fn quota_project(&self) -> Option<&str> {
365        match self {
366            Adc::User(u) => u.quota_project(),
367            _ => None,
368        }
369    }
370}
371
372#[cfg(test)]
373mod tests {
374    use super::*;
375
376    fn env<'a>(
377        pairs: &'a [(&'a str, &'a str)],
378    ) -> impl Fn(&str) -> Option<std::ffi::OsString> + 'a {
379        move |name| pairs.iter().find(|(k, _)| *k == name).map(|(_, v)| std::ffi::OsString::from(v))
380    }
381
382    #[test]
383    fn windows_reads_appdata_even_when_home_is_set() {
384        let e = [("HOME", r"C:\Users\a"), ("APPDATA", r"C:\Users\a\AppData\Roaming")];
385        assert_eq!(
386            gcloud_config_dir(env(&e), true),
387            Some(std::path::PathBuf::from(r"C:\Users\a\AppData\Roaming").join("gcloud"))
388        );
389    }
390
391    #[test]
392    fn posix_reads_home_dot_config() {
393        let e = [("HOME", "/home/a"), ("APPDATA", "/ignored")];
394        assert_eq!(
395            gcloud_config_dir(env(&e), false),
396            Some(std::path::PathBuf::from("/home/a").join(".config").join("gcloud"))
397        );
398    }
399
400    #[test]
401    fn cloudsdk_config_wins_on_every_os() {
402        let e = [("CLOUDSDK_CONFIG", "/custom"), ("HOME", "/home/a"), ("APPDATA", "/appdata")];
403        for windows in [true, false] {
404            assert_eq!(
405                gcloud_config_dir(env(&e), windows),
406                Some(std::path::PathBuf::from("/custom"))
407            );
408        }
409    }
410
411    #[test]
412    fn empty_values_count_as_unset() {
413        let e = [("CLOUDSDK_CONFIG", ""), ("APPDATA", "")];
414        assert_eq!(gcloud_config_dir(env(&e), true), None);
415        assert_eq!(gcloud_config_dir(env(&[("HOME", "")]), false), None);
416    }
417
418    #[test]
419    fn tokens_are_never_printed() {
420        let t = AccessToken::new("ya29.secret", 1);
421        assert!(!format!("{t:?}").contains("ya29"));
422        let u = GcloudUser::from_json(
423            br#"{"type":"authorized_user","client_id":"id","client_secret":"cs","refresh_token":"rt","quota_project_id":"p"}"#,
424        )
425        .unwrap();
426        let dbg = format!("{u:?}");
427        assert!(!dbg.contains("cs") && !dbg.contains("rt"), "{dbg}");
428        assert_eq!(u.quota_project(), Some("p"));
429        assert!(GcloudUser::from_json(br#"{"type":"service_account"}"#).is_err());
430    }
431
432    #[test]
433    fn freshness_has_skew() {
434        let now = 1_000_000;
435        assert!(!AccessToken::new("x", now + 30_000).fresh(now));
436        assert!(AccessToken::new("x", now + 120_000).fresh(now));
437    }
438}
objstore_r/token.rs

objstore_r/
token.rs
