object_store/gcp/

credential.rs

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

use super::client::GoogleCloudStorageClient;
use crate::client::builder::HttpRequestBuilder;
use crate::client::retry::RetryExt;
use crate::client::token::TemporaryToken;
use crate::client::{
    CryptoProvider, HttpClient, HttpError, Signer, SigningAlgorithm, TokenProvider,
};
use crate::gcp::{GcpSigningCredentialProvider, STORE};
use crate::util::{STRICT_ENCODE_SET, append_strict_query_pairs, hex_digest, hex_encode};
use crate::{RetryConfig, StaticCredentialProvider};
use async_trait::async_trait;
use base64::Engine;
use base64::prelude::BASE64_URL_SAFE_NO_PAD;
use chrono::{DateTime, Utc};
use futures_util::TryFutureExt;
use http::{HeaderMap, Method};
use itertools::Itertools;
use percent_encoding::utf8_percent_encode;
use serde::Deserialize;
use std::collections::BTreeMap;
use std::env;
use std::fs::File;
use std::io::BufReader;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::{Duration, Instant};
use tracing::info;
use url::Url;

pub(crate) const DEFAULT_SCOPE: &str = "https://www.googleapis.com/auth/cloud-platform";

pub(crate) const DEFAULT_GCS_BASE_URL: &str = "https://storage.googleapis.com";

const DEFAULT_GCS_PLAYLOAD_STRING: &str = "UNSIGNED-PAYLOAD";
const DEFAULT_GCS_SIGN_BLOB_HOST: &str = "storage.googleapis.com";

const DEFAULT_METADATA_HOST: &str = "metadata.google.internal";
const DEFAULT_METADATA_IP: &str = "169.254.169.254";

#[derive(Debug, thiserror::Error)]
pub(super) enum Error {
    #[error("Unable to open service account file from {}: {}", path.display(), source)]
    OpenCredentials {
        source: std::io::Error,
        path: PathBuf,
    },

    #[error("Unable to decode service account file: {}", source)]
    DecodeCredentials { source: serde_json::Error },

    #[error("Error encoding jwt payload: {}", source)]
    Encode { source: serde_json::Error },

    #[error("Error performing token request: {}", source)]
    TokenRequest {
        source: crate::client::retry::RetryError,
    },

    #[error("Error getting token response body: {}", source)]
    TokenResponseBody { source: HttpError },
}

impl From<Error> for crate::Error {
    fn from(value: Error) -> Self {
        Self::Generic {
            store: STORE,
            source: Box::new(value),
        }
    }
}

/// A Google Cloud Storage Credential for signing
#[derive(Debug)]
pub struct GcpSigningCredential {
    /// The email of the service account
    pub email: String,

    /// An optional RSA private key
    ///
    /// If provided this will be used to sign the URL, otherwise a call will be made to
    /// [`iam.serviceAccounts.signBlob`]. This allows supporting credential sources
    /// that don't expose the service account private key, e.g. [IMDS].
    ///
    /// [IMDS]: https://cloud.google.com/docs/authentication/get-id-token#metadata-server
    /// [`iam.serviceAccounts.signBlob`]: https://cloud.google.com/storage/docs/authentication/creating-signatures
    pub private_key: Option<ServiceAccountKey>,
}

/// A private RSA key for a service account
pub struct ServiceAccountKey(Box<dyn Signer>);

impl std::fmt::Debug for ServiceAccountKey {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_tuple("ServiceAccountKey").finish_non_exhaustive()
    }
}

impl ServiceAccountKey {
    /// Creates a [`ServiceAccountKey`] from the provided [`Signer`]
    pub fn new(signer: Box<dyn Signer>) -> Self {
        Self(signer)
    }

    /// Parses a pem-encoded RSA key
    #[cfg(feature = "aws-lc-rs")]
    pub fn from_pem(encoded: &[u8]) -> crate::Result<Self> {
        let key = crate::client::aws_lc_rs::RsaKeyPair::from_pem(encoded)?;
        Ok(Self::new(Box::new(key)))
    }

    /// Parses an unencrypted PKCS#8-encoded RSA private key.
    #[cfg(feature = "aws-lc-rs")]
    pub fn from_pkcs8(key: &[u8]) -> crate::Result<Self> {
        let key = crate::client::aws_lc_rs::RsaKeyPair::from_pkcs8(key)?;
        Ok(Self::new(Box::new(key)))
    }

    /// Parses an unencrypted PKCS#8-encoded RSA private key.
    #[cfg(feature = "aws-lc-rs")]
    pub fn from_der(key: &[u8]) -> crate::Result<Self> {
        let key = crate::client::aws_lc_rs::RsaKeyPair::from_der(key)?;
        Ok(Self::new(Box::new(key)))
    }

    /// Parses a pem-encoded RSA key
    #[cfg(all(feature = "ring", not(feature = "aws-lc-rs")))]
    pub fn from_pem(encoded: &[u8]) -> crate::Result<Self> {
        let key = crate::client::ring::RsaKeyPair::from_pem(encoded)?;
        Ok(Self::new(Box::new(key)))
    }

