1use std::fmt::Debug;
19use std::ops::Deref;
20use std::process::Command;
21use std::str;
22use std::sync::Arc;
23use std::time::{Duration, Instant, SystemTime};
24
25use async_trait::async_trait;
26use chrono::DateTime;
27use reqwest::header::{ACCEPT, AUTHORIZATION, HeaderValue};
28use reqwest::{Client, Method, Request, RequestBuilder};
29use serde::Deserialize;
30
31use crate::retry::RetryExt;
32use crate::service::HttpService;
33use crate::token::{TemporaryToken, TokenCache};
34use crate::{CredentialProvider, RetryConfig, TokenProvider};
35
36const CONTENT_TYPE_JSON: &str = "application/json";
37const MSI_SECRET_ENV_KEY: &str = "IDENTITY_HEADER";
38const MSI_API_VERSION: &str = "2019-08-01";
39
40pub(crate) const AZURE_STORAGE_SCOPE: &str = "https://storage.azure.com/.default";
42
43pub(crate) const AZURE_STORAGE_RESOURCE: &str = "https://storage.azure.com";
45
46pub(crate) const AZURE_DATABRICKS_RESOURCE: &str = "2ff814a6-3304-4ab8-85cb-cd0e6f879c1d";
48
49#[allow(dead_code)]
52pub(crate) const AZURE_DATABRICKS_SCOPE: &str = "2ff814a6-3304-4ab8-85cb-cd0e6f879c1d/.default";
53
54#[derive(Debug, thiserror::Error)]
55pub enum Error {
56 #[error("Error performing token request: {}", source)]
57 TokenRequest { source: crate::retry::Error },
58
59 #[error("Error getting token response body: {}", source)]
60 TokenResponseBody { source: reqwest::Error },
61
62 #[error("Error reading federated token file ")]
63 FederatedTokenFile,
64
65 #[error("Invalid Access Key: {}", source)]
66 InvalidAccessKey { source: base64::DecodeError },
67
68 #[error("'az account get-access-token' command failed: {message}")]
69 AzureCli { message: String },
70
71 #[error("Failed to parse azure cli response: {source}")]
72 AzureCliResponse { source: serde_json::Error },
73}
74
75pub(crate) type Result<T, E = Error> = std::result::Result<T, E>;
76
77impl From<Error> for crate::Error {
78 fn from(value: Error) -> Self {
79 Self::Generic {
80 source: Box::new(value),
81 }
82 }
83}
84
85#[derive(Debug, Eq, PartialEq)]
87pub enum AzureCredential {
88 BearerToken(String),
90}
91
92#[allow(dead_code)]
94pub mod authority_hosts {
95 pub const AZURE_CHINA: &str = "https://login.chinacloudapi.cn";
97 pub const AZURE_GERMANY: &str = "https://login.microsoftonline.de";
99 pub const AZURE_GOVERNMENT: &str = "https://login.microsoftonline.us";
101 pub const AZURE_PUBLIC_CLOUD: &str = "https://login.microsoftonline.com";
103}
104
105#[derive(Debug)]
107pub struct AzureAuthorizer<'a> {
108 credential: &'a AzureCredential,
109}
110
111impl<'a> AzureAuthorizer<'a> {
112 pub fn new(credential: &'a AzureCredential) -> Self {
114 AzureAuthorizer { credential }
115 }
116
117 pub fn authorize(&self, request: &mut Request) {
119 match self.credential {
120 AzureCredential::BearerToken(token) => {
121 request.headers_mut().append(
122 AUTHORIZATION,
123 HeaderValue::from_str(format!("Bearer {token}").as_str()).unwrap(),
124 );
125 }
126 }
127 }
128}
129
130pub trait AzureCredentialExt {
131 fn with_azure_authorization(
133 self,
134 credential: &Option<impl Deref<Target = AzureCredential>>,
135 ) -> Self;
136}
137
138impl AzureCredentialExt for RequestBuilder {
139 fn with_azure_authorization(
140 self,
141 credential: &Option<impl Deref<Target = AzureCredential>>,
142 ) -> Self {
143 match credential.as_deref() {
144 Some(credential) => {
145 let (client, request) = self.build_split();
146 let mut request = request.expect("request valid");
147 AzureAuthorizer::new(credential).authorize(&mut request);
148 Self::from_parts(client, request)
149 }
150 None => self,
151 }
152 }
153}
154
155#[derive(Deserialize, Debug)]
157struct OAuthTokenResponse {
158 access_token: String,
159 expires_in: u64,
160}
161
162#[derive(Debug)]
173pub(crate) struct ClientSecretOAuthProvider {
174 token_url: String,
175 client_id: String,
176 client_secret: String,
177 scope: String,
178}
179
180impl ClientSecretOAuthProvider {
181 pub(crate) fn new_with_scope(
183 client_id: String,
184 client_secret: String,
185 tenant_id: impl AsRef<str>,
186 authority_host: Option<String>,
187 scope: &str,
188 ) -> Self {
189 let authority_host =
190 authority_host.unwrap_or_else(|| authority_hosts::AZURE_PUBLIC_CLOUD.to_owned());
191
192 Self {
193 token_url: format!(
194 "{}/{}/oauth2/v2.0/token",
195 authority_host,
196 tenant_id.as_ref()
197 ),
198 client_id,
199 client_secret,
200 scope: scope.to_owned(),
201 }
202 }
203}
204
205#[async_trait::async_trait]
206impl TokenProvider for ClientSecretOAuthProvider {
207 type Credential = AzureCredential;
208
209 async fn fetch_token(
210 &self,
211 client: &Client,
212 service: &Arc<dyn HttpService>,
213 retry: &RetryConfig,
214 ) -> crate::Result<TemporaryToken<Arc<AzureCredential>>> {
215 let response: OAuthTokenResponse = client
216 .request(Method::POST, &self.token_url)
217 .header(ACCEPT, HeaderValue::from_static(CONTENT_TYPE_JSON))
218 .form(&[
219 ("client_id", self.client_id.as_str()),
220 ("client_secret", self.client_secret.as_str()),
221 ("scope", self.scope.as_str()),
222 ("grant_type", "client_credentials"),
223 ])
224 .retryable(retry, service.clone())
225 .idempotent(true)
226 .send()
227 .await
228 .map_err(|source| Error::TokenRequest { source })?
229 .json()
230 .await
231 .map_err(|source| Error::TokenResponseBody { source })?;
232
233 Ok(TemporaryToken {
234 token: Arc::new(AzureCredential::BearerToken(response.access_token)),
235 expiry: Some(Instant::now() + Duration::from_secs(response.expires_in)),
236 })
237 }
238}
239
240fn expires_on_string<'de, D>(deserializer: D) -> std::result::Result<Instant, D::Error>
241where
242 D: serde::de::Deserializer<'de>,
243{
244 let v = String::deserialize(deserializer)?;
245 let v = v.parse::<u64>().map_err(serde::de::Error::custom)?;
246 let now = SystemTime::now()
247 .duration_since(SystemTime::UNIX_EPOCH)
248 .map_err(serde::de::Error::custom)?;
249
250 Ok(Instant::now() + Duration::from_secs(v.saturating_sub(now.as_secs())))
251}
252
253#[derive(Debug, Clone, Deserialize)]
254struct ImdsTokenResponse {
255 pub access_token: String,
256 #[serde(deserialize_with = "expires_on_string")]
257 pub expires_on: Instant,
258}
259
260#[derive(Debug)]
272pub(crate) struct ImdsManagedIdentityProvider {
273 msi_endpoint: String,
274 client_id: Option<String>,
275 object_id: Option<String>,
276 msi_res_id: Option<String>,
277 resource: String,
278}
279
280impl ImdsManagedIdentityProvider {
281 pub(crate) fn new(
282 client_id: Option<String>,
283 object_id: Option<String>,
284 msi_res_id: Option<String>,
285 msi_endpoint: Option<String>,
286 ) -> Self {
287 Self::new_with_resource(
288 client_id,
289 object_id,
290 msi_res_id,
291 msi_endpoint,
292 AZURE_STORAGE_RESOURCE,
293 )
294 }
295
296 pub(crate) fn new_with_resource(
298 client_id: Option<String>,
299 object_id: Option<String>,
300 msi_res_id: Option<String>,
301 msi_endpoint: Option<String>,
302 resource: &str,
303 ) -> Self {
304 let msi_endpoint = msi_endpoint
305 .unwrap_or_else(|| "http://169.254.169.254/metadata/identity/oauth2/token".to_owned());
306
307 Self {
308 msi_endpoint,
309 client_id,
310 object_id,
311 msi_res_id,
312 resource: resource.to_owned(),
313 }
314 }
315}
316
317#[async_trait::async_trait]
318impl TokenProvider for ImdsManagedIdentityProvider {
319 type Credential = AzureCredential;
320
321 async fn fetch_token(
322 &self,
323 client: &Client,
324 service: &Arc<dyn HttpService>,
325 retry: &RetryConfig,
326 ) -> crate::Result<TemporaryToken<Arc<AzureCredential>>> {
327 let mut query_items = vec![
328 ("api-version", MSI_API_VERSION),
329 ("resource", self.resource.as_str()),
330 ];
331
332 let mut identity = None;
333 if let Some(client_id) = &self.client_id {
334 identity = Some(("client_id", client_id));
335 }
336 if let Some(object_id) = &self.object_id {
337 identity = Some(("object_id", object_id));
338 }
339 if let Some(msi_res_id) = &self.msi_res_id {
340 identity = Some(("msi_res_id", msi_res_id));
341 }
342 if let Some((key, value)) = identity {
343 query_items.push((key, value));
344 }
345
346 let mut builder = client
347 .request(Method::GET, &self.msi_endpoint)
348 .header("metadata", "true")
349 .query(&query_items);
350
351 if let Ok(val) = std::env::var(MSI_SECRET_ENV_KEY) {
352 builder = builder.header("x-identity-header", val);
353 };
354
355 let response: ImdsTokenResponse = builder
356 .send_retry(retry, service.clone())
357 .await
358 .map_err(|source| Error::TokenRequest { source })?
359 .json()
360 .await
361 .map_err(|source| Error::TokenResponseBody { source })?;
362
363 Ok(TemporaryToken {
364 token: Arc::new(AzureCredential::BearerToken(response.access_token)),
365 expiry: Some(response.expires_on),
366 })
367 }
368}
369
370#[derive(Debug)]
381pub(crate) struct WorkloadIdentityOAuthProvider {
382 token_url: String,
383 client_id: String,
384 federated_token_file: String,
385}
386
387impl WorkloadIdentityOAuthProvider {
388 pub(crate) fn new(
389 client_id: impl Into<String>,
390 federated_token_file: impl Into<String>,
391 tenant_id: impl AsRef<str>,
392 authority_host: Option<String>,
393 ) -> Self {
394 let authority_host =
395 authority_host.unwrap_or_else(|| authority_hosts::AZURE_PUBLIC_CLOUD.to_owned());
396
397 Self {
398 token_url: format!(
399 "{}/{}/oauth2/v2.0/token",
400 authority_host,
401 tenant_id.as_ref()
402 ),
403 client_id: client_id.into(),
404 federated_token_file: federated_token_file.into(),
405 }
406 }
407}
408
409#[async_trait::async_trait]
410impl TokenProvider for WorkloadIdentityOAuthProvider {
411 type Credential = AzureCredential;
412
413 async fn fetch_token(
414 &self,
415 client: &Client,
416 service: &Arc<dyn HttpService>,
417 retry: &RetryConfig,
418 ) -> crate::Result<TemporaryToken<Arc<AzureCredential>>> {
419 let token_str = std::fs::read_to_string(&self.federated_token_file)
420 .map_err(|_| Error::FederatedTokenFile)?;
421
422 let response: OAuthTokenResponse = client
423 .request(Method::POST, &self.token_url)
424 .header(ACCEPT, HeaderValue::from_static(CONTENT_TYPE_JSON))
425 .form(&[
426 ("client_id", self.client_id.as_str()),
427 (
428 "client_assertion_type",
429 "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
430 ),
431 ("client_assertion", token_str.as_str()),
432 ("scope", AZURE_STORAGE_SCOPE),
433 ("grant_type", "client_credentials"),
434 ])
435 .retryable(retry, service.clone())
436 .idempotent(true)
437 .send()
438 .await
439 .map_err(|source| Error::TokenRequest { source })?
440 .json()
441 .await
442 .map_err(|source| Error::TokenResponseBody { source })?;
443
444 Ok(TemporaryToken {
445 token: Arc::new(AzureCredential::BearerToken(response.access_token)),
446 expiry: Some(Instant::now() + Duration::from_secs(response.expires_in)),
447 })
448 }
449}
450
451mod az_cli_date_format {
452 use chrono::{DateTime, TimeZone};
453 use serde::{self, Deserialize, Deserializer};
454
455 pub(crate) fn deserialize<'de, D>(deserializer: D) -> Result<DateTime<chrono::Local>, D::Error>
456 where
457 D: Deserializer<'de>,
458 {
459 let s = String::deserialize(deserializer)?;
460 let date = chrono::NaiveDateTime::parse_from_str(&s, "%Y-%m-%d %H:%M:%S.%6f")
461 .map_err(serde::de::Error::custom)?;
462 chrono::Local
463 .from_local_datetime(&date)
464 .single()
465 .ok_or(serde::de::Error::custom(
466 "azure cli returned ambiguous expiry date",
467 ))
468 }
469}
470
471#[derive(Debug, Clone, Deserialize)]
472#[serde(rename_all = "camelCase")]
473struct AzureCliTokenResponse {
474 pub access_token: String,
475 #[serde(with = "az_cli_date_format")]
476 pub expires_on: DateTime<chrono::Local>,
477 pub token_type: String,
478}
479
480#[derive(Default, Debug)]
481pub(crate) struct AzureCliCredential {
482 cache: TokenCache<Arc<AzureCredential>>,
483}
484
485impl AzureCliCredential {
486 pub(crate) fn new() -> Self {
487 Self::default()
488 }
489
490 async fn fetch_token(&self) -> Result<TemporaryToken<Arc<AzureCredential>>> {
491 let program = if cfg!(target_os = "windows") {
492 "cmd"
493 } else {
494 "az"
495 };
496 let mut args = Vec::new();
497 if cfg!(target_os = "windows") {
498 args.push("/C");
499 args.push("az");
500 }
501 args.push("account");
502 args.push("get-access-token");
503 args.push("--output");
504 args.push("json");
505 args.push("--scope");
506 args.push(AZURE_STORAGE_SCOPE);
507
508 match Command::new(program).args(args).output() {
509 Ok(az_output) if az_output.status.success() => {
510 let output = str::from_utf8(&az_output.stdout).map_err(|_| Error::AzureCli {
511 message: "az response is not a valid utf-8 string".to_string(),
512 })?;
513
514 let token_response = serde_json::from_str::<AzureCliTokenResponse>(output)
515 .map_err(|source| Error::AzureCliResponse { source })?;
516
517 if !token_response.token_type.eq_ignore_ascii_case("bearer") {
518 return Err(Error::AzureCli {
519 message: format!(
520 "got unexpected token type from azure cli: {0}",
521 token_response.token_type
522 ),
523 });
524 }
525 let duration =
526 token_response.expires_on.naive_local() - chrono::Local::now().naive_local();
527 Ok(TemporaryToken {
528 token: Arc::new(AzureCredential::BearerToken(token_response.access_token)),
529 expiry: Some(
530 Instant::now()
531 + duration.to_std().map_err(|_| Error::AzureCli {
532 message: "az returned invalid lifetime".to_string(),
533 })?,
534 ),
535 })
536 }
537 Ok(az_output) => {
538 let message = String::from_utf8_lossy(&az_output.stderr);
539 Err(Error::AzureCli {
540 message: message.into(),
541 })
542 }
543 Err(e) => match e.kind() {
544 std::io::ErrorKind::NotFound => Err(Error::AzureCli {
545 message: "Azure Cli not installed".into(),
546 }),
547 error_kind => Err(Error::AzureCli {
548 message: format!("io error: {error_kind:?}"),
549 }),
550 },
551 }
552 }
553}
554
555#[async_trait]
556impl CredentialProvider for AzureCliCredential {
557 type Credential = AzureCredential;
558
559 async fn get_credential(&self) -> crate::Result<Arc<Self::Credential>> {
560 Ok(self.cache.get_or_insert_with(|| self.fetch_token()).await?)
561 }
562}
563
564#[cfg(test)]
565mod tests {
566 use reqwest::Client;
567 use tempfile::NamedTempFile;
568
569 use super::*;
570 use crate::service::ReqwestService;
571 use mockito;
572
573 #[tokio::test]
574 async fn test_managed_identity() {
575 let mut server = mockito::Server::new_async().await;
576
577 unsafe { std::env::set_var(MSI_SECRET_ENV_KEY, "env-secret") };
578
579 let endpoint = server.url();
580 let client = Client::new();
581 let service: Arc<dyn HttpService> = Arc::new(ReqwestService::new(client.clone()));
582 let retry_config = RetryConfig::default();
583
584 let _mock = server
585 .mock("GET", "/metadata/identity/oauth2/token")
586 .match_query(mockito::Matcher::AllOf(vec![mockito::Matcher::UrlEncoded(
587 "client_id".into(),
588 "client_id".into(),
589 )]))
590 .match_header("x-identity-header", "env-secret")
591 .match_header("metadata", "true")
592 .with_status(200)
593 .with_body(
594 r#"
595 {
596 "access_token": "TOKEN",
597 "refresh_token": "",
598 "expires_in": "3599",
599 "expires_on": "1506484173",
600 "not_before": "1506480273",
601 "resource": "https://management.azure.com/",
602 "token_type": "Bearer"
603 }
604 "#,
605 )
606 .create_async()
607 .await;
608
609 let credential = ImdsManagedIdentityProvider::new(
610 Some("client_id".into()),
611 None,
612 None,
613 Some(format!("{endpoint}/metadata/identity/oauth2/token")),
614 );
615
616 let token = credential
617 .fetch_token(&client, &service, &retry_config)
618 .await
619 .unwrap();
620
621 assert_eq!(
622 token.token.as_ref(),
623 &AzureCredential::BearerToken("TOKEN".into())
624 );
625 }
626
627 #[tokio::test]
628 async fn test_workload_identity() {
629 let mut server = mockito::Server::new_async().await;
630 let tokenfile = NamedTempFile::new().unwrap();
631 let tenant = "tenant";
632 std::fs::write(tokenfile.path(), "federated-token").unwrap();
633
634 let endpoint = server.url();
635 let client = Client::new();
636 let service: Arc<dyn HttpService> = Arc::new(ReqwestService::new(client.clone()));
637 let retry_config = RetryConfig::default();
638
639 let _mock = server
640 .mock("POST", format!("/{tenant}/oauth2/v2.0/token").as_str())
641 .match_body(mockito::Matcher::Regex("federated-token".into()))
642 .with_status(200)
643 .with_body(
644 r#"
645 {
646 "access_token": "TOKEN",
647 "refresh_token": "",
648 "expires_in": 3599,
649 "expires_on": "1506484173",
650 "not_before": "1506480273",
651 "resource": "https://management.azure.com/",
652 "token_type": "Bearer"
653 }
654 "#,
655 )
656 .create_async()
657 .await;
658
659 let credential = WorkloadIdentityOAuthProvider::new(
660 "client_id",
661 tokenfile.path().to_str().unwrap(),
662 tenant,
663 Some(endpoint.to_string()),
664 );
665
666 let token = credential
667 .fetch_token(&client, &service, &retry_config)
668 .await
669 .unwrap();
670
671 assert_eq!(
672 token.token.as_ref(),
673 &AzureCredential::BearerToken("TOKEN".into())
674 );
675 }
676
677 #[tokio::test]
678 async fn test_client_secret_happy_path() {
679 let mut server = mockito::Server::new_async().await;
680 let tenant = "my-tenant";
681
682 let _mock = server
683 .mock("POST", format!("/{tenant}/oauth2/v2.0/token").as_str())
684 .match_body(mockito::Matcher::AllOf(vec![
685 mockito::Matcher::Regex("client_id=myclientid".into()),
686 mockito::Matcher::Regex("grant_type=client_credentials".into()),
687 ]))
688 .with_status(200)
689 .with_body(r#"{"access_token":"AZURE_TOKEN","expires_in":3599,"token_type":"Bearer"}"#)
690 .create_async()
691 .await;
692
693 let client = Client::new();
694 let service: Arc<dyn HttpService> = Arc::new(ReqwestService::new(client.clone()));
695 let retry_config = RetryConfig::default();
696
697 let provider = ClientSecretOAuthProvider::new_with_scope(
698 "myclientid".into(),
699 "myclientsecret".into(),
700 tenant,
701 Some(server.url()),
702 AZURE_STORAGE_SCOPE,
703 );
704
705 let token = provider
706 .fetch_token(&client, &service, &retry_config)
707 .await
708 .unwrap();
709
710 assert_eq!(
711 token.token.as_ref(),
712 &AzureCredential::BearerToken("AZURE_TOKEN".into())
713 );
714 assert!(token.expiry.is_some());
715
716 _mock.assert_async().await;
717 }
718
719 #[tokio::test]
720 async fn test_client_secret_invalid_client() {
721 let mut server = mockito::Server::new_async().await;
722 let tenant = "bad-tenant";
723
724 let _mock = server
725 .mock("POST", format!("/{tenant}/oauth2/v2.0/token").as_str())
726 .with_status(400)
727 .with_body(
728 r#"{"error":"invalid_client","error_description":"Client authentication failed"}"#,
729 )
730 .create_async()
731 .await;
732
733 let client = Client::new();
734 let service: Arc<dyn HttpService> = Arc::new(ReqwestService::new(client.clone()));
735 let retry_config = RetryConfig::default();
736
737 let provider = ClientSecretOAuthProvider::new_with_scope(
738 "badclientid".into(),
739 "badsecret".into(),
740 tenant,
741 Some(server.url()),
742 AZURE_STORAGE_SCOPE,
743 );
744
745 let result = provider.fetch_token(&client, &service, &retry_config).await;
746
747 assert!(result.is_err(), "Expected error for 400 response");
748
749 _mock.assert_async().await;
750 }
751
752 #[tokio::test]
753 async fn test_client_secret_token_refresh() {
754 let mut server = mockito::Server::new_async().await;
755 let tenant = "refresh-tenant";
756
757 let _mock1 = server
759 .mock("POST", format!("/{tenant}/oauth2/v2.0/token").as_str())
760 .with_status(200)
761 .with_body(r#"{"access_token":"FIRST_TOKEN","expires_in":1,"token_type":"Bearer"}"#)
762 .create_async()
763 .await;
764
765 let client = Client::new();
766 let service: Arc<dyn HttpService> = Arc::new(ReqwestService::new(client.clone()));
767 let retry_config = RetryConfig::default();
768
769 let provider = ClientSecretOAuthProvider::new_with_scope(
770 "myclientid".into(),
771 "myclientsecret".into(),
772 tenant,
773 Some(server.url()),
774 AZURE_STORAGE_SCOPE,
775 );
776
777 let token1 = provider
778 .fetch_token(&client, &service, &retry_config)
779 .await
780 .unwrap();
781 assert_eq!(
782 token1.token.as_ref(),
783 &AzureCredential::BearerToken("FIRST_TOKEN".into())
784 );
785 assert!(token1.expiry.is_some());
787
788 let _mock2 = server
790 .mock("POST", format!("/{tenant}/oauth2/v2.0/token").as_str())
791 .with_status(200)
792 .with_body(r#"{"access_token":"SECOND_TOKEN","expires_in":3600,"token_type":"Bearer"}"#)
793 .create_async()
794 .await;
795
796 let token2 = provider
797 .fetch_token(&client, &service, &retry_config)
798 .await
799 .unwrap();
800 assert_eq!(
801 token2.token.as_ref(),
802 &AzureCredential::BearerToken("SECOND_TOKEN".into())
803 );
804
805 _mock1.assert_async().await;
806 _mock2.assert_async().await;
807 }
808}