Additional GCP authentication

object_store/src/gcp/credential.rs

@@ -18,15 +18,26 @@
 use crate::client::retry::RetryExt;
 use crate::client::token::TemporaryToken;
 use crate::RetryConfig;
+use async_trait::async_trait;
 use base64::prelude::BASE64_URL_SAFE_NO_PAD;
 use base64::Engine;
 use reqwest::{Client, Method};
 use ring::signature::RsaKeyPair;
 use snafu::{ResultExt, Snafu};
+use std::env;
+use std::fs::File;
+use std::io::BufReader;
+use std::path::Path;
 use std::time::{Duration, Instant};
 
 #[derive(Debug, Snafu)]
 pub enum Error {
+    #[snafu(display("Unable to open service account file: {}", source))]
+    OpenCredentials { source: std::io::Error },
+
+    #[snafu(display("Unable to decode service account file: {}", source))]
+    DecodeCredentials { source: serde_json::Error },
+
     #[snafu(display("No RSA key found in pem file"))]
     MissingKey,
 
@@ -104,6 +115,15 @@ struct TokenResponse {
     expires_in: u64,
 }
 
+#[async_trait]
+pub(crate) trait TokenProvider: std::fmt::Debug + Send + Sync {
+    async fn fetch_token(
+        &self,
+        client: &Client,
+        retry: &RetryConfig,
+    ) -> Result<TemporaryToken<String>>;
+}
+
 /// Encapsulates the logic to perform an OAuth token challenge
 #[derive(Debug)]
 pub struct OAuthProvider {
@@ -138,9 +158,12 @@ impl OAuthProvider {
             random: ring::rand::SystemRandom::new(),
         })
     }
+}
 