    /// Parses an unencrypted PKCS#8-encoded RSA private key.
    #[cfg(all(feature = "ring", not(feature = "aws-lc-rs")))]
    pub fn from_pkcs8(key: &[u8]) -> crate::Result<Self> {
        let key = crate::client::ring::RsaKeyPair::from_pkcs8(key)?;
        Ok(Self::new(Box::new(key)))
    }

    /// Parses an unencrypted PKCS#8-encoded RSA private key.
    #[cfg(all(feature = "ring", not(feature = "aws-lc-rs")))]
    pub fn from_der(key: &[u8]) -> crate::Result<Self> {
        let key = crate::client::ring::RsaKeyPair::from_der(key)?;
        Ok(Self::new(Box::new(key)))
    }

    fn sign(&self, string_to_sign: &[u8]) -> crate::Result<Vec<u8>> {
        self.0.sign(string_to_sign)
    }
}

/// A Google Cloud Storage Credential
#[derive(Debug, Eq, PartialEq)]
pub struct GcpCredential {
    /// An HTTP bearer token
    pub bearer: String,
}

pub(crate) type Result<T, E = Error> = std::result::Result<T, E>;

#[derive(Debug, Default, serde::Serialize)]
pub(crate) struct JwtHeader<'a> {
    /// The type of JWS: it can only be "JWT" here
    ///
    /// Defined in [RFC7515#4.1.9](https://tools.ietf.org/html/rfc7515#section-4.1.9).
    #[serde(skip_serializing_if = "Option::is_none")]
    pub typ: Option<&'a str>,
    /// The algorithm used
    ///
    /// Defined in [RFC7515#4.1.1](https://tools.ietf.org/html/rfc7515#section-4.1.1).
    pub alg: &'a str,
    /// Content type
    ///
    /// Defined in [RFC7519#5.2](https://tools.ietf.org/html/rfc7519#section-5.2).
    #[serde(skip_serializing_if = "Option::is_none")]
    pub cty: Option<&'a str>,
    /// JSON Key URL
    ///
    /// Defined in [RFC7515#4.1.2](https://tools.ietf.org/html/rfc7515#section-4.1.2).
    #[serde(skip_serializing_if = "Option::is_none")]
    pub jku: Option<&'a str>,
    /// Key ID
    ///
    /// Defined in [RFC7515#4.1.4](https://tools.ietf.org/html/rfc7515#section-4.1.4).
    #[serde(skip_serializing_if = "Option::is_none")]
    pub kid: Option<&'a str>,
    /// X.509 URL
    ///
    /// Defined in [RFC7515#4.1.5](https://tools.ietf.org/html/rfc7515#section-4.1.5).
    #[serde(skip_serializing_if = "Option::is_none")]
    pub x5u: Option<&'a str>,
    /// X.509 certificate thumbprint
    ///
    /// Defined in [RFC7515#4.1.7](https://tools.ietf.org/html/rfc7515#section-4.1.7).
    #[serde(skip_serializing_if = "Option::is_none")]
    pub x5t: Option<&'a str>,
}

#[derive(serde::Serialize)]
struct TokenClaims<'a> {
    iss: &'a str,
    sub: &'a str,
    scope: &'a str,
    exp: u64,
    iat: u64,
}

#[derive(serde::Deserialize, Debug)]
struct TokenResponse {
    access_token: String,
    expires_in: u64,
    id_token: Option<String>,
}

/// Self-signed JWT (JSON Web Token).
///
/// # References
/// - <https://google.aip.dev/auth/4111>
#[derive(Debug)]
pub(crate) struct SelfSignedJwt {
    issuer: String,
    scope: String,
    private_key: ServiceAccountKey,
    key_id: String,
}

impl SelfSignedJwt {
    /// Create a new [`SelfSignedJwt`]
    pub(crate) fn new(
        key_id: String,
        issuer: String,
        private_key: ServiceAccountKey,
        scope: String,
    ) -> Result<Self> {
        Ok(Self {
            issuer,
            scope,
            private_key,
            key_id,
        })
    }
}

#[async_trait]
impl TokenProvider for SelfSignedJwt {
    type Credential = GcpCredential;

    /// Fetch a fresh token
    async fn fetch_token(
        &self,
        _client: &HttpClient,
        _retry: &RetryConfig,
    ) -> crate::Result<TemporaryToken<Arc<GcpCredential>>> {
        let now = seconds_since_epoch();
        let exp = now + 3600;

        let claims = TokenClaims {
            iss: &self.issuer,
            sub: &self.issuer,
            scope: &self.scope,
            iat: now,
            exp,
        };

        let jwt_header = b64_encode_obj(&JwtHeader {
            alg: "RS256",
            typ: Some("JWT"),
            kid: Some(&self.key_id),
            ..Default::default()
        })?;

        let claim_str = b64_encode_obj(&claims)?;
        let message = [jwt_header.as_ref(), claim_str.as_ref()].join(".");
        let sig_bytes = self.private_key.sign(message.as_bytes())?;
        let signature = BASE64_URL_SAFE_NO_PAD.encode(sig_bytes);
        let bearer = [message, signature].join(".");

        Ok(TemporaryToken {
            token: Arc::new(GcpCredential { bearer }),
            expiry: Some(Instant::now() + Duration::from_secs(3600)),
        })
    }
}

