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

olai_http/
bearer.rs