+#[async_trait]
+impl TokenProvider for OAuthProvider {
     /// Fetch a fresh token
-    pub async fn fetch_token(
+    async fn fetch_token(
         &self,
         client: &Client,
         retry: &RetryConfig,
@@ -195,6 +218,88 @@ impl OAuthProvider {
     }
 }
 
+fn reader_credentials_file<T>(
+    service_account_path: impl AsRef<std::path::Path>,
+) -> Result<T>
+where
+    T: serde::de::DeserializeOwned,
+{
+    let file = File::open(service_account_path).context(OpenCredentialsSnafu)?;
+    let reader = BufReader::new(file);
+    serde_json::from_reader(reader).context(DecodeCredentialsSnafu)
+}
+
+/// A deserialized `service-account-********.json`-file.
+#[derive(serde::Deserialize, Debug)]
+pub(crate) struct ServiceAccountCredentials {
+    /// The private key in RSA format.
+    pub private_key: String,
+
+    /// The email address associated with the service account.
+    pub client_email: String,
+
+    /// Base URL for GCS
+    #[serde(default = "default_gcs_base_url")]
+    pub gcs_base_url: String,
+
+    /// Disable oauth and use empty tokens.
+    #[serde(default = "default_disable_oauth")]
+    pub disable_oauth: bool,
+}
+
+pub(crate) fn default_gcs_base_url() -> String {
+    "https://storage.googleapis.com".to_owned()
+}
+
+pub(crate) fn default_disable_oauth() -> bool {
+    false
+}
+
+impl ServiceAccountCredentials {
+    /// Create a new [`ServiceAccountCredentials`] from a file.
+    pub fn from_file<P: AsRef<std::path::Path>>(path: P) -> Result<Self> {
+        reader_credentials_file(path)
+    }
+
+    /// Create a new [`ServiceAccountCredentials`] from a string.
+    pub fn from_key(key: &str) -> Result<Self> {
+        serde_json::from_str(key).context(DecodeCredentialsSnafu)
+    }
+
+    /// Create an [`OAuthProvider`] from this credentials struct.
+    pub fn token_provider(
+        self,
+        scope: &str,
+        audience: &str,
+    ) -> Result<Box<dyn TokenProvider>> {
+        Ok(Box::new(OAuthProvider::new(
+            self.client_email,
+            self.private_key,
+            scope.to_string(),
+            audience.to_string(),
+        )?) as Box<dyn TokenProvider>)
+    }
+}
+
+/// A no-op provider that returns empty tokens
+#[derive(Debug)]
+pub struct NoOpProvider;
+
+#[async_trait]
+impl TokenProvider for NoOpProvider {
+    /// Fetch a fresh token
+    async fn fetch_token(
+        &self,
+        _client: &Client,
+        _retry: &RetryConfig,
+    ) -> Result<TemporaryToken<String>> {
+        Ok(TemporaryToken {
+            token: "".to_string(),
+            expiry: Instant::now(),
+        })
+    }
+}
+
 /// Returns the number of seconds since unix epoch
 fn seconds_since_epoch() -> u64 {
     std::time::SystemTime::now()
@@ -205,7 +310,7 @@ fn seconds_since_epoch() -> u64 {
 
 fn decode_first_rsa_key(private_key_pem: String) -> Result<RsaKeyPair> {
     use rustls_pemfile::Item;
-    use std::io::{BufReader, Cursor};
+    use std::io::Cursor;
 
     let mut cursor = Cursor::new(private_key_pem);
     let mut reader = BufReader::new(&mut cursor);
@@ -222,3 +327,118 @@ fn b64_encode_obj<T: serde::Serialize>(obj: &T) -> Result<String> {
     let string = serde_json::to_string(obj).context(EncodeSnafu)?;
     Ok(BASE64_URL_SAFE_NO_PAD.encode(string))
 }
+
+/// A provider that uses the Google Cloud Platform metadata server to fetch a token.
+#[derive(Debug, Default)]
+pub struct InstanceCredentialProvider {
+    audience: String,
+}
+
+impl InstanceCredentialProvider {
+    pub fn new<T: Into<String>>(audience: T) -> Self {
+        Self {
+            audience: audience.into(),
+        }
+    }
+}
+
+#[async_trait]
+impl TokenProvider for InstanceCredentialProvider {
+    async fn fetch_token(
+        &self,
+        client: &Client,
+        retry: &RetryConfig,
+    ) -> Result<TemporaryToken<String>> {
+        println!("fetching token from metadata server");
+        const TOKEN_URL: &str =
+            "http://metadata/computeMetadata/v1/instance/service-accounts/default/token";
+        let response: TokenResponse = client
+            .request(Method::GET, TOKEN_URL)
+            .header("Metadata-Flavor", "Google")
+            .query(&[("audience", &self.audience)])
+            .send_retry(retry)
+            .await
+            .context(TokenRequestSnafu)?
+            .json()
+            .await
+            .context(TokenResponseBodySnafu)?;
+        let token = TemporaryToken {
+            token: response.access_token,
+            expiry: Instant::now() + Duration::from_secs(response.expires_in),
+        };
+        Ok(token)
+    }
+}
+
+/// A deserialized `application_default_credentials.json`-file.
+#[derive(serde::Deserialize, Debug)]
+pub struct ApplicationDefaultCredentials {
+    client_id: String,
+    client_secret: String,
+    refresh_token: String,
+    #[serde(rename = "type")]
+    type_: String,
+}
+
+impl ApplicationDefaultCredentials {
+    const DEFAULT_TOKEN_GCP_URI: &'static str =
+        "https://accounts.google.com/o/oauth2/token";
+    const CREDENTIALS_PATH: &'static str =
+        ".config/gcloud/application_default_credentials.json";
+    const EXPECTED_TYPE: &str = "authorized_user";
+
+    // Create a new application default credential in the following situations:
+    //  1. a file is passed in and the type matches.
+    //  2. without argument if the well-known configuration file is present.
+    pub fn new(path: Option<&str>) -> Result<Option<Self>, Error> {
+        if let Some(path) = path {
+            if let Ok(credentials) = reader_credentials_file::<Self>(path) {
+                if credentials.type_ == Self::EXPECTED_TYPE {
+                    return Ok(Some(credentials));
+                }
+            }
+            // Other credential mechanisms may be able to use this path.
+            return Ok(None);
+        }
+        if let Some(home) = env::var_os("HOME") {
+            let path = Path::new(&home).join(Self::CREDENTIALS_PATH);
+
+            // It's expected for this file to not exist unless it has been explicitly configured by the user.
+            if path.try_exists().unwrap_or(false) {
+                return reader_credentials_file::<Self>(path).map(Some);
+            }
+        }
+        Ok(None)
+    }
+}
+
+#[async_trait]
+impl TokenProvider for ApplicationDefaultCredentials {
+    async fn fetch_token(
+        &self,
+        client: &Client,
+        retry: &RetryConfig,
+    ) -> Result<TemporaryToken<String>, Error> {
+        let body = [
+            ("grant_type", "refresh_token"),
+            ("client_id", &self.client_id),
+            ("client_secret", &self.client_secret),
+            ("refresh_token", &self.refresh_token),
+        ];
+
+        let response = client
+            .request(Method::POST, Self::DEFAULT_TOKEN_GCP_URI)
+            .form(&body)
+            .send_retry(retry)
+            .await
+            .context(TokenRequestSnafu)?
+            .json::<TokenResponse>()
+            .await
+            .context(TokenResponseBodySnafu)?;
+        let token = TemporaryToken {
+            token: response.access_token,
+            expiry: Instant::now() + Duration::from_secs(response.expires_in),
+        };
+        Ok(token)
+    }
+}