fn read_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).map_err(|source| {
        let path = service_account_path.as_ref().to_owned();
        Error::OpenCredentials { source, path }
    })?;
    let reader = BufReader::new(file);
    serde_json::from_reader(reader).map_err(|source| Error::DecodeCredentials { source })
}

/// A deserialized `service-account-********.json`-file.
#[derive(serde::Deserialize, Debug, Clone)]
pub(crate) struct ServiceAccountCredentials {
    /// The private key in RSA format.
    pub private_key: String,

    /// The private key ID
    pub private_key_id: String,

    /// The email address associated with the service account.
    pub client_email: String,

    /// Base URL for GCS
    #[serde(default)]
    pub gcs_base_url: Option<String>,

    /// Disable oauth and use empty tokens.
    #[serde(default)]
    pub disable_oauth: bool,
}

impl ServiceAccountCredentials {
    /// Create a new [`ServiceAccountCredentials`] from a file.
    pub(crate) fn from_file<P: AsRef<Path>>(path: P) -> Result<Self> {
        read_credentials_file(path)
    }

    /// Create a new [`ServiceAccountCredentials`] from a string.
    pub(crate) fn from_key(key: &str) -> Result<Self> {
        serde_json::from_str(key).map_err(|source| Error::DecodeCredentials { source })
    }

    /// Create a [`SelfSignedJwt`] from this credentials struct.
    ///
    /// We use a scope of [`DEFAULT_SCOPE`] as opposed to an audience
    /// as GCS appears to not support audience
    ///
    /// # References
    /// - <https://stackoverflow.com/questions/63222450/service-account-authorization-without-oauth-can-we-get-file-from-google-cloud/71834557#71834557>
    /// - <https://www.codejam.info/2022/05/google-cloud-service-account-authorization-without-oauth.html>
    pub(crate) fn token_provider(
        self,
        crypto: &dyn CryptoProvider,
    ) -> crate::Result<SelfSignedJwt> {
        let key = crypto.sign(SigningAlgorithm::RS256, self.private_key.as_bytes())?;
        Ok(SelfSignedJwt::new(
            self.private_key_id,
            self.client_email,
            ServiceAccountKey::new(key),
            DEFAULT_SCOPE.to_string(),
        )?)
    }

    pub(crate) fn signing_credentials(
        self,
        crypto: &dyn CryptoProvider,
    ) -> crate::Result<GcpSigningCredentialProvider> {
        let key = crypto.sign(SigningAlgorithm::RS256, self.private_key.as_bytes())?;
        Ok(Arc::new(StaticCredentialProvider::new(
            GcpSigningCredential {
                email: self.client_email,
                private_key: Some(ServiceAccountKey::new(key)),
            },
        )))
    }
}

/// Returns the number of seconds since unix epoch
fn seconds_since_epoch() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::SystemTime::UNIX_EPOCH)
        .unwrap()
        .as_secs()
}

fn b64_encode_obj<T: serde::Serialize>(obj: &T) -> Result<String> {
    let string = serde_json::to_string(obj).map_err(|source| Error::Encode { source })?;
    Ok(BASE64_URL_SAFE_NO_PAD.encode(string))
}

/// A provider that uses the Google Cloud Platform metadata server to fetch a token.
///
/// <https://cloud.google.com/docs/authentication/get-id-token#metadata-server>
#[derive(Debug, Default)]
pub(crate) struct InstanceCredentialProvider {}

/// Make a request to the metadata server to fetch a token, using a a given hostname.
async fn make_metadata_request(
    client: &HttpClient,
    hostname: &str,
    retry: &RetryConfig,
) -> crate::Result<TokenResponse> {
    let url =
        format!("http://{hostname}/computeMetadata/v1/instance/service-accounts/default/token");
    let response: TokenResponse = client
        .get(url)
        .header("Metadata-Flavor", "Google")
        .query(&[("audience", "https://www.googleapis.com/oauth2/v4/token")])
        .send_retry(retry)
        .await
        .map_err(|source| Error::TokenRequest { source })?
        .into_body()
        .json()
        .await
        .map_err(|source| Error::TokenResponseBody { source })?;
    Ok(response)
}

#[async_trait]
impl TokenProvider for InstanceCredentialProvider {
    type Credential = GcpCredential;

    /// Fetch a token from the metadata server.
    /// Since the connection is local we need to enable http access and don't actually use the client object passed in.
    /// Respects the `GCE_METADATA_HOST`, `GCE_METADATA_ROOT`, and `GCE_METADATA_IP`
    /// environment variables.
    ///
    /// References: <https://googleapis.dev/python/google-auth/latest/reference/google.auth.environment_vars.html>
    async fn fetch_token(
        &self,
        client: &HttpClient,
        retry: &RetryConfig,
    ) -> crate::Result<TemporaryToken<Arc<GcpCredential>>> {
        let metadata_host = if let Ok(host) = env::var("GCE_METADATA_HOST") {
            host
        } else if let Ok(host) = env::var("GCE_METADATA_ROOT") {
            host
        } else {
            DEFAULT_METADATA_HOST.to_string()
        };
        let metadata_ip = if let Ok(ip) = env::var("GCE_METADATA_IP") {
            ip
        } else {
            DEFAULT_METADATA_IP.to_string()
        };

        info!("fetching token from metadata server");
        let response = make_metadata_request(client, &metadata_host, retry)
            .or_else(|_| make_metadata_request(client, &metadata_ip, retry))
            .await?;

        let token = TemporaryToken {
            token: Arc::new(GcpCredential {
                bearer: response.access_token,
            }),
            expiry: Some(Instant::now() + Duration::from_secs(response.expires_in)),
        };
        Ok(token)
    }
}

