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

olai_http/
bearer.rs
