olai_http/
bearer.rs

1//! Bearer-token authentication for arbitrary HTTP APIs.
2//!
3//! The per-cloud and Databricks modules sign requests for one specific
4//! service. This module builds bearer tokens for *any* audience: a provider is
5//! configured with the token endpoint, scope, or audience of the API it is
6//! for, so the same flows work against an API fronted by Entra ID, Okta,
7//! Keycloak, Cognito, Google, or a self-issued token service.
8//!
9//! [`BearerAuth`] holds the HTTP settings shared by every token request and
10//! creates [`BearerCredentialProvider`]s:
11//!
12//! | Flow | Constructor |
13//! |------|-------------|
14//! | Static token | [`BearerAuth::static_token`] |
15//! | OAuth 2.0 client credentials against any token endpoint | [`BearerAuth::client_credentials`] |
16//! | Token read from an environment variable / a file (e.g. a projected Kubernetes token) | [`BearerAuth::token_env`], [`BearerAuth::token_file`] |
17//! | Azure service principal, managed identity, workload identity, Azure CLI | [`BearerAuth::azure_client_secret`], [`BearerAuth::azure_managed_identity`], [`BearerAuth::azure_workload_identity`], [`BearerAuth::azure_cli`] |
18//! | GCP ID token from the metadata server or a service-account key | [`BearerAuth::gcp_id_token`], [`BearerAuth::gcp_service_account_id_token`] |
19//! | RFC 8693 token exchange of any of the above | [`BearerAuth::token_exchange`] |
20//!
21//! Hand a provider to [`CloudClient::new_with_bearer`] to sign requests with it.
22//!
23//! [`CloudClient::new_with_bearer`]: crate::CloudClient::new_with_bearer
24
25use std::fmt;
26use std::sync::Arc;
27use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
28
29use async_trait::async_trait;
30use reqwest::header::{ACCEPT, HeaderValue};
31use reqwest::{Client, Method};
32use serde::Deserialize;
33use tokio::runtime::Handle;
34
35use crate::azure::credential::{
36    AzureCliCredential, AzureCredential, ClientSecretOAuthProvider, ImdsManagedIdentityProvider,
37    WorkloadIdentityOAuthProvider,
38};
39use crate::error::generic;
40use crate::gcp::credential::{ServiceAccountCredentials, ServiceAccountKey};
41use crate::retry::RetryExt;
42use crate::service::{HttpService, make_service};
43use crate::token::jwt_exp_secs;
44use crate::{
45    ClientOptions, CredentialProvider, Result, RetryConfig, StaticCredentialProvider,
46    TemporaryToken, TokenCredentialProvider, TokenProvider,
47};
48
49/// How long a token without a known expiry is used before it is fetched again.
50const UNKNOWN_EXPIRY: Duration = Duration::from_secs(10 * 60);
51
52/// RFC 8693 token-exchange grant type.
53pub(crate) const TOKEN_EXCHANGE_GRANT: &str = "urn:ietf:params:oauth:grant-type:token-exchange";
54
55/// RFC 8693 token type of an OpenID Connect ID token.
56pub const ID_TOKEN_TYPE: &str = "urn:ietf:params:oauth:token-type:id_token";
57
58/// RFC 8693 token type of an OAuth 2.0 access token.
59pub const ACCESS_TOKEN_TYPE: &str = "urn:ietf:params:oauth:token-type:access_token";
60
61/// RFC 8693 token type of a JWT.
62pub const JWT_TOKEN_TYPE: &str = "urn:ietf:params:oauth:token-type:jwt";
63
64/// RFC 7523 JWT-bearer grant type.
65pub(crate) const JWT_BEARER_GRANT: &str = "urn:ietf:params:oauth:grant-type:jwt-bearer";
66
67/// The token endpoint of Google service accounts that do not name one.
68const GCP_DEFAULT_TOKEN_URI: &str = "https://oauth2.googleapis.com/token";
69
70const GCP_METADATA_IDENTITY_URL: &str =
71    "http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/identity";
72
73/// A bearer token, sent as `Authorization: Bearer <token>`.
74#[derive(Clone, PartialEq, Eq)]
75pub struct BearerCredential {
76    /// The token. This is secret material.
77    pub token: String,
78}
79
80impl fmt::Debug for BearerCredential {
81    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
82        f.debug_struct("BearerCredential")
83            .field("token", &"***")
84            .finish()
85    }
86}
87
88/// A shared provider of [`BearerCredential`]s.
89pub type BearerCredentialProvider = Arc<dyn CredentialProvider<Credential = BearerCredential>>;
90
91/// OAuth 2.0 client-credentials grant configuration.
92#[derive(Clone, Default)]
93pub struct ClientCredentials {
94    /// The token endpoint, e.g. `https://login.example.com/oauth2/token`.
95    pub token_url: String,
96    /// The client id.
97    pub client_id: String,
98    /// The client secret.
99    pub client_secret: String,
100    /// Space-separated scopes to request, if the issuer needs them.
101    pub scope: Option<String>,
102    /// The audience (resource) to request, if the issuer needs it (e.g. Auth0).
103    pub audience: Option<String>,
104}
105
106impl fmt::Debug for ClientCredentials {
107    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
108        f.debug_struct("ClientCredentials")
109            .field("token_url", &self.token_url)
110            .field("client_id", &self.client_id)
111            .field("client_secret", &"***")
112            .field("scope", &self.scope)
113            .field("audience", &self.audience)
114            .finish()
115    }
116}
117
118/// RFC 8693 token-exchange configuration.
119#[derive(Clone)]
120pub struct TokenExchange {
121    /// The token-exchange endpoint.
122    pub endpoint: String,
123    /// The provider of the token to exchange.
124    pub subject: BearerCredentialProvider,
125    /// The type of the subject token; defaults to [`ID_TOKEN_TYPE`].
126    pub subject_token_type: Option<String>,
127    /// The type of token to request; defaults to [`ACCESS_TOKEN_TYPE`].
128    pub requested_token_type: Option<String>,
129    /// The audience to request, if the exchange needs it.
130    pub audience: Option<String>,
131    /// Space-separated scopes to request, if the exchange needs them.
132    pub scope: Option<String>,
133}
134
135impl fmt::Debug for TokenExchange {
136    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
137        f.debug_struct("TokenExchange")
138            .field("endpoint", &self.endpoint)
139            .field("subject", &self.subject)
140            .field("subject_token_type", &self.subject_token_type)
141            .field("requested_token_type", &self.requested_token_type)
142            .field("audience", &self.audience)
143            .field("scope", &self.scope)
144            .finish()
145    }
146}
147
148/// Azure managed-identity configuration.
149#[derive(Debug, Clone, Default)]
150pub struct AzureManagedIdentity {
151    /// The resource to request a token for, e.g. `api://my-app`.
152    pub resource: String,
153    /// The client id of a user-assigned identity.
154    pub client_id: Option<String>,
155    /// The object id of a user-assigned identity.
156    pub object_id: Option<String>,
157    /// The Azure resource id of a user-assigned identity.
158    pub msi_resource_id: Option<String>,
159    /// Overrides the instance metadata token endpoint.
160    pub endpoint: Option<String>,
161}
162
163/// A GCP service-account key.
164#[derive(Clone)]
165pub enum GcpServiceAccount {
166    /// The path of a service-account JSON key file, e.g. the one named by
167    /// `GOOGLE_APPLICATION_CREDENTIALS`.
168    File(String),
169    /// The contents of a service-account JSON key file. This is secret
170    /// material.
171    Key(String),
172}
173
174/// A [`GcpServiceAccount`] read and parsed.
175pub(crate) struct LoadedServiceAccount {
176    pub(crate) credentials: ServiceAccountCredentials,
177    pub(crate) key: ServiceAccountKey,
178    /// The key's token endpoint, or Google's.
179    pub(crate) token_uri: String,
180}
181
182impl GcpServiceAccount {
183    /// Reads and parses the key. Fails unless it is a `service_account` credential;
184    /// `purpose` names what needs one, for the error.
185    pub(crate) fn load(&self, purpose: &str) -> Result<LoadedServiceAccount> {
186        let json = match self {
187            Self::File(path) => std::fs::read_to_string(path).map_err(|e| {
188                generic(format!("failed to read service-account key `{path}`: {e}"))
189            })?,
190            Self::Key(key) => key.clone(),
191        };
192        let value: serde_json::Value = serde_json::from_str(&json)
193            .map_err(|e| generic(format!("invalid service-account key {self:?}: {e}")))?;
194        if let Some(kind) = value.get("type").and_then(|kind| kind.as_str())
195            && kind != "service_account"
196        {
197            return Err(generic(format!(
198                "{self:?} holds `{kind}` credentials; {purpose} need a service-account key"
199            )));
200        }
201        let credentials: ServiceAccountCredentials = serde_json::from_value(value)
202            .map_err(|e| generic(format!("invalid service-account key {self:?}: {e}")))?;
203        let key = ServiceAccountKey::from_pem(credentials.private_key.as_bytes())?;
204        let token_uri = credentials
205            .token_uri
206            .clone()
207            .unwrap_or_else(|| GCP_DEFAULT_TOKEN_URI.to_string());
208        Ok(LoadedServiceAccount {
209            credentials,
210            key,
211            token_uri,
212        })
213    }
214}
215
216impl fmt::Debug for GcpServiceAccount {
217    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
218        match self {
219            Self::File(path) => f.debug_tuple("File").field(path).finish(),
220            Self::Key(_) => f.debug_tuple("Key").field(&"***").finish(),
221        }
222    }
223}
224
225/// Builds [`BearerCredentialProvider`]s that share HTTP settings.
226///
227/// Providers that fetch tokens over HTTP use the configured
228/// [`ClientOptions`] and [`RetryConfig`], and spawn their requests on the
229/// runtime set with [`with_runtime`](Self::with_runtime). All of them cache
230/// tokens and refresh them before they expire.
231#[derive(Debug, Clone, Default)]
232pub struct BearerAuth {
233    client_options: ClientOptions,
234    retry: RetryConfig,
235    runtime: Option<Handle>,
236}
237
238impl BearerAuth {
239    /// Create a builder with default HTTP settings.
240    pub fn new() -> Self {
241        Self::default()
242    }
243
244    /// Set the HTTP client options used for token requests.
245    pub fn with_client_options(mut self, options: ClientOptions) -> Self {
246        self.client_options = options;
247        self
248    }
249
250    /// Set the retry configuration used for token requests.
251    pub fn with_retry(mut self, retry: RetryConfig) -> Self {
252        self.retry = retry;
253        self
254    }
255
256    /// Spawn token requests on `handle`.
257    pub fn with_runtime(mut self, handle: Handle) -> Self {
258        self.runtime = Some(handle);
259        self
260    }
261
262    /// A provider that always returns `token`.
263    pub fn static_token(token: impl Into<String>) -> BearerCredentialProvider {
264        Arc::new(StaticCredentialProvider::new(BearerCredential {
265            token: token.into(),
266        }))
267    }
268
269    /// OAuth 2.0 client credentials against any token endpoint.
270    pub fn client_credentials(
271        &self,
272        config: ClientCredentials,
273    ) -> Result<BearerCredentialProvider> {
274        self.fetching(ClientCredentialsProvider(config), false)
275    }
276
277    /// A token read from the environment variable `var` on every refresh.
278    ///
279    /// A JWT is reused until shortly before its `exp`; other tokens are re-read
280    /// every few minutes.
281    pub fn token_env(&self, var: impl Into<String>) -> BearerCredentialProvider {
282        Arc::new(LocalTokenProvider::new(TokenLocation::Env(var.into())))
283    }
284
285    /// A token read from the file at `path` on every refresh, e.g. a projected
286    /// Kubernetes service-account token that rotates in place.
287    ///
288    /// A JWT is reused until shortly before its `exp`; other tokens are re-read
289    /// every few minutes.
290    pub fn token_file(&self, path: impl Into<String>) -> BearerCredentialProvider {
291        Arc::new(LocalTokenProvider::new(TokenLocation::File(path.into())))
292    }
293
294    /// An Azure AD token for `scope` (e.g. `api://my-app/.default`) from a
295    /// service principal's client secret.
296    pub fn azure_client_secret(
297        &self,
298        tenant_id: &str,
299        client_id: impl Into<String>,
300        client_secret: impl Into<String>,
301        scope: &str,
302        authority_host: Option<String>,
303    ) -> Result<BearerCredentialProvider> {
304        let provider = ClientSecretOAuthProvider::new_with_scope(
305            client_id.into(),
306            client_secret.into(),
307            tenant_id,
308            authority_host,
309            scope,
310        );
311        Ok(azure_bearer(self.azure_fetching(provider, false)?))
312    }
313
314    /// An Azure AD token from the managed identity of the host, optionally a
315    /// user-assigned one.
316    pub fn azure_managed_identity(
317        &self,
318        identity: AzureManagedIdentity,
319    ) -> Result<BearerCredentialProvider> {
320        let provider = ImdsManagedIdentityProvider::new_with_resource(
321            identity.client_id,
322            identity.object_id,
323            identity.msi_resource_id,
324            identity.endpoint,
325            &identity.resource,
326        );
327        Ok(azure_bearer(self.azure_fetching(provider, true)?))
328    }
329
330    /// An Azure AD token for `scope` from a federated token file, as used by
331    /// AKS workload identity (`AZURE_FEDERATED_TOKEN_FILE`).
332    pub fn azure_workload_identity(
333        &self,
334        tenant_id: &str,
335        client_id: impl Into<String>,
336        federated_token_file: impl Into<String>,
337        scope: &str,
338        authority_host: Option<String>,
339    ) -> Result<BearerCredentialProvider> {
340        let provider = WorkloadIdentityOAuthProvider::new(
341            client_id,
342            federated_token_file,
343            tenant_id,
344            authority_host,
345        )
346        .with_scope(scope);
347        Ok(azure_bearer(self.azure_fetching(provider, false)?))
348    }
349
350    /// An Azure AD token for `scope` from the signed-in Azure CLI
351    /// (`az account get-access-token`).
352    pub fn azure_cli(&self, scope: impl Into<String>) -> BearerCredentialProvider {
353        azure_bearer(Arc::new(AzureCliCredential::new_with_scope(scope)))
354    }
355
356    /// An ID token for `audience` from the GCP metadata server of the host's
357    /// service account.
358    ///
359    /// `endpoint` overrides the metadata identity endpoint.
360    pub fn gcp_id_token(
361        &self,
362        audience: impl Into<String>,
363        endpoint: Option<String>,
364    ) -> Result<BearerCredentialProvider> {
365        let provider = GcpIdTokenProvider {
366            audience: audience.into(),
367            endpoint: endpoint.unwrap_or_else(|| GCP_METADATA_IDENTITY_URL.to_string()),
368        };
369        self.fetching(provider, true)
370    }
371
372    /// An ID token for `audience`, signed for by a service-account key.
373    ///
374    /// The key must be a `service_account` credential: user credentials from
375    /// `gcloud auth application-default login` cannot mint ID tokens for an
376    /// arbitrary audience.
377    pub fn gcp_service_account_id_token(
378        &self,
379        account: GcpServiceAccount,
380        audience: impl Into<String>,
381    ) -> Result<BearerCredentialProvider> {
382        let LoadedServiceAccount {
383            credentials,
384            key,
385            token_uri,
386        } = account.load("ID tokens")?;
387        let provider = GcpServiceAccountIdTokenProvider {
388            credentials,
389            key,
390            token_uri,
391            audience: audience.into(),
392        };
393        self.fetching(provider, false)
394    }
395
396    /// Exchange the subject token for another token (RFC 8693), e.g. an
397    /// identity-provider ID token for an API's own access token.
398    pub fn token_exchange(&self, config: TokenExchange) -> Result<BearerCredentialProvider> {
399        self.fetching(TokenExchangeProvider(config), false)
400    }
401
402    fn http(&self, metadata: bool) -> Result<(Client, Arc<dyn HttpService>)> {
403        let client = if metadata {
404            self.client_options.metadata_client()?
405        } else {
406            self.client_options.client()?
407        };
408        let service = make_service(client.clone(), self.runtime.as_ref());
409        Ok((client, service))
410    }
411
412    fn fetching<T>(&self, provider: T, metadata: bool) -> Result<BearerCredentialProvider>
413    where
414        T: TokenProvider<Credential = BearerCredential> + 'static,
415    {
416        let (client, service) = self.http(metadata)?;
417        Ok(Arc::new(TokenCredentialProvider::new(
418            provider,
419            client,
420            service,
421            self.retry.clone(),
422        )))
423    }
424
425    fn azure_fetching<T>(
426        &self,
427        provider: T,
428        metadata: bool,
429    ) -> Result<Arc<dyn CredentialProvider<Credential = AzureCredential>>>
430    where
431        T: TokenProvider<Credential = AzureCredential> + 'static,
432    {
433        let (client, service) = self.http(metadata)?;
434        Ok(Arc::new(TokenCredentialProvider::new(
435            provider,
436            client,
437            service,
438            self.retry.clone(),
439        )))
440    }
441}
442
443/// Signs requests with a bearer token from a [`BearerCredentialProvider`].
444#[derive(Debug)]
445pub(crate) struct BearerProviderSigner(pub(crate) BearerCredentialProvider);
446
447impl crate::RequestSigner for BearerProviderSigner {
448    fn sign<'a>(
449        &'a self,
450        req: reqwest::RequestBuilder,
451    ) -> futures::future::BoxFuture<'a, Result<reqwest::RequestBuilder>> {
452        Box::pin(async move {
453            let credential = self.0.get_credential().await?;
454            Ok(req.bearer_auth(&credential.token))
455        })
456    }
457}
458
459fn azure_bearer(
460    inner: Arc<dyn CredentialProvider<Credential = AzureCredential>>,
461) -> BearerCredentialProvider {
462    Arc::new(AzureBearer(inner))
463}
464
465/// Serves an Azure AD bearer token as a [`BearerCredential`].
466#[derive(Debug)]
467struct AzureBearer(Arc<dyn CredentialProvider<Credential = AzureCredential>>);
468
469#[async_trait]
470impl CredentialProvider for AzureBearer {
471    type Credential = BearerCredential;
472
473    async fn get_credential(&self) -> Result<Arc<BearerCredential>> {
474        let credential = self.0.get_credential().await?;
475        let AzureCredential::BearerToken(token) = credential.as_ref();
476        Ok(Arc::new(BearerCredential {
477            token: token.clone(),
478        }))
479    }
480}
481
482#[derive(Debug, Deserialize)]
483struct TokenResponse {
484    access_token: String,
485    #[serde(default)]
486    expires_in: Option<u64>,
487}
488
489impl TokenResponse {
490    fn into_token(self) -> TemporaryToken<Arc<BearerCredential>> {
491        let expiry = match self.expires_in {
492            Some(seconds) => Instant::now() + Duration::from_secs(seconds),
493            None => {
494                jwt_expiry(&self.access_token).unwrap_or_else(|| Instant::now() + UNKNOWN_EXPIRY)
495            }
496        };
497        TemporaryToken {
498            token: Arc::new(BearerCredential {
499                token: self.access_token,
500            }),
501            expiry: Some(expiry),
502        }
503    }
504}
505
506#[derive(Debug)]
507struct ClientCredentialsProvider(ClientCredentials);
508
509#[async_trait]
510impl TokenProvider for ClientCredentialsProvider {
511    type Credential = BearerCredential;
512
513    async fn fetch_token(
514        &self,
515        client: &Client,
516        service: &Arc<dyn HttpService>,
517        retry: &RetryConfig,
518    ) -> Result<TemporaryToken<Arc<BearerCredential>>> {
519        let config = &self.0;
520        let mut form = vec![
521            ("grant_type", "client_credentials"),
522            ("client_id", config.client_id.as_str()),
523            ("client_secret", config.client_secret.as_str()),
524        ];
525        if let Some(scope) = &config.scope {
526            form.push(("scope", scope));
527        }
528        if let Some(audience) = &config.audience {
529            form.push(("audience", audience));
530        }
531        let response: TokenResponse = client
532            .request(Method::POST, &config.token_url)
533            .header(ACCEPT, HeaderValue::from_static("application/json"))
534            .form(&form)
535            .retryable(retry, service.clone())
536            .idempotent(true)
537            .send()
538            .await
539            .map_err(|e| e.error())?
540            .json()
541            .await?;
542        Ok(response.into_token())
543    }
544}
545
546#[derive(Debug)]
547struct TokenExchangeProvider(TokenExchange);
548
549#[async_trait]
550impl TokenProvider for TokenExchangeProvider {
551    type Credential = BearerCredential;
552
553    async fn fetch_token(
554        &self,
555        client: &Client,
556        service: &Arc<dyn HttpService>,
557        retry: &RetryConfig,
558    ) -> Result<TemporaryToken<Arc<BearerCredential>>> {
559        let config = &self.0;
560        let subject = config.subject.get_credential().await?;
561        let mut form = vec![
562            ("grant_type", TOKEN_EXCHANGE_GRANT),
563            ("subject_token", subject.token.as_str()),
564            (
565                "subject_token_type",
566                config
567                    .subject_token_type
568                    .as_deref()
569                    .unwrap_or(ID_TOKEN_TYPE),
570            ),
571            (
572                "requested_token_type",
573                config
574                    .requested_token_type
575                    .as_deref()
576                    .unwrap_or(ACCESS_TOKEN_TYPE),
577            ),
578        ];
579        if let Some(audience) = &config.audience {
580            form.push(("audience", audience));
581        }
582        if let Some(scope) = &config.scope {
583            form.push(("scope", scope));
584        }
585        let response: TokenResponse = client
586            .request(Method::POST, &config.endpoint)
587            .header(ACCEPT, HeaderValue::from_static("application/json"))
588            .form(&form)
589            .retryable(retry, service.clone())
590            .idempotent(true)
591            .send()
592            .await
593            .map_err(|e| e.error())?
594            .json()
595            .await?;
596        Ok(response.into_token())
597    }
598}
599
600#[derive(Debug)]
601struct GcpIdTokenProvider {
602    audience: String,
603    endpoint: String,
604}
605
606#[async_trait]
607impl TokenProvider for GcpIdTokenProvider {
608    type Credential = BearerCredential;
609
610    async fn fetch_token(
611        &self,
612        client: &Client,
613        service: &Arc<dyn HttpService>,
614        retry: &RetryConfig,
615    ) -> Result<TemporaryToken<Arc<BearerCredential>>> {
616        let token = client
617            .request(Method::GET, &self.endpoint)
618            .header("Metadata-Flavor", "Google")
619            .query(&[("audience", self.audience.as_str()), ("format", "full")])
620            .send_retry(retry, service.clone())
621            .await
622            .map_err(|e| e.error())?
623            .text()
624            .await?
625            .trim()
626            .to_string();
627        let expiry = jwt_expiry(&token).unwrap_or_else(|| Instant::now() + UNKNOWN_EXPIRY);
628        Ok(TemporaryToken {
629            token: Arc::new(BearerCredential { token }),
630            expiry: Some(expiry),
631        })
632    }
633}
634
635struct GcpServiceAccountIdTokenProvider {
636    credentials: ServiceAccountCredentials,
637    key: ServiceAccountKey,
638    token_uri: String,
639    audience: String,
640}
641
642impl fmt::Debug for GcpServiceAccountIdTokenProvider {
643    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
644        f.debug_struct("GcpServiceAccountIdTokenProvider")
645            .field("client_email", &self.credentials.client_email)
646            .field("token_uri", &self.token_uri)
647            .field("audience", &self.audience)
648            .finish_non_exhaustive()
649    }
650}
651
652#[async_trait]
653impl TokenProvider for GcpServiceAccountIdTokenProvider {
654    type Credential = BearerCredential;
655
656    async fn fetch_token(
657        &self,
658        client: &Client,
659        service: &Arc<dyn HttpService>,
660        retry: &RetryConfig,
661    ) -> Result<TemporaryToken<Arc<BearerCredential>>> {
662        #[derive(Deserialize)]
663        struct IdTokenResponse {
664            id_token: String,
665        }
666
667        let assertion =
668            self.credentials
669                .id_token_assertion(&self.key, &self.token_uri, &self.audience)?;
670        let response: IdTokenResponse = client
671            .request(Method::POST, &self.token_uri)
672            .header(ACCEPT, HeaderValue::from_static("application/json"))
673            .form(&[("grant_type", JWT_BEARER_GRANT), ("assertion", &assertion)])
674            .retryable(retry, service.clone())
675            .idempotent(true)
676            .send()
677            .await
678            .map_err(|e| e.error())?
679            .json()
680            .await?;
681        let token = response.id_token;
682        let expiry = jwt_expiry(&token).unwrap_or_else(|| Instant::now() + UNKNOWN_EXPIRY);
683        Ok(TemporaryToken {
684            token: Arc::new(BearerCredential { token }),
685            expiry: Some(expiry),
686        })
687    }
688}
689
690#[derive(Debug)]
691enum TokenLocation {
692    Env(String),
693    File(String),
694}
695
696/// Reads a token from the environment or a file, re-reading it near expiry.
697#[derive(Debug)]
698struct LocalTokenProvider {
699    location: TokenLocation,
700    cache: crate::TokenCache<Arc<BearerCredential>>,
701}
702
703impl LocalTokenProvider {
704    fn new(location: TokenLocation) -> Self {
705        Self {
706            location,
707            cache: Default::default(),
708        }
709    }
710
711    fn read(&self) -> Result<TemporaryToken<Arc<BearerCredential>>> {
712        let token = match &self.location {
713            TokenLocation::Env(var) => std::env::var(var)
714                .map_err(|_| generic(format!("environment variable `{var}` is not set")))?,
715            TokenLocation::File(path) => std::fs::read_to_string(path)
716                .map_err(|e| generic(format!("failed to read token file `{path}`: {e}")))?,
717        };
718        let token = token.trim().to_string();
719        if token.is_empty() {
720            return Err(generic(format!(
721                "the token in {:?} is empty",
722                self.location
723            )));
724        }
725        let expiry = jwt_expiry(&token).unwrap_or_else(|| Instant::now() + UNKNOWN_EXPIRY);
726        Ok(TemporaryToken {
727            token: Arc::new(BearerCredential { token }),
728            expiry: Some(expiry),
729        })
730    }
731}
732
733#[async_trait]
734impl CredentialProvider for LocalTokenProvider {
735    type Credential = BearerCredential;
736
737    async fn get_credential(&self) -> Result<Arc<BearerCredential>> {
738        self.cache
739            .get_or_insert_with(|| async { self.read() })
740            .await
741    }
742}
743
744/// The `exp` of a JWT as an [`Instant`], if `token` is a JWT with one.
745fn jwt_expiry(token: &str) -> Option<Instant> {
746    let exp = jwt_exp_secs(token)?;
747    let now = SystemTime::now().duration_since(UNIX_EPOCH).ok()?.as_secs();
748    Some(Instant::now() + Duration::from_secs(exp.saturating_sub(now)))
749}
750
751#[cfg(test)]
752mod tests {
753    use base64::Engine;
754    use mockito::Matcher;
755
756    use super::*;
757    use crate::{ClientConfigKey, CloudClient};
758
759    fn auth() -> BearerAuth {
760        BearerAuth::new().with_client_options(
761            ClientOptions::new().with_config(ClientConfigKey::AllowHttp, "true"),
762        )
763    }
764
765    /// A JWT (unsigned) whose payload is `{"exp": <exp>}`.
766    fn jwt(exp: u64) -> String {
767        let payload =
768            base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(format!(r#"{{"exp":{exp}}}"#));
769        format!("eyJhbGciOiJub25lIn0.{payload}.")
770    }
771
772    fn unix_now() -> u64 {
773        SystemTime::now()
774            .duration_since(UNIX_EPOCH)
775            .unwrap()
776            .as_secs()
777    }
778
779    /// Send one request through a client signed by `provider` and assert the
780    /// bearer it carried.
781    async fn assert_signs(
782        server: &mut mockito::Server,
783        provider: BearerCredentialProvider,
784        token: &str,
785    ) {
786        let api = server
787            .mock("GET", "/api")
788            .match_header("authorization", format!("Bearer {token}").as_str())
789            .with_status(200)
790            .create_async()
791            .await;
792        let options = ClientOptions::new().with_config(ClientConfigKey::AllowHttp, "true");
793        let client =
794            CloudClient::new_with_bearer(provider, &options, RetryConfig::default(), None).unwrap();
795        let status = client
796            .get(format!("{}/api", server.url()))
797            .send()
798            .await
799            .unwrap()
800            .status();
801        assert_eq!(status, 200);
802        api.assert_async().await;
803    }
804
805    #[tokio::test]
806    async fn client_credentials_against_any_token_endpoint() {
807        let mut server = mockito::Server::new_async().await;
808        let token = server
809            .mock("POST", "/oauth2/token")
810            .match_body(Matcher::AllOf(vec![
811                Matcher::UrlEncoded("grant_type".into(), "client_credentials".into()),
812                Matcher::UrlEncoded("client_id".into(), "app".into()),
813                Matcher::UrlEncoded("client_secret".into(), "secret".into()),
814                Matcher::UrlEncoded("scope".into(), "catalog.read".into()),
815                Matcher::UrlEncoded("audience".into(), "https://uc.example.com".into()),
816            ]))
817            .with_header("content-type", "application/json")
818            .with_body(r#"{"access_token":"cc-token","expires_in":3600}"#)
819            .expect(1)
820            .create_async()
821            .await;
822
823        let provider = auth()
824            .client_credentials(ClientCredentials {
825                token_url: format!("{}/oauth2/token", server.url()),
826                client_id: "app".into(),
827                client_secret: "secret".into(),
828                scope: Some("catalog.read".into()),
829                audience: Some("https://uc.example.com".into()),
830            })
831            .unwrap();
832        assert_signs(&mut server, provider, "cc-token").await;
833        token.assert_async().await;
834    }
835
836    /// The subject token is exchanged per RFC 8693 and the issued token signs.
837    #[tokio::test]
838    async fn token_exchange_swaps_the_subject_token() {
839        let mut server = mockito::Server::new_async().await;
840        let exchanged = jwt(unix_now() + 3600);
841        let exchange = server
842            .mock("POST", "/api/1.0/unity-control/auth/tokens")
843            .match_body(Matcher::AllOf(vec![
844                Matcher::UrlEncoded("grant_type".into(), TOKEN_EXCHANGE_GRANT.into()),
845                Matcher::UrlEncoded("subject_token".into(), "idp-id-token".into()),
846                Matcher::UrlEncoded("subject_token_type".into(), ID_TOKEN_TYPE.into()),
847                Matcher::UrlEncoded("requested_token_type".into(), ACCESS_TOKEN_TYPE.into()),
848            ]))
849            .with_header("content-type", "application/json")
850            // No `expires_in`: the JWT's `exp` decides when to refresh.
851            .with_body(format!(
852                r#"{{"access_token":"{exchanged}","token_type":"Bearer"}}"#
853            ))
854            .expect(1)
855            .create_async()
856            .await;
857
858        let provider = auth()
859            .token_exchange(TokenExchange {
860                endpoint: format!("{}/api/1.0/unity-control/auth/tokens", server.url()),
861                subject: BearerAuth::static_token("idp-id-token"),
862                subject_token_type: None,
863                requested_token_type: None,
864                audience: None,
865                scope: None,
866            })
867            .unwrap();
868        // Two calls, one exchange: the issued token is cached until near `exp`.
869        provider.get_credential().await.unwrap();
870        assert_signs(&mut server, provider, &exchanged).await;
871        exchange.assert_async().await;
872    }
873
874    #[tokio::test]
875    async fn azure_client_secret_uses_the_given_scope() {
876        let mut server = mockito::Server::new_async().await;
877        let aad = server
878            .mock("POST", "/tenant-1/oauth2/v2.0/token")
879            .match_body(Matcher::UrlEncoded(
880                "scope".into(),
881                "api://my-catalog/.default".into(),
882            ))
883            .with_header("content-type", "application/json")
884            .with_body(r#"{"access_token":"aad-token","expires_in":3600}"#)
885            .create_async()
886            .await;
887
888        let provider = auth()
889            .azure_client_secret(
890                "tenant-1",
891                "app",
892                "secret",
893                "api://my-catalog/.default",
894                Some(server.url()),
895            )
896            .unwrap();
897        assert_signs(&mut server, provider, "aad-token").await;
898        aad.assert_async().await;
899    }
900
901    #[tokio::test]
902    async fn azure_managed_identity_uses_the_given_resource() {
903        let mut server = mockito::Server::new_async().await;
904        let imds = server
905            .mock("GET", "/metadata/identity/oauth2/token")
906            .match_query(Matcher::AllOf(vec![
907                Matcher::UrlEncoded("resource".into(), "api://my-catalog".into()),
908                Matcher::UrlEncoded("client_id".into(), "uami".into()),
909            ]))
910            .match_header("metadata", "true")
911            .with_header("content-type", "application/json")
912            .with_body(format!(
913                r#"{{"access_token":"msi-token","expires_on":"{}","expires_in":"3600"}}"#,
914                unix_now() + 3600
915            ))
916            .create_async()
917            .await;
918
919        let provider = auth()
920            .azure_managed_identity(AzureManagedIdentity {
921                resource: "api://my-catalog".into(),
922                client_id: Some("uami".into()),
923                endpoint: Some(format!("{}/metadata/identity/oauth2/token", server.url())),
924                ..Default::default()
925            })
926            .unwrap();
927        assert_signs(&mut server, provider, "msi-token").await;
928        imds.assert_async().await;
929    }
930
931    #[test]
932    fn gcp_service_account_id_token_rejects_user_credentials() {
933        let key = r#"{"type":"authorized_user","client_id":"id","client_secret":"secret","refresh_token":"refresh"}"#;
934        let err = auth()
935            .gcp_service_account_id_token(GcpServiceAccount::Key(key.into()), "aud")
936            .unwrap_err();
937        assert!(err.to_string().contains("`authorized_user`"), "{err}");
938    }
939
940    #[test]
941    fn gcp_service_account_id_token_validates_the_key() {
942        let missing = auth()
943            .gcp_service_account_id_token(
944                GcpServiceAccount::File("/nonexistent.json".into()),
945                "aud",
946            )
947            .unwrap_err();
948        assert!(
949            missing.to_string().contains("/nonexistent.json"),
950            "{missing}"
951        );
952
953        let key = r#"{"type":"service_account","private_key":"not a key","private_key_id":"kid","client_email":"sa@example.iam.gserviceaccount.com"}"#;
954        assert!(
955            auth()
956                .gcp_service_account_id_token(GcpServiceAccount::Key(key.into()), "aud")
957                .is_err()
958        );
959        let debug = format!("{:?}", GcpServiceAccount::Key(key.into()));
960        assert!(!debug.contains("private_key"), "{debug}");
961    }
962
963    #[tokio::test]
964    async fn gcp_id_token_requests_the_audience() {
965        let mut server = mockito::Server::new_async().await;
966        let id_token = jwt(unix_now() + 3600);
967        let metadata = server
968            .mock("GET", "/identity")
969            .match_query(Matcher::UrlEncoded(
970                "audience".into(),
971                "https://uc.example.com".into(),
972            ))
973            .match_header("metadata-flavor", "Google")
974            .with_body(&id_token)
975            .create_async()
976            .await;
977
978        let provider = auth()
979            .gcp_id_token(
980                "https://uc.example.com",
981                Some(format!("{}/identity", server.url())),
982            )
983            .unwrap();
984        assert_signs(&mut server, provider, &id_token).await;
985        metadata.assert_async().await;
986    }
987
988    #[tokio::test]
989    async fn token_file_is_trimmed_and_missing_files_error() {
990        let dir = tempfile::tempdir().unwrap();
991        let path = dir.path().join("token");
992        std::fs::write(&path, "file-token\n").unwrap();
993
994        let provider = auth().token_file(path.to_string_lossy());
995        assert_eq!(provider.get_credential().await.unwrap().token, "file-token");
996
997        let missing = auth().token_file(dir.path().join("absent").to_string_lossy());
998        assert!(missing.get_credential().await.is_err());
999    }
1000
1001    #[test]
1002    fn jwt_expiry_reads_exp_and_ignores_opaque_tokens() {
1003        let in_an_hour = jwt_expiry(&jwt(unix_now() + 3600)).unwrap();
1004        let remaining = in_an_hour.saturating_duration_since(Instant::now());
1005        assert!(remaining > Duration::from_secs(3500) && remaining <= Duration::from_secs(3600));
1006        assert!(jwt_expiry("opaque-token").is_none());
1007        assert!(jwt_expiry("a.not-base64!.c").is_none());
1008    }
1009
1010    #[test]
1011    fn debug_redacts_secrets() {
1012        let credential = BearerCredential {
1013            token: "super-secret".into(),
1014        };
1015        let config = ClientCredentials {
1016            client_secret: "super-secret".into(),
1017            ..Default::default()
1018        };
1019        for debug in [format!("{credential:?}"), format!("{config:?}")] {
1020            assert!(!debug.contains("super-secret"), "secret leaked: {debug}");
1021        }
1022    }
1023}
olai_http/bearer.rs

olai_http/
bearer.rs