/// Make a request to the metadata server to fetch the client email, using a given hostname.
async fn make_metadata_request_for_email(
    client: &HttpClient,
    hostname: &str,
    retry: &RetryConfig,
) -> crate::Result<String> {
    let url =
        format!("http://{hostname}/computeMetadata/v1/instance/service-accounts/default/email",);
    let response = client
        .get(url)
        .header("Metadata-Flavor", "Google")
        .send_retry(retry)
        .await
        .map_err(|source| Error::TokenRequest { source })?
        .into_body()
        .text()
        .await
        .map_err(|source| Error::TokenResponseBody { source })?;
    Ok(response)
}

/// A provider that uses the Google Cloud Platform metadata server to fetch a email for signing.
///
/// <https://cloud.google.com/appengine/docs/legacy/standard/java/accessing-instance-metadata>
#[derive(Debug, Default)]
pub(crate) struct InstanceSigningCredentialProvider {}

#[async_trait]
impl TokenProvider for InstanceSigningCredentialProvider {
    type Credential = GcpSigningCredential;

    /// Fetch a token from the metadata server.
    /// Since the connection is local we need to enable http access and don't actually use the client object passed in.
    /// Respects the `GCE_METADATA_HOST`, `GCE_METADATA_ROOT`, and `GCE_METADATA_IP`
    /// environment variables.
    ///
    /// References: <https://googleapis.dev/python/google-auth/latest/reference/google.auth.environment_vars.html>
    async fn fetch_token(
        &self,
        client: &HttpClient,
        retry: &RetryConfig,
    ) -> crate::Result<TemporaryToken<Arc<GcpSigningCredential>>> {
        let metadata_host = if let Ok(host) = env::var("GCE_METADATA_HOST") {
            host
        } else if let Ok(host) = env::var("GCE_METADATA_ROOT") {
            host
        } else {
            DEFAULT_METADATA_HOST.to_string()
        };

        let metadata_ip = if let Ok(ip) = env::var("GCE_METADATA_IP") {
            ip
        } else {
            DEFAULT_METADATA_IP.to_string()
        };

        info!("fetching token from metadata server");

        let email = make_metadata_request_for_email(client, &metadata_host, retry)
            .or_else(|_| make_metadata_request_for_email(client, &metadata_ip, retry))
            .await?;

        let token = TemporaryToken {
            token: Arc::new(GcpSigningCredential {
                email,
                private_key: None,
            }),
            expiry: None,
        };
        Ok(token)
    }
}

/// A deserialized `application_default_credentials.json`-file.
///
/// # References
/// - <https://cloud.google.com/docs/authentication/application-default-credentials#personal>
/// - <https://google.aip.dev/auth/4110>
#[derive(serde::Deserialize, Clone)]
#[serde(tag = "type")]
pub(crate) enum ApplicationDefaultCredentials {
    /// Service Account.
    ///
    /// # References
    /// - <https://google.aip.dev/auth/4112>
    #[serde(rename = "service_account")]
    ServiceAccount(ServiceAccountCredentials),
    /// Authorized user via "gcloud CLI Integration".
    ///
    /// # References
    /// - <https://google.aip.dev/auth/4113>
    #[serde(rename = "authorized_user")]
    AuthorizedUser(AuthorizedUserCredentials),
}

impl ApplicationDefaultCredentials {
    const CREDENTIALS_PATH: &'static str = if cfg!(windows) {
        "gcloud/application_default_credentials.json"
    } else {
        ".config/gcloud/application_default_credentials.json"
    };

    // 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(crate) fn read(path: Option<&str>) -> Result<Option<Self>, Error> {
        if let Some(path) = path {
            return read_credentials_file::<Self>(path).map(Some);
        }

        let home_var = if cfg!(windows) { "APPDATA" } else { "HOME" };
        if let Some(home) = env::var_os(home_var) {
            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.exists() {
                return read_credentials_file::<Self>(path).map(Some);
            }
        }
        Ok(None)
    }
}

const DEFAULT_TOKEN_GCP_URI: &str = "https://accounts.google.com/o/oauth2/token";

/// <https://google.aip.dev/auth/4113>
#[derive(Debug, Deserialize, Clone)]
pub(crate) struct AuthorizedUserCredentials {
    client_id: String,
    client_secret: String,
    refresh_token: String,
}

#[derive(Debug, Deserialize)]
pub(crate) struct AuthorizedUserSigningCredentials {
    credential: AuthorizedUserCredentials,
}

