haste_document_storage/providers/gcs/
auth.rs

1//! `OAuth2` access tokens for the Google Cloud Storage API.
2
3use std::{
4    fmt::{self, Debug},
5    path::{Path, PathBuf},
6    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
7};
8
9use jsonwebtoken::{Algorithm, EncodingKey, Header};
10use reqwest::Client;
11use serde::{Deserialize, Serialize};
12use tokio::sync::Mutex;
13
14use super::http::{check, parse_json, transport};
15use crate::{error::StorageError, retry::with_retry};
16
17const SCOPE: &str = "https://www.googleapis.com/auth/devstorage.read_write";
18const DEFAULT_TOKEN_URI: &str = "https://oauth2.googleapis.com/token";
19const DEFAULT_METADATA_HOST: &str = "metadata.google.internal";
20const METADATA_TOKEN_PATH: &str = "/computeMetadata/v1/instance/service-accounts/default/token";
21/// Tokens are refreshed this long before they expire.
22const REFRESH_MARGIN: Duration = Duration::from_secs(60);
23/// Lifetime requested for service account assertions, the longest Google allows.
24const ASSERTION_LIFETIME: Duration = Duration::from_secs(3600);
25
26/// Where the Google credentials come from.
27#[derive(Clone)]
28pub enum GcsCredentials {
29    /// Google's Application Default Credentials lookup: the file named by
30    /// `GOOGLE_APPLICATION_CREDENTIALS`, then the file written by
31    /// `gcloud auth application-default login`, then the metadata server.
32    ApplicationDefault,
33    /// A service account key file as downloaded from the Cloud Console.
34    ServiceAccountKeyFile(PathBuf),
35    /// The contents of a service account key file.
36    ServiceAccountKeyJson(String),
37    /// The metadata server of GCE, GKE, Cloud Run and similar runtimes.
38    MetadataServer,
39    /// A fixed `OAuth2` access token, for tests and emulators.
40    AccessToken(String),
41    /// No `Authorization` header at all, for emulators such as fake-gcs-server.
42    Anonymous,
43}
44
45/// Redacts key material and tokens.
46impl Debug for GcsCredentials {
47    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
48        match self {
49            Self::ApplicationDefault => f.write_str("ApplicationDefault"),
50            Self::ServiceAccountKeyFile(path) => {
51                f.debug_tuple("ServiceAccountKeyFile").field(path).finish()
52            }
53            Self::ServiceAccountKeyJson(_) => f.write_str("ServiceAccountKeyJson(REDACTED)"),
54            Self::MetadataServer => f.write_str("MetadataServer"),
55            Self::AccessToken(_) => f.write_str("AccessToken(REDACTED)"),
56            Self::Anonymous => f.write_str("Anonymous"),
57        }
58    }
59}
60
61/// Hands out bearer tokens, refreshing short-lived ones before they expire.
62pub(super) struct Authenticator {
63    source: TokenSource,
64    cached: Mutex<Option<CachedToken>>,
65}
66
67enum TokenSource {
68    /// Always the same value; `None` sends no `Authorization` header.
69    Static(Option<String>),
70    Refreshing(Refresh),
71}
72
73/// A way to obtain a fresh access token.
74enum Refresh {
75    ServiceAccount {
76        email: String,
77        key: EncodingKey,
78        token_uri: String,
79    },
80    AuthorizedUser(AuthorizedUser),
81    MetadataServer {
82        url: String,
83    },
84}
85
86struct CachedToken {
87    value: String,
88    expires_at: Instant,
89}
90
91impl Authenticator {
92    pub(super) async fn new(credentials: &GcsCredentials) -> Result<Self, StorageError> {
93        let source = match credentials {
94            GcsCredentials::ApplicationDefault => application_default().await?,
95            GcsCredentials::ServiceAccountKeyFile(path) => from_file(path).await?,
96            GcsCredentials::ServiceAccountKeyJson(json) => from_json(json)?,
97            GcsCredentials::MetadataServer => TokenSource::Refreshing(Refresh::MetadataServer {
98                url: metadata_url(),
99            }),
100            GcsCredentials::AccessToken(token) => TokenSource::Static(Some(token.clone())),
101            GcsCredentials::Anonymous => TokenSource::Static(None),
102        };
103
104        Ok(Self {
105            source,
106            cached: Mutex::new(None),
107        })
108    }
109
110    /// The bearer token to send, or `None` for anonymous access.
111    pub(super) async fn bearer(&self, client: &Client) -> Result<Option<String>, StorageError> {
112        let refresh = match &self.source {
113            TokenSource::Static(token) => return Ok(token.clone()),
114            TokenSource::Refreshing(refresh) => refresh,
115        };
116
117        // Holding the lock while fetching means one refresh at a time.
118        let mut cached = self.cached.lock().await;
119        if let Some(token) = cached.as_ref()
120            && token.expires_at.saturating_duration_since(Instant::now()) > REFRESH_MARGIN
121        {
122            return Ok(Some(token.value.clone()));
123        }
124
125        let token = with_retry("token", || refresh.fetch(client))
126            .await
127            .map_err(credentials_rejected)?;
128        let value = token.value.clone();
129        *cached = Some(token);
130        Ok(Some(value))
131    }
132}
133
134impl Refresh {
135    async fn fetch(&self, client: &Client) -> Result<CachedToken, StorageError> {
136        let requested_at = Instant::now();
137        let request = match self {
138            Self::ServiceAccount {
139                email,
140                key,
141                token_uri,
142            } => client.post(token_uri).form(&[
143                ("grant_type", "urn:ietf:params:oauth:grant-type:jwt-bearer"),
144                ("assertion", &sign_assertion(email, key, token_uri)?),
145            ]),
146            Self::AuthorizedUser(user) => client.post(DEFAULT_TOKEN_URI).form(&[
147                ("grant_type", "refresh_token"),
148                ("client_id", &user.client_id),
149                ("client_secret", &user.client_secret),
150                ("refresh_token", &user.refresh_token),
151            ]),
152            Self::MetadataServer { url } => client.get(url).header("Metadata-Flavor", "Google"),
153        };
154
155        let response = request.send().await.map_err(transport)?;
156        let token: TokenResponse = parse_json(check(response, "token endpoint").await?).await?;
157        Ok(CachedToken {
158            value: token.access_token,
159            expires_at: requested_at + Duration::from_secs(token.expires_in),
160        })
161    }
162}
163
164/// Token endpoints answer bad credentials with 400 as often as with 401.
165fn credentials_rejected(error: StorageError) -> StorageError {
166    match error {
167        StorageError::Backend(400 | 401 | 403, message) | StorageError::Forbidden(message) => {
168            StorageError::Unauthorized(message)
169        }
170        other => other,
171    }
172}
173
174#[derive(Deserialize)]
175struct ServiceAccountKey {
176    client_email: String,
177    private_key: String,
178    #[serde(default = "default_token_uri")]
179    token_uri: String,
180}
181
182fn default_token_uri() -> String {
183    DEFAULT_TOKEN_URI.to_string()
184}
185
186#[derive(Deserialize)]
187struct AuthorizedUser {
188    client_id: String,
189    client_secret: String,
190    refresh_token: String,
191}
192
193#[derive(Serialize)]
194struct AssertionClaims<'a> {
195    iss: &'a str,
196    scope: &'a str,
197    aud: &'a str,
198    iat: u64,
199    exp: u64,
200}
201
202#[derive(Deserialize)]
203struct TokenResponse {
204    access_token: String,
205    expires_in: u64,
206}
207
208/// A signed JWT that the token endpoint exchanges for an access token.
209fn sign_assertion(email: &str, key: &EncodingKey, token_uri: &str) -> Result<String, StorageError> {
210    let now = SystemTime::now()
211        .duration_since(UNIX_EPOCH)
212        .map_err(|error| StorageError::Configuration(format!("system clock is invalid: {error}")))?
213        .as_secs();
214    let claims = AssertionClaims {
215        iss: email,
216        scope: SCOPE,
217        aud: token_uri,
218        iat: now,
219        exp: now + ASSERTION_LIFETIME.as_secs(),
220    };
221    jsonwebtoken::encode(&Header::new(Algorithm::RS256), &claims, key).map_err(|error| {
222        StorageError::Configuration(format!("could not sign service account assertion: {error}"))
223    })
224}
225
226async fn application_default() -> Result<TokenSource, StorageError> {
227    if let Some(path) = std::env::var_os("GOOGLE_APPLICATION_CREDENTIALS") {
228        return from_file(Path::new(&path)).await;
229    }
230    if let Some(path) = gcloud_credentials_file()
231        && tokio::fs::try_exists(&path).await.unwrap_or(false)
232    {
233        return from_file(&path).await;
234    }
235    Ok(TokenSource::Refreshing(Refresh::MetadataServer {
236        url: metadata_url(),
237    }))
238}
239
240/// Where `gcloud auth application-default login` writes its credentials.
241fn gcloud_credentials_file() -> Option<PathBuf> {
242    let config_dir = std::env::var_os("CLOUDSDK_CONFIG")
243        .map(PathBuf::from)
244        .or_else(|| {
245            if cfg!(windows) {
246                std::env::var_os("APPDATA").map(|dir| PathBuf::from(dir).join("gcloud"))
247            } else {
248                std::env::var_os("HOME").map(|home| PathBuf::from(home).join(".config/gcloud"))
249            }
250        })?;
251    Some(config_dir.join("application_default_credentials.json"))
252}
253
254fn metadata_url() -> String {
255    let host =
256        std::env::var("GCE_METADATA_HOST").unwrap_or_else(|_| DEFAULT_METADATA_HOST.to_string());
257    format!("http://{host}{METADATA_TOKEN_PATH}")
258}
259
260async fn from_file(path: &Path) -> Result<TokenSource, StorageError> {
261    let json = tokio::fs::read_to_string(path).await.map_err(|error| {
262        StorageError::Configuration(format!(
263            "could not read credentials file {}: {error}",
264            path.display()
265        ))
266    })?;
267    from_json(&json)
268}
269
270/// Parses a credentials file: a service account key or an authorized user
271/// (the shape `gcloud auth application-default login` produces).
272fn from_json(json: &str) -> Result<TokenSource, StorageError> {
273    let invalid = |error: serde_json::Error| {
274        StorageError::Configuration(format!("invalid Google credentials JSON: {error}"))
275    };
276    let value: serde_json::Value = serde_json::from_str(json).map_err(invalid)?;
277    let kind = value
278        .get("type")
279        .and_then(serde_json::Value::as_str)
280        .map(str::to_owned);
281
282    let refresh = match kind.as_deref() {
283        Some("service_account") => {
284            let account: ServiceAccountKey = serde_json::from_value(value).map_err(invalid)?;
285            let key =
286                EncodingKey::from_rsa_pem(account.private_key.as_bytes()).map_err(|error| {
287                    StorageError::Configuration(format!(
288                        "invalid service account private key: {error}"
289                    ))
290                })?;
291            Refresh::ServiceAccount {
292                email: account.client_email,
293                key,
294                token_uri: account.token_uri,
295            }
296        }
297        Some("authorized_user") => {
298            Refresh::AuthorizedUser(serde_json::from_value(value).map_err(invalid)?)
299        }
300        other => {
301            return Err(StorageError::Configuration(format!(
302                "unsupported Google credentials type {other:?}; expected service_account or authorized_user"
303            )));
304        }
305    };
306    Ok(TokenSource::Refreshing(refresh))
307}
haste_document_storage/providers/gcs/auth.rs

haste_document_storage/providers/gcs/
auth.rs
