cloud_sdk_reqwest/asynchronous/
client.rs

1use core::fmt;
2use std::sync::Arc;
3
4use cloud_sdk::authentication::{
5    AsyncAuthenticatedTransport, AuthenticatedRequest, CredentialGeneration,
6};
7use cloud_sdk::transport::{
8    BoundTransport, EndpointIdentity, EndpointIdentityError, ResponseStorageSanitizer,
9    ResponseWriter, TransportFailure,
10};
11use cloud_sdk_sanitization::{SecretBuffer, sanitize_bytes};
12
13use crate::shared::{
14    AuthenticatedTransportFailure, BearerCredential, BearerCredentialScope,
15    BearerCredentialSnapshot, BearerRefreshHandoff, BearerToken, CredentialStateError,
16    CredentialStore, CredentialUpdateError, HttpsEndpoint, TokenRefreshError, TokenRotationError,
17    TransportError, map_authentication_error, validate_bearer_authentication,
18};
19
20use super::RawAsyncClient;
21
22/// Hardened provider-neutral reqwest asynchronous bearer transport.
23///
24/// The adapter uses reqwest's Tokio-based execution internally but does not
25/// install or own a runtime. Callers must poll it from a compatible executor.
26#[derive(Clone)]
27pub struct AsyncClient {
28    client: RawAsyncClient,
29    endpoint: HttpsEndpoint,
30    scope: Arc<BearerCredentialScope>,
31    credentials: Arc<CredentialStore>,
32    allow_insecure_loopback: bool,
33}
34
35impl AsyncClient {
36    pub(super) fn new(
37        client: RawAsyncClient,
38        endpoint: HttpsEndpoint,
39        credential: BearerCredential,
40        allow_insecure_loopback: bool,
41    ) -> Self {
42        Self {
43            client,
44            endpoint,
45            scope: Arc::new(credential.scope),
46            credentials: Arc::new(CredentialStore::new(credential.token)),
47            allow_insecure_loopback,
48        }
49    }
50
51    /// Captures the current generation without exposing token bytes.
52    pub fn credential_snapshot(&self) -> Result<BearerCredentialSnapshot, CredentialStateError> {
53        self.credentials.snapshot()
54    }
55
56    /// Atomically replaces the token while retaining immutable scope.
57    pub fn rotate_bearer_token(
58        &self,
59        replacement: BearerToken,
60    ) -> Result<CredentialGeneration, CredentialUpdateError> {
61        self.credentials.rotate(replacement)
62    }
63
64    /// Validates and rotates mutable bytes, clearing the complete source.
65    pub fn rotate_bearer_token_from_mut_bytes(
66        &self,
67        source: &mut [u8],
68    ) -> Result<CredentialGeneration, TokenRotationError> {
69        self.credentials.rotate_from_mut_bytes(source)
70    }
71
72    /// Validates and rotates guarded storage, which clears on return.
73    pub fn rotate_bearer_token_from_secret_buffer(
74        &self,
75        source: SecretBuffer<'_>,
76    ) -> Result<CredentialGeneration, TokenRotationError> {
77        self.credentials.rotate_from_secret_buffer(source)
78    }
79
80    /// Installs a refresh only if its captured generation is still current.
81    pub fn refresh_bearer_token(
82        &self,
83        handoff: BearerRefreshHandoff,
84        replacement: BearerToken,
85    ) -> Result<CredentialGeneration, TokenRefreshError> {
86        self.credentials.refresh(handoff, replacement)
87    }
88
89    /// Validates refreshed mutable bytes, clears them, and rejects stale work.
90    pub fn refresh_bearer_token_from_mut_bytes(
91        &self,
92        handoff: BearerRefreshHandoff,
93        source: &mut [u8],
94    ) -> Result<CredentialGeneration, TokenRefreshError> {
95        self.credentials.refresh_from_mut_bytes(handoff, source)
96    }
97
98    /// Consumes guarded refreshed storage and rejects stale work.
99    pub fn refresh_bearer_token_from_secret_buffer(
100        &self,
101        handoff: BearerRefreshHandoff,
102        source: SecretBuffer<'_>,
103    ) -> Result<CredentialGeneration, TokenRefreshError> {
104        self.credentials.refresh_from_secret_buffer(handoff, source)
105    }
106
107    async fn send_inner(
108        &self,
109        authenticated: AuthenticatedRequest<'_, '_>,
110        response_writer: &mut ResponseWriter<'_>,
111    ) -> Result<(), AuthenticatedTransportFailure> {
112        let endpoint_identity = self.endpoint.identity().map_err(|_| {
113            TransportFailure::not_sent(TransportError::AuthenticationEndpointMismatch)
114        })?;
115        validate_bearer_authentication(
116            endpoint_identity,
117            &self.scope,
118            authenticated.policy(),
119            self.allow_insecure_loopback,
120        )
121        .map_err(|error| TransportFailure::not_sent(map_authentication_error(error)))?;
122        let token_snapshot = self
123            .credentials
124            .snapshot()
125            .map_err(|_| TransportFailure::not_sent(TransportError::CredentialStateUnavailable))?;
126        let authorization = token_snapshot
127            .header_value()
128            .map_err(|_| TransportFailure::not_sent(TransportError::HeaderRejected))?;
129        drop(token_snapshot);
130        self.client
131            .execute_authenticated(
132                authenticated.transport_request(),
133                authenticated.response_policy(),
134                authorization,
135                response_writer,
136            )
137            .await
138            .map_err(|failure| failure.map(TransportError::RawHttp))
139    }
140}
141
142impl AsyncAuthenticatedTransport for AsyncClient {
143    type Error = AuthenticatedTransportFailure;
144
145    async fn send_authenticated<'transport, 'request, 'policy, 'writer>(
146        &'transport self,
147        request: AuthenticatedRequest<'request, 'policy>,
148        response: &'writer mut ResponseWriter<'_>,
149    ) -> Result<(), Self::Error>
150    where
151        'transport: 'writer,
152        'request: 'writer,
153        'policy: 'writer,
154    {
155        self.send_inner(request, response).await
156    }
157}
158
159impl ResponseStorageSanitizer for AsyncClient {
160    fn sanitize_response_storage(&self, response_storage: &mut [u8]) {
161        sanitize_bytes(response_storage);
162    }
163}
164
165impl BoundTransport for AsyncClient {
166    fn endpoint_identity(&self) -> Result<EndpointIdentity<'_>, EndpointIdentityError> {
167        self.endpoint.identity()
168    }
169}
170
171impl fmt::Debug for AsyncClient {
172    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
173        formatter
174            .debug_struct("AsyncClient")
175            .field("endpoint", &"[redacted]")
176            .field("scope", &"[redacted]")
177            .field("credentials", &"[redacted]")
178            .finish_non_exhaustive()
179    }
180}
cloud_sdk_reqwest/asynchronous/client.rs

cloud_sdk_reqwest/asynchronous/
client.rs