///<https://oauth2.googleapis.com/tokeninfo?access_token=ACCESS_TOKEN>
#[derive(Debug, Deserialize)]
struct EmailResponse {
    email: String,
}

#[derive(Debug, Deserialize)]
struct IdTokenClaims {
    email: String,
}

async fn get_token_response(
    client_id: &str,
    client_secret: &str,
    refresh_token: &str,
    client: &HttpClient,
    retry: &RetryConfig,
) -> Result<TokenResponse> {
    client
        .post(DEFAULT_TOKEN_GCP_URI)
        .form([
            ("grant_type", "refresh_token"),
            ("client_id", client_id),
            ("client_secret", client_secret),
            ("refresh_token", refresh_token),
        ])
        .retryable(retry)
        .idempotent(true)
        .send()
        .await
        .map_err(|source| Error::TokenRequest { source })?
        .into_body()
        .json::<TokenResponse>()
        .await
        .map_err(|source| Error::TokenResponseBody { source })
}

impl AuthorizedUserSigningCredentials {
    pub(crate) fn from(credential: AuthorizedUserCredentials) -> crate::Result<Self> {
        Ok(Self { credential })
    }

    async fn client_email(
        &self,
        client: &HttpClient,
        retry: &RetryConfig,
    ) -> crate::Result<String> {
        let response = get_token_response(
            &self.credential.client_id,
            &self.credential.client_secret,
            &self.credential.refresh_token,
            client,
            retry,
        )
        .await?;

        // Extract email from id_token if available
        if let Some(id_token) = response.id_token {
            // Split the JWT string by dots to get the payload section
            let parts: Vec<&str> = id_token.split('.').collect();
            if parts.len() == 3 {
                // Decode the base64-encoded payload (middle part)
                if let Ok(payload) = BASE64_URL_SAFE_NO_PAD.decode(parts[1]) {
                    // Parse the payload as JSON and extract the email
                    if let Ok(claims) = serde_json::from_slice::<IdTokenClaims>(&payload) {
                        return Ok(claims.email);
                    }
                }
                // If any of the parsing steps fail, fallback to other method
            }
        }

        // Fallback to the original method if id_token is not available or invalid
        let response = client
            .get("https://oauth2.googleapis.com/tokeninfo")
            .query(&[("access_token", response.access_token)])
            .send_retry(retry)
            .await
            .map_err(|source| Error::TokenRequest { source })?
            .into_body()
            .json::<EmailResponse>()
            .await
            .map_err(|source: HttpError| Error::TokenResponseBody { source })?;

        Ok(response.email)
    }
}

#[async_trait]
impl TokenProvider for AuthorizedUserSigningCredentials {
    type Credential = GcpSigningCredential;

    async fn fetch_token(
        &self,
        client: &HttpClient,
        retry: &RetryConfig,
    ) -> crate::Result<TemporaryToken<Arc<GcpSigningCredential>>> {
        let email = self.client_email(client, retry).await?;
        Ok(TemporaryToken {
            token: Arc::new(GcpSigningCredential {
                email,
                private_key: None,
            }),
            expiry: None,
        })
    }
}

#[async_trait]
impl TokenProvider for AuthorizedUserCredentials {
    type Credential = GcpCredential;

    async fn fetch_token(
        &self,
        client: &HttpClient,
        retry: &RetryConfig,
    ) -> crate::Result<TemporaryToken<Arc<GcpCredential>>> {
        let response = get_token_response(
            &self.client_id,
            &self.client_secret,
            &self.refresh_token,
            client,
            retry,
        )
        .await?;

        Ok(TemporaryToken {
            token: Arc::new(GcpCredential {
                bearer: response.access_token,
            }),
            expiry: Some(Instant::now() + Duration::from_secs(response.expires_in)),
        })
    }
}

/// Normalize a header value for the canonical request: trim leading/trailing whitespace and
/// collapse runs of internal whitespace to a single space, as required by the canonical request
/// specification.
///
/// <https://cloud.google.com/storage/docs/authentication/canonical-requests#about-headers>
fn trim_header_value(value: &str) -> String {
    value.split_whitespace().collect::<Vec<_>>().join(" ")
}

/// A Google Cloud Storage Authorizer for generating signed URL using [Google SigV4]
///
/// [Google SigV4]: https://cloud.google.com/storage/docs/access-control/signed-urls
#[derive(Debug)]
pub(crate) struct GCSAuthorizer {
    date: Option<DateTime<Utc>>,
    credential: Arc<GcpSigningCredential>,
}

impl GCSAuthorizer {
    /// Create a new [`GCSAuthorizer`]
    pub(crate) fn new(credential: Arc<GcpSigningCredential>) -> Self {
        Self {
            date: None,
            credential,
        }
    }

