| // 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::UserDelegationKey; |
| use crate::RetryConfig; |
| use crate::azure::STORE; |
| use crate::client::builder::{HttpRequestBuilder, add_query_pairs}; |
| use crate::client::retry::RetryExt; |
| use crate::client::token::{TemporaryToken, TokenCache}; |
| use crate::client::{ |
| CredentialProvider, CryptoProvider, DigestAlgorithm, HttpClient, HttpError, HttpRequest, |
| TokenProvider, crypto_provider, |
| }; |
| use async_trait::async_trait; |
| use base64::Engine; |
| use base64::prelude::{BASE64_STANDARD, BASE64_URL_SAFE_NO_PAD}; |
| use chrono::{DateTime, SecondsFormat, Utc}; |
| use http::Method; |
| use http::header::{ |
| ACCEPT, AUTHORIZATION, CONTENT_ENCODING, CONTENT_LANGUAGE, CONTENT_LENGTH, CONTENT_TYPE, DATE, |
| HeaderMap, HeaderName, HeaderValue, IF_MATCH, IF_MODIFIED_SINCE, IF_NONE_MATCH, |
| IF_UNMODIFIED_SINCE, RANGE, |
| }; |
| use serde::Deserialize; |
| use std::borrow::Cow; |
| use std::collections::HashMap; |
| use std::fmt::Debug; |
| use std::ops::Deref; |
| use std::process::Command; |
| use std::str; |
| use std::sync::Arc; |
| use std::time::{Duration, Instant, SystemTime}; |
| use url::Url; |
| |
| static AZURE_VERSION: HeaderValue = HeaderValue::from_static("2023-11-03"); |
| pub(crate) static VERSION: HeaderName = HeaderName::from_static("x-ms-version"); |
| pub(crate) static BLOB_TYPE: HeaderName = HeaderName::from_static("x-ms-blob-type"); |
| pub(crate) static COPY_SOURCE: HeaderName = HeaderName::from_static("x-ms-copy-source"); |
| static CONTENT_MD5: HeaderName = HeaderName::from_static("content-md5"); |
| static PARTNER_TOKEN: HeaderName = HeaderName::from_static("x-ms-partner-token"); |
| static CLUSTER_IDENTIFIER: HeaderName = HeaderName::from_static("x-ms-cluster-identifier"); |
| static WORKLOAD_RESOURCE: HeaderName = HeaderName::from_static("x-ms-workload-resource-moniker"); |
| static PROXY_HOST: HeaderName = HeaderName::from_static("x-ms-proxy-host"); |
| pub(crate) const RFC1123_FMT: &str = "%a, %d %h %Y %T GMT"; |
| const CONTENT_TYPE_JSON: &str = "application/json"; |
| const MSI_SECRET_ENV_KEY: &str = "IDENTITY_HEADER"; |
| const MSI_API_VERSION: &str = "2019-08-01"; |
| const TOKEN_MIN_TTL: u64 = 300; |
| |
| /// OIDC scope used when interacting with OAuth2 APIs |
| /// |
| /// <https://learn.microsoft.com/en-us/azure/active-directory/develop/scopes-oidc#the-default-scope> |
| const AZURE_STORAGE_SCOPE: &str = "https://storage.azure.com/.default"; |
| |
| /// Resource ID used when obtaining an access token from the metadata endpoint |
| /// |
| /// <https://learn.microsoft.com/en-us/azure/storage/blobs/authorize-access-azure-active-directory#microsoft-authentication-library-msal> |
| const AZURE_STORAGE_RESOURCE: &str = "https://storage.azure.com"; |
| |
| #[derive(Debug, thiserror::Error)] |
| pub enum Error { |
| #[error("Error performing token request: {}", source)] |
| TokenRequest { |
| source: crate::client::retry::RetryError, |
| }, |
| |
| #[error("Error getting token response body: {}", source)] |
| TokenResponseBody { source: HttpError }, |
| |
| #[error("Error reading federated token file ")] |
| FederatedTokenFile, |
| |
| #[error("Invalid Access Key: {}", source)] |
| InvalidAccessKey { source: base64::DecodeError }, |
| |
| #[error("'az account get-access-token' command failed: {message}")] |
| AzureCli { message: String }, |
| |
| #[error("Failed to parse azure cli response: {source}")] |
| AzureCliResponse { source: serde_json::Error }, |
| |
| #[error("Generating SAS keys with SAS tokens auth is not supported")] |
| SASforSASNotSupported, |
| } |
| |
| pub(crate) type Result<T, E = Error> = std::result::Result<T, E>; |
| |
| impl From<Error> for crate::Error { |
| fn from(value: Error) -> Self { |
| Self::Generic { |
| store: STORE, |
| source: Box::new(value), |
| } |
| } |
| } |
| |
| /// A shared Azure Storage Account Key |
| #[derive(Debug, Clone, Eq, PartialEq)] |
| pub struct AzureAccessKey(Vec<u8>); |
| |
| impl AzureAccessKey { |
| /// Create a new [`AzureAccessKey`], checking it for validity |
| pub fn try_new(key: &str) -> Result<Self> { |
| let key = BASE64_STANDARD |
| .decode(key) |
| .map_err(|source| Error::InvalidAccessKey { source })?; |
| |
| Ok(Self(key)) |
| } |
| } |
| |
| /// An Azure storage credential |
| #[derive(Debug, Eq, PartialEq)] |
| pub enum AzureCredential { |
| /// A shared access key |
| /// |
| /// <https://learn.microsoft.com/en-us/rest/api/storageservices/authorize-with-shared-key> |
| AccessKey(AzureAccessKey), |
| /// A shared access signature |
| /// |
| /// <https://learn.microsoft.com/en-us/rest/api/storageservices/delegate-access-with-shared-access-signature> |
| SASToken(Vec<(String, String)>), |
| /// An authorization token |
| /// |
| /// <https://learn.microsoft.com/en-us/rest/api/storageservices/authorize-with-azure-active-directory> |
| BearerToken(String), |
| } |
| |
| impl AzureCredential { |
| /// Determines if the credential requires the request be treated as sensitive |
| pub fn sensitive_request(&self) -> bool { |
| match self { |
| Self::AccessKey(_) => false, |
| Self::BearerToken(_) => false, |
| // SAS tokens are sent as query parameters in the url |
| Self::SASToken(_) => true, |
| } |
| } |
| } |
| |
| /// A list of known Azure authority hosts |
| pub mod authority_hosts { |
| /// China-based Azure Authority Host |
| pub const AZURE_CHINA: &str = "https://login.chinacloudapi.cn"; |
| /// Germany-based Azure Authority Host |
| pub const AZURE_GERMANY: &str = "https://login.microsoftonline.de"; |
| /// US Government Azure Authority Host |
| pub const AZURE_GOVERNMENT: &str = "https://login.microsoftonline.us"; |
| /// Public Cloud Azure Authority Host |
| pub const AZURE_PUBLIC_CLOUD: &str = "https://login.microsoftonline.com"; |
| } |
| |
| pub(crate) struct AzureSigner { |
| signing_key: AzureAccessKey, |
| start: DateTime<Utc>, |
| end: DateTime<Utc>, |
| account: String, |
| delegation_key: Option<UserDelegationKey>, |
| } |
| |
| impl AzureSigner { |
| pub(crate) fn new( |
| signing_key: AzureAccessKey, |
| account: String, |
| start: DateTime<Utc>, |
| end: DateTime<Utc>, |
| delegation_key: Option<UserDelegationKey>, |
| ) -> Self { |
| Self { |
| signing_key, |
| account, |
| start, |
| end, |
| delegation_key, |
| } |
| } |
| |
| pub(crate) fn sign( |
| &self, |
| crypto: &dyn CryptoProvider, |
| method: &Method, |
| url: &mut Url, |
| ) -> crate::Result<()> { |
| let (str_to_sign, query_pairs) = match &self.delegation_key { |
| Some(delegation_key) => string_to_sign_user_delegation_sas( |
| url, |
| method, |
| &self.account, |
| &self.start, |
| &self.end, |
| delegation_key, |
| ), |
| None => string_to_sign_service_sas(url, method, &self.account, &self.start, &self.end), |
| }; |
| let mut ctx = crypto.hmac(DigestAlgorithm::Sha256, &self.signing_key.0)?; |
| ctx.update(str_to_sign.as_bytes()); |
| let auth = ctx.finish()?; |
| |
| url.query_pairs_mut().extend_pairs(query_pairs); |
| url.query_pairs_mut() |
| .append_pair("sig", BASE64_STANDARD.encode(auth).as_str()); |
| Ok(()) |
| } |
| } |
| |
| fn add_date_and_version_headers(request: &mut HttpRequest) { |
| // rfc2822 string should never contain illegal characters |
| let date = Utc::now(); |
| let date_str = date.format(RFC1123_FMT).to_string(); |
| // we formatted the data string ourselves, so unwrapping should be fine |
| let date_val = HeaderValue::from_str(&date_str).unwrap(); |
| request.headers_mut().insert(DATE, date_val); |
| // Preserve an explicit per-request override when a call site needs a newer |
| // service version than the backend default. |
| if !request.headers().contains_key(&VERSION) { |
| request |
| .headers_mut() |
| .insert(&VERSION, AZURE_VERSION.clone()); |
| } |
| } |
| |
| /// Authorize a [`HttpRequest`] with an [`AzureAuthorizer`] |
| #[derive(Debug)] |
| pub struct AzureAuthorizer<'a> { |
| credential: &'a AzureCredential, |
| crypto: Option<&'a dyn CryptoProvider>, |
| account: &'a str, |
| } |
| |
| impl<'a> AzureAuthorizer<'a> { |
| /// Create a new [`AzureAuthorizer`] |
| pub fn new(credential: &'a AzureCredential, account: &'a str) -> Self { |
| AzureAuthorizer { |
| credential, |
| account, |
| crypto: None, |
| } |
| } |
| |
| /// Specify the crypto provider |
| pub fn with_crypto(mut self, crypto: &'a dyn CryptoProvider) -> Self { |
| self.crypto = Some(crypto); |
| self |
| } |
| |
| fn crypto(&self) -> crate::Result<&dyn CryptoProvider> { |
| crypto_provider(self.crypto) |
| } |
| |
| /// Authorize `request` |
| /// |
| /// # Panics |
| /// |
| /// Panics on cryptography error |
| #[deprecated(note = "use AzureAuthorizer::try_authorize")] |
| pub fn authorize(&self, request: &mut HttpRequest) { |
| self.try_authorize(request).unwrap() |
| } |
| |
| /// Authorize `request` |
| pub fn try_authorize(&self, request: &mut HttpRequest) -> crate::Result<()> { |
| add_date_and_version_headers(request); |
| |
| match self.credential { |
| AzureCredential::AccessKey(key) => { |
| let crypto = self.crypto()?; |
| let url = Url::parse(&request.uri().to_string()).unwrap(); |
| let signature = generate_authorization( |
| crypto, |
| request.headers(), |
| &url, |
| request.method(), |
| self.account, |
| key, |
| )?; |
| |
| // "signature" is a base 64 encoded string so it should never |
| // contain illegal characters |
| request.headers_mut().append( |
| AUTHORIZATION, |
| HeaderValue::from_str(signature.as_str()).unwrap(), |
| ); |
| } |
| AzureCredential::BearerToken(token) => { |
| request.headers_mut().append( |
| AUTHORIZATION, |
| HeaderValue::from_str(format!("Bearer {token}").as_str()).unwrap(), |
| ); |
| } |
| AzureCredential::SASToken(query_pairs) => { |
| add_query_pairs(request.uri_mut(), query_pairs); |
| } |
| } |
| Ok(()) |
| } |
| } |
| |
| pub(crate) trait CredentialExt: Sized { |
| /// Apply authorization to requests against azure storage accounts |
| /// <https://docs.microsoft.com/en-us/rest/api/storageservices/authorize-requests-to-azure-storage> |
| fn with_azure_authorization( |
| self, |
| crypto: Option<&dyn CryptoProvider>, |
| credential: &Option<impl Deref<Target = AzureCredential>>, |
| account: &str, |
| ) -> crate::Result<Self>; |
| } |
| |
| impl CredentialExt for HttpRequestBuilder { |
| fn with_azure_authorization( |
| self, |
| crypto: Option<&dyn CryptoProvider>, |
| credential: &Option<impl Deref<Target = AzureCredential>>, |
| account: &str, |
| ) -> crate::Result<Self> { |
| let (client, request) = self.into_parts(); |
| let mut request = request.expect("request valid"); |
| |
| match credential.as_deref() { |
| Some(credential) => { |
| let mut authorizer = AzureAuthorizer::new(credential, account); |
| if let Some(crypto) = crypto { |
| authorizer = authorizer.with_crypto(crypto); |
| } |
| authorizer.try_authorize(&mut request)?; |
| } |
| None => { |
| add_date_and_version_headers(&mut request); |
| } |
| } |
| |
| Ok(Self::from_parts(client, request)) |
| } |
| } |
| |
| /// Generate signed key for authorization via access keys |
| /// <https://docs.microsoft.com/en-us/rest/api/storageservices/authorize-with-shared-key> |
| fn generate_authorization( |
| crypto: &dyn CryptoProvider, |
| h: &HeaderMap, |
| u: &Url, |
| method: &Method, |
| account: &str, |
| key: &AzureAccessKey, |
| ) -> crate::Result<String> { |
| let str_to_sign = string_to_sign(h, u, method, account); |
| let mut ctx = crypto.hmac(DigestAlgorithm::Sha256, &key.0)?; |
| ctx.update(str_to_sign.as_bytes()); |
| let auth = ctx.finish()?; |
| Ok(format!( |
| "SharedKey {}:{}", |
| account, |
| BASE64_STANDARD.encode(auth) |
| )) |
| } |
| |
| fn add_if_exists<'a>(h: &'a HeaderMap, key: &HeaderName) -> &'a str { |
| h.get(key) |
| .map(|s| s.to_str()) |
| .transpose() |
| .ok() |
| .flatten() |
| .unwrap_or_default() |
| } |
| |
| fn string_to_sign_sas( |
| u: &Url, |
| method: &Method, |
| account: &str, |
| start: &DateTime<Utc>, |
| end: &DateTime<Utc>, |
| ) -> (String, String, String, String, String) { |
| // NOTE: for now only blob signing is supported. |
| let signed_resource = "b".to_string(); |
| |
| // https://learn.microsoft.com/en-us/rest/api/storageservices/create-service-sas#permissions-for-a-directory-container-or-blob |
| let signed_permissions = match *method { |
| // read and list permissions |
| Method::GET => match signed_resource.as_str() { |
| "c" => "rl", |
| "b" => "r", |
| _ => unreachable!(), |
| }, |
| // write permissions (also allows crating a new blob in a sub-key) |
| Method::PUT => "w", |
| // delete permissions |
| Method::DELETE => "d", |
| // other methods are not used in any of the current operations |
| _ => "", |
| } |
| .to_string(); |
| let signed_start = start.to_rfc3339_opts(SecondsFormat::Secs, true); |
| let signed_expiry = end.to_rfc3339_opts(SecondsFormat::Secs, true); |
| let canonicalized_resource = if u.host_str().unwrap_or_default().contains(account) { |
| format!("/blob/{}{}", account, u.path()) |
| } else { |
| // NOTE: in case of the emulator, the account name is not part of the host |
| // but the path starts with the account name |
| format!("/blob{}", u.path()) |
| }; |
| |
| ( |
| signed_resource, |
| signed_permissions, |
| signed_start, |
| signed_expiry, |
| canonicalized_resource, |
| ) |
| } |
| |
| /// Create a string to be signed for authorization via [service sas]. |
| /// |
| /// [service sas]: https://learn.microsoft.com/en-us/rest/api/storageservices/create-service-sas#version-2020-12-06-and-later |
| fn string_to_sign_service_sas( |
| u: &Url, |
| method: &Method, |
| account: &str, |
| start: &DateTime<Utc>, |
| end: &DateTime<Utc>, |
| ) -> (String, HashMap<&'static str, String>) { |
| let (signed_resource, signed_permissions, signed_start, signed_expiry, canonicalized_resource) = |
| string_to_sign_sas(u, method, account, start, end); |
| |
| let string_to_sign = format!( |
| "{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}", |
| signed_permissions, |
| signed_start, |
| signed_expiry, |
| canonicalized_resource, |
| "", // signed identifier |
| "", // signed ip |
| "", // signed protocol |
| AZURE_VERSION.to_str().unwrap(), // signed version |
| signed_resource, // signed resource |
| "", // signed snapshot time |
| "", // signed encryption scope |
| "", // rscc - response header: Cache-Control |
| "", // rscd - response header: Content-Disposition |
| "", // rsce - response header: Content-Encoding |
| "", // rscl - response header: Content-Language |
| "", // rsct - response header: Content-Type |
| ); |
| |
| let mut pairs = HashMap::new(); |
| pairs.insert("sv", AZURE_VERSION.to_str().unwrap().to_string()); |
| pairs.insert("sp", signed_permissions); |
| pairs.insert("st", signed_start); |
| pairs.insert("se", signed_expiry); |
| pairs.insert("sr", signed_resource); |
| |
| (string_to_sign, pairs) |
| } |
| |
| /// Create a string to be signed for authorization via [user delegation sas]. |
| /// |
| /// [user delegation sas]: https://learn.microsoft.com/en-us/rest/api/storageservices/create-user-delegation-sas#version-2020-12-06-and-later |
| fn string_to_sign_user_delegation_sas( |
| u: &Url, |
| method: &Method, |
| account: &str, |
| start: &DateTime<Utc>, |
| end: &DateTime<Utc>, |
| delegation_key: &UserDelegationKey, |
| ) -> (String, HashMap<&'static str, String>) { |
| let (signed_resource, signed_permissions, signed_start, signed_expiry, canonicalized_resource) = |
| string_to_sign_sas(u, method, account, start, end); |
| |
| let string_to_sign = format!( |
| "{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}", |
| signed_permissions, |
| signed_start, |
| signed_expiry, |
| canonicalized_resource, |
| delegation_key.signed_oid, // signed key object id |
| delegation_key.signed_tid, // signed key tenant id |
| delegation_key.signed_start, // signed key start |
| delegation_key.signed_expiry, // signed key expiry |
| delegation_key.signed_service, // signed key service |
| delegation_key.signed_version, // signed key version |
| "", // signed authorized user object id |
| "", // signed unauthorized user object id |
| "", // signed correlation id |
| "", // signed ip |
| "", // signed protocol |
| AZURE_VERSION.to_str().unwrap(), // signed version |
| signed_resource, // signed resource |
| "", // signed snapshot time |
| "", // signed encryption scope |
| "", // rscc - response header: Cache-Control |
| "", // rscd - response header: Content-Disposition |
| "", // rsce - response header: Content-Encoding |
| "", // rscl - response header: Content-Language |
| "", // rsct - response header: Content-Type |
| ); |
| |
| let mut pairs = HashMap::new(); |
| pairs.insert("sv", AZURE_VERSION.to_str().unwrap().to_string()); |
| pairs.insert("sp", signed_permissions); |
| pairs.insert("st", signed_start); |
| pairs.insert("se", signed_expiry); |
| pairs.insert("sr", signed_resource); |
| pairs.insert("skoid", delegation_key.signed_oid.clone()); |
| pairs.insert("sktid", delegation_key.signed_tid.clone()); |
| pairs.insert("skt", delegation_key.signed_start.clone()); |
| pairs.insert("ske", delegation_key.signed_expiry.clone()); |
| pairs.insert("sks", delegation_key.signed_service.clone()); |
| pairs.insert("skv", delegation_key.signed_version.clone()); |
| |
| (string_to_sign, pairs) |
| } |
| |
| /// <https://docs.microsoft.com/en-us/rest/api/storageservices/authorize-with-shared-key#constructing-the-signature-string> |
| fn string_to_sign(h: &HeaderMap, u: &Url, method: &Method, account: &str) -> String { |
| // content length must only be specified if != 0 |
| // this is valid from 2015-02-21 |
| let content_length = h |
| .get(&CONTENT_LENGTH) |
| .map(|s| s.to_str()) |
| .transpose() |
| .ok() |
| .flatten() |
| .filter(|&v| v != "0") |
| .unwrap_or_default(); |
| format!( |
| "{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}{}", |
| method.as_ref(), |
| add_if_exists(h, &CONTENT_ENCODING), |
| add_if_exists(h, &CONTENT_LANGUAGE), |
| content_length, |
| add_if_exists(h, &CONTENT_MD5), |
| add_if_exists(h, &CONTENT_TYPE), |
| add_if_exists(h, &DATE), |
| add_if_exists(h, &IF_MODIFIED_SINCE), |
| add_if_exists(h, &IF_MATCH), |
| add_if_exists(h, &IF_NONE_MATCH), |
| add_if_exists(h, &IF_UNMODIFIED_SINCE), |
| add_if_exists(h, &RANGE), |
| canonicalize_header(h), |
| canonicalize_resource(account, u) |
| ) |
| } |
| |
| /// <https://docs.microsoft.com/en-us/rest/api/storageservices/authorize-with-shared-key#constructing-the-canonicalized-headers-string> |
| fn canonicalize_header(headers: &HeaderMap) -> String { |
| let mut names = headers |
| .iter() |
| .filter(|&(k, _)| k.as_str().starts_with("x-ms")) |
| // TODO remove unwraps |
| .map(|(k, _)| (k.as_str(), headers.get(k).unwrap().to_str().unwrap())) |
| .collect::<Vec<_>>(); |
| names.sort_unstable(); |
| |
| let mut result = String::new(); |
| for (name, value) in names { |
| result.push_str(name); |
| result.push(':'); |
| result.push_str(value); |
| result.push('\n'); |
| } |
| result |
| } |
| |
| /// <https://docs.microsoft.com/en-us/rest/api/storageservices/authorize-with-shared-key#constructing-the-canonicalized-resource-string> |
| fn canonicalize_resource(account: &str, uri: &Url) -> String { |
| let mut can_res: String = String::new(); |
| can_res.push('/'); |
| can_res.push_str(account); |
| can_res.push_str(uri.path().to_string().as_str()); |
| can_res.push('\n'); |
| |
| // query parameters |
| let query_pairs = uri.query_pairs(); |
| { |
| let mut qps: Vec<String> = Vec::new(); |
| for (q, _) in query_pairs { |
| if !(qps.iter().any(|x| x == &*q)) { |
| qps.push(q.into_owned()); |
| } |
| } |
| |
| qps.sort(); |
| |
| for qparam in qps { |
| // find correct parameter |
| let ret = lexy_sort(query_pairs, &qparam); |
| |
| can_res = can_res + &qparam.to_lowercase() + ":"; |
| |
| for (i, item) in ret.iter().enumerate() { |
| if i > 0 { |
| can_res.push(','); |
| } |
| can_res.push_str(item); |
| } |
| |
| can_res.push('\n'); |
| } |
| }; |
| |
| can_res[0..can_res.len() - 1].to_owned() |
| } |
| |
| fn lexy_sort<'a>( |
| vec: impl Iterator<Item = (Cow<'a, str>, Cow<'a, str>)> + 'a, |
| query_param: &str, |
| ) -> Vec<Cow<'a, str>> { |
| let mut values = vec |
| .filter(|(k, _)| *k == query_param) |
| .map(|(_, v)| v) |
| .collect::<Vec<_>>(); |
| values.sort_unstable(); |
| values |
| } |
| |
| /// <https://learn.microsoft.com/en-us/azure/active-directory/develop/v2-oauth2-client-creds-grant-flow#successful-response-1> |
| #[derive(Deserialize, Debug)] |
| struct OAuthTokenResponse { |
| access_token: String, |
| expires_in: u64, |
| } |
| |
| /// Encapsulates the logic to perform an OAuth token challenge |
| /// |
| /// <https://learn.microsoft.com/en-us/azure/active-directory/develop/v2-oauth2-client-creds-grant-flow#first-case-access-token-request-with-a-shared-secret> |
| #[derive(Debug)] |
| pub(crate) struct ClientSecretOAuthProvider { |
| token_url: String, |
| client_id: String, |
| client_secret: String, |
| } |
| |
| impl ClientSecretOAuthProvider { |
| /// Create a new [`ClientSecretOAuthProvider`] for an azure backed store |
| pub(crate) fn new( |
| client_id: String, |
| client_secret: String, |
| tenant_id: impl AsRef<str>, |
| authority_host: Option<String>, |
| ) -> Self { |
| let authority_host = |
| authority_host.unwrap_or_else(|| authority_hosts::AZURE_PUBLIC_CLOUD.to_owned()); |
| |
| Self { |
| token_url: format!( |
| "{}/{}/oauth2/v2.0/token", |
| authority_host, |
| tenant_id.as_ref() |
| ), |
| client_id, |
| client_secret, |
| } |
| } |
| } |
| |
| #[async_trait::async_trait] |
| impl TokenProvider for ClientSecretOAuthProvider { |
| type Credential = AzureCredential; |
| |
| /// Fetch a token |
| async fn fetch_token( |
| &self, |
| client: &HttpClient, |
| retry: &RetryConfig, |
| ) -> crate::Result<TemporaryToken<Arc<AzureCredential>>> { |
| let response: OAuthTokenResponse = client |
| .request(Method::POST, &self.token_url) |
| .header(ACCEPT, HeaderValue::from_static(CONTENT_TYPE_JSON)) |
| .form([ |
| ("client_id", self.client_id.as_str()), |
| ("client_secret", self.client_secret.as_str()), |
| ("scope", AZURE_STORAGE_SCOPE), |
| ("grant_type", "client_credentials"), |
| ]) |
| .retryable(retry) |
| .idempotent(true) |
| .send() |
| .await |
| .map_err(|source| Error::TokenRequest { source })? |
| .into_body() |
| .json() |
| .await |
| .map_err(|source| Error::TokenResponseBody { source })?; |
| |
| Ok(TemporaryToken { |
| token: Arc::new(AzureCredential::BearerToken(response.access_token)), |
| expiry: Some(Instant::now() + Duration::from_secs(response.expires_in)), |
| }) |
| } |
| } |
| |
| fn expires_on_string<'de, D>(deserializer: D) -> std::result::Result<Instant, D::Error> |
| where |
| D: serde::de::Deserializer<'de>, |
| { |
| let v = String::deserialize(deserializer)?; |
| let v = v.parse::<u64>().map_err(serde::de::Error::custom)?; |
| let now = SystemTime::now() |
| .duration_since(SystemTime::UNIX_EPOCH) |
| .map_err(serde::de::Error::custom)?; |
| |
| Ok(Instant::now() + Duration::from_secs(v.saturating_sub(now.as_secs()))) |
| } |
| |
| /// NOTE: expires_on is a String version of unix epoch time, not an integer. |
| /// <https://learn.microsoft.com/en-gb/azure/active-directory/managed-identities-azure-resources/how-to-use-vm-token#get-a-token-using-http> |
| /// <https://learn.microsoft.com/en-us/azure/app-service/overview-managed-identity?tabs=portal%2Chttp#connect-to-azure-services-in-app-code> |
| #[derive(Debug, Clone, Deserialize)] |
| struct ImdsTokenResponse { |
| pub access_token: String, |
| #[serde(deserialize_with = "expires_on_string")] |
| pub expires_on: Instant, |
| } |
| |
| /// Attempts authentication using a managed identity that has been assigned to the deployment environment. |
| /// |
| /// This authentication type works in Azure VMs, App Service and Azure Functions applications, as well as the Azure Cloud Shell |
| /// <https://learn.microsoft.com/en-gb/azure/active-directory/managed-identities-azure-resources/how-to-use-vm-token#get-a-token-using-http> |
| #[derive(Debug)] |
| pub(crate) struct ImdsManagedIdentityProvider { |
| msi_endpoint: String, |
| client_id: Option<String>, |
| object_id: Option<String>, |
| msi_res_id: Option<String>, |
| } |
| |
| impl ImdsManagedIdentityProvider { |
| /// Create a new [`ImdsManagedIdentityProvider`] for an azure backed store |
| pub(crate) fn new( |
| client_id: Option<String>, |
| object_id: Option<String>, |
| msi_res_id: Option<String>, |
| msi_endpoint: Option<String>, |
| ) -> Self { |
| let msi_endpoint = msi_endpoint |
| .unwrap_or_else(|| "http://169.254.169.254/metadata/identity/oauth2/token".to_owned()); |
| |
| Self { |
| msi_endpoint, |
| client_id, |
| object_id, |
| msi_res_id, |
| } |
| } |
| } |
| |
| #[async_trait::async_trait] |
| impl TokenProvider for ImdsManagedIdentityProvider { |
| type Credential = AzureCredential; |
| |
| /// Fetch a token |
| async fn fetch_token( |
| &self, |
| client: &HttpClient, |
| retry: &RetryConfig, |
| ) -> crate::Result<TemporaryToken<Arc<AzureCredential>>> { |
| let mut query_items = vec![ |
| ("api-version", MSI_API_VERSION), |
| ("resource", AZURE_STORAGE_RESOURCE), |
| ]; |
| |
| let mut identity = None; |
| if let Some(client_id) = &self.client_id { |
| identity = Some(("client_id", client_id)); |
| } |
| if let Some(object_id) = &self.object_id { |
| identity = Some(("object_id", object_id)); |
| } |
| if let Some(msi_res_id) = &self.msi_res_id { |
| identity = Some(("msi_res_id", msi_res_id)); |
| } |
| if let Some((key, value)) = identity { |
| query_items.push((key, value)); |
| } |
| |
| let mut builder = client |
| .request(Method::GET, &self.msi_endpoint) |
| .header("metadata", "true") |
| .query(&query_items); |
| |
| if let Ok(val) = std::env::var(MSI_SECRET_ENV_KEY) { |
| builder = builder.header("x-identity-header", val); |
| }; |
| |
| let response: ImdsTokenResponse = builder |
| .send_retry(retry) |
| .await |
| .map_err(|source| Error::TokenRequest { source })? |
| .into_body() |
| .json() |
| .await |
| .map_err(|source| Error::TokenResponseBody { source })?; |
| |
| Ok(TemporaryToken { |
| token: Arc::new(AzureCredential::BearerToken(response.access_token)), |
| expiry: Some(response.expires_on), |
| }) |
| } |
| } |
| |
| /// Credential for using workload identity federation |
| /// |
| /// <https://learn.microsoft.com/en-us/azure/active-directory/develop/workload-identity-federation> |
| #[derive(Debug)] |
| pub(crate) struct WorkloadIdentityOAuthProvider { |
| token_url: String, |
| client_id: String, |
| federated_token_file: String, |
| } |
| |
| impl WorkloadIdentityOAuthProvider { |
| /// Create a new [`WorkloadIdentityOAuthProvider`] for an azure backed store |
| pub(crate) fn new( |
| client_id: impl Into<String>, |
| federated_token_file: impl Into<String>, |
| tenant_id: impl AsRef<str>, |
| authority_host: Option<String>, |
| ) -> Self { |
| let authority_host = |
| authority_host.unwrap_or_else(|| authority_hosts::AZURE_PUBLIC_CLOUD.to_owned()); |
| |
| Self { |
| token_url: format!( |
| "{}/{}/oauth2/v2.0/token", |
| authority_host, |
| tenant_id.as_ref() |
| ), |
| client_id: client_id.into(), |
| federated_token_file: federated_token_file.into(), |
| } |
| } |
| } |
| |
| #[async_trait::async_trait] |
| impl TokenProvider for WorkloadIdentityOAuthProvider { |
| type Credential = AzureCredential; |
| |
| /// Fetch a token |
| async fn fetch_token( |
| &self, |
| client: &HttpClient, |
| retry: &RetryConfig, |
| ) -> crate::Result<TemporaryToken<Arc<AzureCredential>>> { |
| let token_str = std::fs::read_to_string(&self.federated_token_file) |
| .map_err(|_| Error::FederatedTokenFile)?; |
| |
| // https://learn.microsoft.com/en-us/azure/active-directory/develop/v2-oauth2-client-creds-grant-flow#third-case-access-token-request-with-a-federated-credential |
| let response: OAuthTokenResponse = client |
| .request(Method::POST, &self.token_url) |
| .header(ACCEPT, HeaderValue::from_static(CONTENT_TYPE_JSON)) |
| .form([ |
| ("client_id", self.client_id.as_str()), |
| ( |
| "client_assertion_type", |
| "urn:ietf:params:oauth:client-assertion-type:jwt-bearer", |
| ), |
| ("client_assertion", token_str.as_str()), |
| ("scope", AZURE_STORAGE_SCOPE), |
| ("grant_type", "client_credentials"), |
| ]) |
| .retryable(retry) |
| .idempotent(true) |
| .send() |
| .await |
| .map_err(|source| Error::TokenRequest { source })? |
| .into_body() |
| .json() |
| .await |
| .map_err(|source| Error::TokenResponseBody { source })?; |
| |
| Ok(TemporaryToken { |
| token: Arc::new(AzureCredential::BearerToken(response.access_token)), |
| expiry: Some(Instant::now() + Duration::from_secs(response.expires_in)), |
| }) |
| } |
| } |
| |
| mod az_cli_date_format { |
| use chrono::{DateTime, TimeZone}; |
| use serde::{self, Deserialize, Deserializer}; |
| |
| pub(crate) fn deserialize<'de, D>(deserializer: D) -> Result<DateTime<chrono::Local>, D::Error> |
| where |
| D: Deserializer<'de>, |
| { |
| let s = String::deserialize(deserializer)?; |
| // expiresOn from azure cli uses the local timezone |
| let date = chrono::NaiveDateTime::parse_from_str(&s, "%Y-%m-%d %H:%M:%S.%6f") |
| .map_err(serde::de::Error::custom)?; |
| chrono::Local |
| .from_local_datetime(&date) |
| .single() |
| .ok_or(serde::de::Error::custom( |
| "azure cli returned ambiguous expiry date", |
| )) |
| } |
| } |
| |
| #[derive(Debug, Clone, Deserialize)] |
| #[serde(rename_all = "camelCase")] |
| struct AzureCliTokenResponse { |
| pub access_token: String, |
| #[serde(with = "az_cli_date_format")] |
| pub expires_on: DateTime<chrono::Local>, |
| pub token_type: String, |
| } |
| |
| #[derive(Default, Debug)] |
| pub(crate) struct AzureCliCredential { |
| cache: TokenCache<Arc<AzureCredential>>, |
| } |
| |
| impl AzureCliCredential { |
| pub(crate) fn new() -> Self { |
| Self::default() |
| } |
| |
| /// Fetch a token |
| async fn fetch_token(&self) -> Result<TemporaryToken<Arc<AzureCredential>>> { |
| // on window az is a cmd and it should be called like this |
| // see https://doc.rust-lang.org/nightly/std/process/struct.Command.html |
| let program = if cfg!(target_os = "windows") { |
| "cmd" |
| } else { |
| "az" |
| }; |
| let mut args = Vec::new(); |
| if cfg!(target_os = "windows") { |
| args.push("/C"); |
| args.push("az"); |
| } |
| args.push("account"); |
| args.push("get-access-token"); |
| args.push("--output"); |
| args.push("json"); |
| args.push("--scope"); |
| args.push(AZURE_STORAGE_SCOPE); |
| |
| match Command::new(program).args(args).output() { |
| Ok(az_output) if az_output.status.success() => { |
| let output = str::from_utf8(&az_output.stdout).map_err(|_| Error::AzureCli { |
| message: "az response is not a valid utf-8 string".to_string(), |
| })?; |
| |
| let token_response = serde_json::from_str::<AzureCliTokenResponse>(output) |
| .map_err(|source| Error::AzureCliResponse { source })?; |
| |
| if !token_response.token_type.eq_ignore_ascii_case("bearer") { |
| return Err(Error::AzureCli { |
| message: format!( |
| "got unexpected token type from azure cli: {0}", |
| token_response.token_type |
| ), |
| }); |
| } |
| let duration = |
| token_response.expires_on.naive_local() - chrono::Local::now().naive_local(); |
| Ok(TemporaryToken { |
| token: Arc::new(AzureCredential::BearerToken(token_response.access_token)), |
| expiry: Some( |
| Instant::now() |
| + duration.to_std().map_err(|_| Error::AzureCli { |
| message: "az returned invalid lifetime".to_string(), |
| })?, |
| ), |
| }) |
| } |
| Ok(az_output) => { |
| let message = String::from_utf8_lossy(&az_output.stderr); |
| Err(Error::AzureCli { |
| message: message.into(), |
| }) |
| } |
| Err(e) => match e.kind() { |
| std::io::ErrorKind::NotFound => Err(Error::AzureCli { |
| message: "Azure Cli not installed".into(), |
| }), |
| error_kind => Err(Error::AzureCli { |
| message: format!("io error: {error_kind:?}"), |
| }), |
| }, |
| } |
| } |
| } |
| |
| /// Encapsulates the logic to perform an OAuth token challenge for Fabric |
| #[derive(Debug)] |
| pub(crate) struct FabricTokenOAuthProvider { |
| fabric_token_service_url: String, |
| fabric_workload_host: String, |
| fabric_session_token: String, |
| fabric_cluster_identifier: String, |
| storage_access_token: Option<String>, |
| token_expiry: Option<u64>, |
| } |
| |
| #[derive(Debug, Deserialize)] |
| struct Claims { |
| exp: u64, |
| } |
| |
| impl FabricTokenOAuthProvider { |
| /// Create a new [`FabricTokenOAuthProvider`] for an azure backed store |
| pub(crate) fn new( |
| fabric_token_service_url: impl Into<String>, |
| fabric_workload_host: impl Into<String>, |
| fabric_session_token: impl Into<String>, |
| fabric_cluster_identifier: impl Into<String>, |
| storage_access_token: Option<String>, |
| ) -> Self { |
| let (storage_access_token, token_expiry) = match storage_access_token { |
| Some(token) => match Self::validate_and_get_expiry(&token) { |
| Some(expiry) if expiry > Self::get_current_timestamp() + TOKEN_MIN_TTL => { |
| (Some(token), Some(expiry)) |
| } |
| _ => (None, None), |
| }, |
| None => (None, None), |
| }; |
| |
| Self { |
| fabric_token_service_url: fabric_token_service_url.into(), |
| fabric_workload_host: fabric_workload_host.into(), |
| fabric_session_token: fabric_session_token.into(), |
| fabric_cluster_identifier: fabric_cluster_identifier.into(), |
| storage_access_token, |
| token_expiry, |
| } |
| } |
| |
| fn validate_and_get_expiry(token: &str) -> Option<u64> { |
| let payload = token.split('.').nth(1)?; |
| let decoded_bytes = BASE64_URL_SAFE_NO_PAD.decode(payload).ok()?; |
| let decoded_str = str::from_utf8(&decoded_bytes).ok()?; |
| let claims: Claims = serde_json::from_str(decoded_str).ok()?; |
| Some(claims.exp) |
| } |
| |
| fn get_current_timestamp() -> u64 { |
| SystemTime::now() |
| .duration_since(SystemTime::UNIX_EPOCH) |
| .map_or(0, |d| d.as_secs()) |
| } |
| } |
| |
| #[async_trait::async_trait] |
| impl TokenProvider for FabricTokenOAuthProvider { |
| type Credential = AzureCredential; |
| |
| /// Fetch a token |
| async fn fetch_token( |
| &self, |
| client: &HttpClient, |
| retry: &RetryConfig, |
| ) -> crate::Result<TemporaryToken<Arc<AzureCredential>>> { |
| if let Some(storage_access_token) = &self.storage_access_token { |
| if let Some(expiry) = self.token_expiry { |
| let exp_in = expiry.saturating_sub(Self::get_current_timestamp()); |
| if exp_in > TOKEN_MIN_TTL { |
| return Ok(TemporaryToken { |
| token: Arc::new(AzureCredential::BearerToken(storage_access_token.clone())), |
| expiry: Some(Instant::now() + Duration::from_secs(exp_in)), |
| }); |
| } |
| } |
| } |
| |
| let query_items = vec![("resource", AZURE_STORAGE_RESOURCE)]; |
| let access_token: String = client |
| .request(Method::GET, &self.fabric_token_service_url) |
| .header(&PARTNER_TOKEN, self.fabric_session_token.as_str()) |
| .header(&CLUSTER_IDENTIFIER, self.fabric_cluster_identifier.as_str()) |
| .header(&WORKLOAD_RESOURCE, self.fabric_cluster_identifier.as_str()) |
| .header(&PROXY_HOST, self.fabric_workload_host.as_str()) |
| .query(&query_items) |
| .retryable(retry) |
| .idempotent(true) |
| .send() |
| .await |
| .map_err(|source| Error::TokenRequest { source })? |
| .into_body() |
| .text() |
| .await |
| .map_err(|source| Error::TokenResponseBody { source })?; |
| let exp_in = Self::validate_and_get_expiry(&access_token).map_or(3600, |expiry| { |
| expiry.saturating_sub(Self::get_current_timestamp()) |
| }); |
| Ok(TemporaryToken { |
| token: Arc::new(AzureCredential::BearerToken(access_token)), |
| expiry: Some(Instant::now() + Duration::from_secs(exp_in)), |
| }) |
| } |
| } |
| |
| #[async_trait] |
| impl CredentialProvider for AzureCliCredential { |
| type Credential = AzureCredential; |
| |
| async fn get_credential(&self) -> crate::Result<Arc<Self::Credential>> { |
| Ok(self.cache.get_or_insert_with(|| self.fetch_token()).await?) |
| } |
| } |
| |
| #[cfg(test)] |
| mod tests { |
| use futures_executor::block_on; |
| use http::{Response, StatusCode}; |
| use http_body_util::BodyExt; |
| use reqwest::{Client, Method}; |
| use tempfile::NamedTempFile; |
| |
| use super::*; |
| use crate::azure::MicrosoftAzureBuilder; |
| use crate::azure::client::RequestVersionExt; |
| use crate::client::HttpRequestBody; |
| use crate::client::mock_server::MockServer; |
| use crate::{ObjectStoreExt, Path}; |
| |
| #[cfg(feature = "reqwest")] |
| #[tokio::test] |
| async fn test_managed_identity() { |
| let server = MockServer::new().await; |
| |
| unsafe { std::env::set_var(MSI_SECRET_ENV_KEY, "env-secret") }; |
| |
| let endpoint = server.url(); |
| let client = HttpClient::new(Client::new()); |
| let retry_config = RetryConfig::default(); |
| |
| // Test IMDS |
| server.push_fn(|req| { |
| assert_eq!(req.uri().path(), "/metadata/identity/oauth2/token"); |
| assert!(req.uri().query().unwrap().contains("client_id=client_id")); |
| assert_eq!(req.method(), &Method::GET); |
| let t = req |
| .headers() |
| .get("x-identity-header") |
| .unwrap() |
| .to_str() |
| .unwrap(); |
| assert_eq!(t, "env-secret"); |
| let t = req.headers().get("metadata").unwrap().to_str().unwrap(); |
| assert_eq!(t, "true"); |
| Response::new( |
| r#" |
| { |
| "access_token": "TOKEN", |
| "refresh_token": "", |
| "expires_in": "3599", |
| "expires_on": "1506484173", |
| "not_before": "1506480273", |
| "resource": "https://management.azure.com/", |
| "token_type": "Bearer" |
| } |
| "# |
| .to_string(), |
| ) |
| }); |
| |
| let credential = ImdsManagedIdentityProvider::new( |
| Some("client_id".into()), |
| None, |
| None, |
| Some(format!("{endpoint}/metadata/identity/oauth2/token")), |
| ); |
| |
| let token = credential |
| .fetch_token(&client, &retry_config) |
| .await |
| .unwrap(); |
| |
| assert_eq!( |
| token.token.as_ref(), |
| &AzureCredential::BearerToken("TOKEN".into()) |
| ); |
| } |
| |
| #[cfg(feature = "reqwest")] |
| #[tokio::test] |
| async fn test_workload_identity() { |
| let server = MockServer::new().await; |
| let tokenfile = NamedTempFile::new().unwrap(); |
| let tenant = "tenant"; |
| std::fs::write(tokenfile.path(), "federated-token").unwrap(); |
| |
| let endpoint = server.url(); |
| let client = HttpClient::new(Client::new()); |
| let retry_config = RetryConfig::default(); |
| |
| // Test IMDS |
| server.push_fn(move |req| { |
| assert_eq!(req.uri().path(), format!("/{tenant}/oauth2/v2.0/token")); |
| assert_eq!(req.method(), &Method::POST); |
| let body = block_on(async move { req.into_body().collect().await.unwrap().to_bytes() }); |
| let body = String::from_utf8(body.to_vec()).unwrap(); |
| assert!(body.contains("federated-token")); |
| Response::new( |
| r#" |
| { |
| "access_token": "TOKEN", |
| "refresh_token": "", |
| "expires_in": 3599, |
| "expires_on": "1506484173", |
| "not_before": "1506480273", |
| "resource": "https://management.azure.com/", |
| "token_type": "Bearer" |
| } |
| "# |
| .to_string(), |
| ) |
| }); |
| |
| let credential = WorkloadIdentityOAuthProvider::new( |
| "client_id", |
| tokenfile.path().to_str().unwrap(), |
| tenant, |
| Some(endpoint.to_string()), |
| ); |
| |
| let token = credential |
| .fetch_token(&client, &retry_config) |
| .await |
| .unwrap(); |
| |
| assert_eq!( |
| token.token.as_ref(), |
| &AzureCredential::BearerToken("TOKEN".into()) |
| ); |
| } |
| |
| #[cfg(feature = "reqwest")] |
| #[tokio::test] |
| async fn test_no_credentials() { |
| let server = MockServer::new().await; |
| |
| let endpoint = server.url(); |
| let store = MicrosoftAzureBuilder::new() |
| .with_account("test") |
| .with_container_name("test") |
| .with_allow_http(true) |
| .with_bearer_token_authorization("token") |
| .with_endpoint(endpoint.to_string()) |
| .with_skip_signature(true) |
| .build() |
| .unwrap(); |
| |
| server.push_fn(|req| { |
| assert_eq!(req.method(), &Method::GET); |
| assert!(req.headers().get("Authorization").is_none()); |
| Response::builder() |
| .status(StatusCode::NOT_FOUND) |
| .body("not found".to_string()) |
| .unwrap() |
| }); |
| |
| let path = Path::from("file.txt"); |
| match store.get(&path).await { |
| Err(crate::Error::NotFound { .. }) => {} |
| _ => { |
| panic!("unexpected response"); |
| } |
| } |
| } |
| |
| #[test] |
| fn bearer_authorization_does_not_require_crypto_provider() { |
| let credential = Some(Arc::new(AzureCredential::BearerToken("TOKEN".into()))); |
| let builder = HttpRequestBuilder::new(HttpClient::new(Client::new())) |
| .method(Method::GET) |
| .uri("https://account.blob.core.windows.net/container/file.txt") |
| .body(HttpRequestBody::empty()) |
| .with_azure_authorization(None, &credential, "account") |
| .unwrap(); |
| |
| let (_, request) = builder.into_parts(); |
| let request = request.unwrap(); |
| assert_eq!( |
| request.headers().get(AUTHORIZATION).unwrap(), |
| "Bearer TOKEN" |
| ); |
| assert!(request.headers().contains_key(&DATE)); |
| assert!(request.headers().contains_key(&VERSION)); |
| } |
| |
| #[test] |
| fn sas_authorization_does_not_require_crypto_provider() { |
| let credential = Some(Arc::new(AzureCredential::SASToken(vec![ |
| ("sig".into(), "signature".into()), |
| ("se".into(), "expiry".into()), |
| ]))); |
| let builder = HttpRequestBuilder::new(HttpClient::new(Client::new())) |
| .method(Method::GET) |
| .uri("https://account.blob.core.windows.net/container/file.txt") |
| .body(HttpRequestBody::empty()) |
| .with_azure_authorization(None, &credential, "account") |
| .unwrap(); |
| |
| let (_, request) = builder.into_parts(); |
| let request = request.unwrap(); |
| let query = request.uri().query().unwrap(); |
| assert!(query.contains("sig=signature")); |
| assert!(query.contains("se=expiry")); |
| assert!(request.headers().contains_key(&DATE)); |
| assert!(request.headers().contains_key(&VERSION)); |
| } |
| |
| #[cfg(feature = "reqwest")] |
| #[tokio::test] |
| async fn test_fabric_refresh_expired_token() { |
| let server = MockServer::new().await; |
| |
| // Create an expired initial token (1 hour in the past) |
| let expired_timestamp = FabricTokenOAuthProvider::get_current_timestamp() - 3600; |
| let claims = format!(r#"{{"exp":{expired_timestamp}}}"#); |
| let encoded_claims = BASE64_URL_SAFE_NO_PAD.encode(claims.as_bytes()); |
| let expired_token = format!("header.{encoded_claims}.signature"); |
| |
| // Create a fresh token that the mock API will return (1 hour in the future) |
| let fresh_timestamp = FabricTokenOAuthProvider::get_current_timestamp() + 3600; |
| let fresh_claims = format!(r#"{{"exp":{fresh_timestamp}}}"#); |
| let fresh_encoded = BASE64_URL_SAFE_NO_PAD.encode(fresh_claims.as_bytes()); |
| let fresh_token = format!("header.{fresh_encoded}.signature"); |
| let expected_token = fresh_token.clone(); |
| |
| // Mock the Fabric token service to return a fresh token |
| server.push_fn(move |req| { |
| assert_eq!(req.headers().get(&PARTNER_TOKEN).unwrap(), "session-token"); |
| assert_eq!( |
| req.headers().get(&CLUSTER_IDENTIFIER).unwrap(), |
| "cluster-id" |
| ); |
| assert_eq!(req.headers().get(&PROXY_HOST).unwrap(), "fake"); |
| |
| Response::new(fresh_token) |
| }); |
| |
| let provider = FabricTokenOAuthProvider { |
| fabric_token_service_url: server.url().to_string(), |
| fabric_workload_host: "fake".to_string(), |
| fabric_session_token: "session-token".to_string(), |
| fabric_cluster_identifier: "cluster-id".to_string(), |
| storage_access_token: Some(expired_token), |
| token_expiry: Some(expired_timestamp), |
| }; |
| |
| let client = HttpClient::new(Client::new()); |
| let retry = RetryConfig::default(); |
| |
| let result = provider.fetch_token(&client, &retry).await; |
| |
| assert!( |
| result.is_ok(), |
| "fetch_token should handle expired cached token gracefully" |
| ); |
| |
| let temp_token = result.unwrap(); |
| |
| // Verify we got a fresh token from the API (not the expired cached one) |
| if let AzureCredential::BearerToken(token) = temp_token.token.as_ref() { |
| assert_eq!( |
| token, &expected_token, |
| "Should have fetched fresh token from API, not returned expired cached token" |
| ); |
| } |
| } |
| |
| #[cfg(feature = "reqwest")] |
| #[test] |
| fn test_azure_authorization_preserves_explicit_version() { |
| let credential = Some(Arc::new(AzureCredential::BearerToken("token".to_string()))); |
| let client = HttpClient::new(Client::new()); |
| |
| let request = client |
| .request(Method::GET, "http://example.com/container/blob") |
| .with_azure_version("2026-02-06") |
| .with_azure_authorization(None, &credential, "account") |
| .unwrap() |
| .into_parts() |
| .1 |
| .unwrap(); |
| |
| assert_eq!(request.headers().get(&VERSION).unwrap(), "2026-02-06"); |
| } |
| } |