    /// Generate a signed URL, additionally folding `extra_query` parameters and `signed_headers`
    /// into the [GOOG4 signed URLs] signature.
    ///
    /// `extra_query` lets callers sign query parameters the recipient must send (e.g. an upload
    /// `partNumber`/`uploadId`), and `signed_headers` binds request headers (e.g. `content-type`)
    /// to the signature. For a signed URL these values are fixed at signing time. The mandatory
    /// `host` header is always signed.
    ///
    /// [GOOG4 signed URLs]: https://cloud.google.com/storage/docs/access-control/signed-urls
    #[allow(clippy::too_many_arguments)]
    pub(crate) async fn sign_with(
        &self,
        crypto: &dyn CryptoProvider,
        method: Method,
        url: &mut Url,
        extra_query: &[(String, String)],
        signed_headers: &HeaderMap,
        expires_in: Duration,
        client: &GoogleCloudStorageClient,
    ) -> crate::Result<()> {
        let email = &self.credential.email;
        let date = self.date.unwrap_or_else(Utc::now);
        let scope = self.scope(date);
        let credential_with_scope = format!("{email}/{scope}");

        // Append any caller-provided query parameters before signing so they are folded into
        // the canonical query string and become part of the signature.
        append_strict_query_pairs(url, extra_query);

        // The `host` header is always signed; callers may bind additional headers whose values
        // are committed to the signature at signing time.
        let mut headers = HeaderMap::with_capacity(1 + signed_headers.len());
        headers.insert("host", DEFAULT_GCS_SIGN_BLOB_HOST.parse().unwrap());
        for (name, value) in signed_headers {
            headers.append(name.clone(), value.clone());
        }

        let (_, signed_headers) = Self::canonicalize_headers(&headers);

        url.query_pairs_mut()
            .append_pair("X-Goog-Algorithm", "GOOG4-RSA-SHA256")
            .append_pair("X-Goog-Credential", &credential_with_scope)
            .append_pair("X-Goog-Date", &date.format("%Y%m%dT%H%M%SZ").to_string())
            .append_pair("X-Goog-Expires", &expires_in.as_secs().to_string())
            .append_pair("X-Goog-SignedHeaders", &signed_headers);

        let string_to_sign = self.string_to_sign(crypto, date, &method, url, &headers)?;
        let signature = match &self.credential.private_key {
            Some(key) => hex_encode(&key.sign(string_to_sign.as_bytes())?),
            None => client.sign_blob(&string_to_sign, email).await?,
        };

        url.query_pairs_mut()
            .append_pair("X-Goog-Signature", &signature);
        Ok(())
    }

    /// Get scope for the request
    ///
    /// <https://cloud.google.com/storage/docs/authentication/signatures#credential-scope>
    fn scope(&self, date: DateTime<Utc>) -> String {
        format!("{}/auto/storage/goog4_request", date.format("%Y%m%d"),)
    }

    /// Canonicalizes query parameters into the GCP canonical form
    /// form like:
    ///```plaintext
    ///HTTP_VERB
    ///PATH_TO_RESOURCE
    ///CANONICAL_QUERY_STRING
    ///CANONICAL_HEADERS
    ///
    ///SIGNED_HEADERS
    ///PAYLOAD
    ///```
    ///
    /// <https://cloud.google.com/storage/docs/authentication/canonical-requests>
    fn canonicalize_request(url: &Url, method: &Method, headers: &HeaderMap) -> String {
        let verb = method.as_str();
        let path = url.path();
        let query = Self::canonicalize_query(url);
        let (canonical_headers, signed_headers) = Self::canonicalize_headers(headers);

        format!(
            "{verb}\n{path}\n{query}\n{canonical_headers}\n\n{signed_headers}\n{DEFAULT_GCS_PLAYLOAD_STRING}"
        )
    }

    /// Canonicalizes query parameters into the GCP canonical form
    /// form like `max-keys=2&prefix=object`
    ///
    /// Parameters are sorted by encoded name, with ties broken by encoded value, as required by
    /// the canonical request specification — sorting by the decoded name (or ignoring the value
    /// for duplicate names) can order parameters differently from how the server verifies them.
    ///
    /// <https://cloud.google.com/storage/docs/authentication/canonical-requests#about-query-strings>
    fn canonicalize_query(url: &Url) -> String {
        url.query_pairs()
            .map(|(k, v)| {
                (
                    utf8_percent_encode(k.as_ref(), &STRICT_ENCODE_SET).to_string(),
                    utf8_percent_encode(v.as_ref(), &STRICT_ENCODE_SET).to_string(),
                )
            })
            .sorted_unstable()
            .map(|(k, v)| format!("{k}={v}"))
            .join("&")
    }

    /// Canonicalizes header into the GCP canonical form
    ///
    /// <https://cloud.google.com/storage/docs/authentication/canonical-requests#about-headers>
    fn canonicalize_headers(header_map: &HeaderMap) -> (String, String) {
        //FIXME add error handling for invalid header values
        let mut headers = BTreeMap::<String, Vec<&str>>::new();
        for (k, v) in header_map {
            headers
                .entry(k.as_str().to_lowercase())
                .or_default()
                .push(std::str::from_utf8(v.as_bytes()).unwrap());
        }

        let canonicalize_headers = headers
            .iter()
            .map(|(k, v)| {
                format!(
                    "{}:{}",
                    k.trim(),
                    v.iter().map(|v| trim_header_value(v)).join(",")
                )
            })
            .join("\n");

        let signed_headers = headers.keys().join(";");

        (canonicalize_headers, signed_headers)
    }

    ///construct the string to sign
    ///form like:
    ///```plaintext
    ///SIGNING_ALGORITHM
    ///ACTIVE_DATETIME
    ///CREDENTIAL_SCOPE
    ///HASHED_CANONICAL_REQUEST
    ///```
    ///`ACTIVE_DATETIME` format:`YYYYMMDD'T'HHMMSS'Z'`
    /// <https://cloud.google.com/storage/docs/authentication/signatures#string-to-sign>
    pub(crate) fn string_to_sign(
        &self,
        crypto: &dyn CryptoProvider,
        date: DateTime<Utc>,
        request_method: &Method,
        url: &Url,
        headers: &HeaderMap,
    ) -> crate::Result<String> {
        let canonical_request = Self::canonicalize_request(url, request_method, headers);
        let hashed_canonical_req = hex_digest(crypto, canonical_request.as_bytes())?;
        let scope = self.scope(date);

        Ok(format!(
            "{}\n{}\n{}\n{}",
            "GOOG4-RSA-SHA256",
            date.format("%Y%m%dT%H%M%SZ"),
            scope,
            hashed_canonical_req
        ))
    }
}

pub(crate) trait CredentialExt {
    /// Apply bearer authentication to the request if the credential is not None
    fn with_bearer_auth(self, credential: Option<&GcpCredential>) -> Self;
}

impl CredentialExt for HttpRequestBuilder {
    fn with_bearer_auth(self, credential: Option<&GcpCredential>) -> Self {
        match credential {
            Some(credential) => {
                if credential.bearer.is_empty() {
                    self
                } else {
                    self.bearer_auth(&credential.bearer)
                }
            }
            None => self,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::client::{
        ClientOptions, DigestAlgorithm, DigestContext, HmacContext, StaticCredentialProvider,
    };
    use crate::gcp::client::{GoogleCloudStorageClient, GoogleCloudStorageConfig};
    use http::{HeaderName, HeaderValue};

    const SIGNATURE_BYTES: &[u8] = &[0x00, 0x01, 0x02, 0xab, 0xcd];

    struct FixedSigner;

    impl Signer for FixedSigner {
        fn sign(&self, _string_to_sign: &[u8]) -> crate::Result<Vec<u8>> {
            Ok(SIGNATURE_BYTES.to_vec())
        }
    }

    #[derive(Debug)]
    struct FixedCryptoProvider;

    impl CryptoProvider for FixedCryptoProvider {
        fn digest(&self, _algorithm: DigestAlgorithm) -> crate::Result<Box<dyn DigestContext>> {
            Ok(Box::new(FixedDigestContext))
        }

        fn hmac(
            &self,
            _algorithm: DigestAlgorithm,
            _secret: &[u8],
        ) -> crate::Result<Box<dyn HmacContext>> {
            panic!("GCS signed URL should not use HMAC")
        }

        fn sign(
            &self,
            _algorithm: SigningAlgorithm,
            _pem: &[u8],
        ) -> crate::Result<Box<dyn Signer>> {
            Ok(Box::new(FixedSigner))
        }
    }

    struct FixedDigestContext;

    impl DigestContext for FixedDigestContext {
        fn update(&mut self, _data: &[u8]) {}

        fn finish(&mut self) -> crate::Result<&[u8]> {
            Ok(&[0x12, 0x34])
        }
    }

    #[derive(Debug)]
    struct UnusedHttpService;

    #[async_trait::async_trait]
    impl crate::client::HttpService for UnusedHttpService {
        async fn call(
            &self,
            _req: crate::client::HttpRequest,
        ) -> std::result::Result<crate::client::HttpResponse, HttpError> {
            panic!("SelfSignedJwt should not make HTTP requests")
        }
    }

    #[test]
    fn self_signed_jwt_base64url_encodes_raw_signature_bytes() {
        let jwt = SelfSignedJwt::new(
            "key-id".into(),
            "service-account@example.com".into(),
            ServiceAccountKey::new(Box::new(FixedSigner)),
            DEFAULT_SCOPE.to_string(),
        )
        .unwrap();
        let client = HttpClient::new(UnusedHttpService);
        let token = futures_executor::block_on(jwt.fetch_token(&client, &RetryConfig::default()))
            .unwrap()
            .token;
        let signature = token.bearer.rsplit('.').next().unwrap();

        assert_eq!(signature, BASE64_URL_SAFE_NO_PAD.encode(SIGNATURE_BYTES));
        assert_ne!(
            signature,
            BASE64_URL_SAFE_NO_PAD.encode(hex_encode(SIGNATURE_BYTES))
        );
    }

    #[test]
    fn signed_url_hex_encodes_local_signature_bytes() {
        let signing_credential = Arc::new(GcpSigningCredential {
            email: "service-account@example.com".into(),
            private_key: Some(ServiceAccountKey::new(Box::new(FixedSigner))),
        });
        let authorizer = GCSAuthorizer::new(Arc::clone(&signing_credential));
        let config = GoogleCloudStorageConfig {
            base_url: DEFAULT_GCS_BASE_URL.into(),
            credentials: Arc::new(StaticCredentialProvider::new(GcpCredential {
                bearer: "bearer".into(),
            })),
            signing_credentials: Arc::new(StaticCredentialProvider::new(GcpSigningCredential {
                email: "service-account@example.com".into(),
                private_key: None,
            })),
            crypto: None,
            bucket_name: "bucket".into(),
            retry_config: RetryConfig::default(),
            client_options: ClientOptions::default(),
            skip_signature: false,
        };
        let client =
            GoogleCloudStorageClient::new(config, HttpClient::new(UnusedHttpService)).unwrap();
        let mut url = Url::parse("https://storage.googleapis.com/bucket/object").unwrap();

        futures_executor::block_on(authorizer.sign_with(
            &FixedCryptoProvider,
            Method::GET,
            &mut url,
            &[],
            &HeaderMap::new(),
            Duration::from_secs(60),
            &client,
        ))
        .unwrap();

        let signature = url
            .query_pairs()
            .find(|(key, _)| key == "X-Goog-Signature")
            .unwrap()
            .1;
        assert_eq!(signature, hex_encode(SIGNATURE_BYTES));
    }

    #[test]
    fn signed_url_with_folds_query_and_headers() {
        let signing_credential = Arc::new(GcpSigningCredential {
            email: "service-account@example.com".into(),
            private_key: Some(ServiceAccountKey::new(Box::new(FixedSigner))),
        });
        let authorizer = GCSAuthorizer::new(Arc::clone(&signing_credential));
        let config = GoogleCloudStorageConfig {
            base_url: DEFAULT_GCS_BASE_URL.into(),
            credentials: Arc::new(StaticCredentialProvider::new(GcpCredential {
                bearer: "bearer".into(),
            })),
            signing_credentials: Arc::new(StaticCredentialProvider::new(GcpSigningCredential {
                email: "service-account@example.com".into(),
                private_key: None,
            })),
            crypto: None,
            bucket_name: "bucket".into(),
            retry_config: RetryConfig::default(),
            client_options: ClientOptions::default(),
            skip_signature: false,
        };
        let client =
            GoogleCloudStorageClient::new(config, HttpClient::new(UnusedHttpService)).unwrap();
        let mut url = Url::parse("https://storage.googleapis.com/bucket/object").unwrap();

        let mut signed_headers = HeaderMap::new();
        signed_headers.insert(
            HeaderName::from_static("content-type"),
            HeaderValue::from_static("text/plain"),
        );

        futures_executor::block_on(authorizer.sign_with(
            &FixedCryptoProvider,
            Method::PUT,
            &mut url,
            &[
                ("partNumber".to_string(), "1".to_string()),
                ("uploadId".to_string(), "abc123".to_string()),
            ],
            &signed_headers,
            Duration::from_secs(60),
            &client,
        ))
        .unwrap();

        let pairs: std::collections::HashMap<_, _> = url.query_pairs().into_owned().collect();
        // Caller-provided query parameters are present in the signed URL.
        assert_eq!(pairs.get("partNumber").map(String::as_str), Some("1"));
        assert_eq!(pairs.get("uploadId").map(String::as_str), Some("abc123"));
        // The extra header is reflected in X-Goog-SignedHeaders, sorted alongside `host`.
        assert_eq!(
            pairs.get("X-Goog-SignedHeaders").map(String::as_str),
            Some("content-type;host")
        );
        assert!(pairs.contains_key("X-Goog-Signature"));
    }

    #[test]
    fn test_canonicalize_headers() {
        let mut input_header = HeaderMap::new();
        input_header.insert("content-type", "text/plain".parse().unwrap());
        input_header.insert("host", "storage.googleapis.com".parse().unwrap());
        input_header.insert("x-goog-meta-reviewer", "jane".parse().unwrap());
        input_header.append("x-goog-meta-reviewer", "john".parse().unwrap());
        assert_eq!(
            GCSAuthorizer::canonicalize_headers(&input_header),
            (
                "content-type:text/plain
host:storage.googleapis.com
x-goog-meta-reviewer:jane,john"
                    .into(),
                "content-type;host;x-goog-meta-reviewer".to_string()
            )
        );
    }

    #[test]
    fn test_canonicalize_query() {
        let mut url = Url::parse("https://storage.googleapis.com/bucket/object").unwrap();
        url.query_pairs_mut()
            .append_pair("max-keys", "2")
            .append_pair("prefix", "object");
        assert_eq!(
            GCSAuthorizer::canonicalize_query(&url),
            "max-keys=2&prefix=object".to_string()
        );
    }

    #[test]
    fn trim_header_value_collapses_internal_whitespace() {
        // The canonical request collapses runs of whitespace to a single space and trims the
        // ends, rather than stripping all whitespace, so a signed `content-type` with parameters
        // matches the value the recipient sends.
        assert_eq!(trim_header_value("  foo   bar  "), "foo bar");
        assert_eq!(
            trim_header_value("text/plain;  charset=utf-8"),
            "text/plain; charset=utf-8"
        );
        assert_eq!(trim_header_value("single"), "single");
    }
}