cloud_sdk_reqwest/asynchronous/
client.rs

1use core::fmt;
2use std::sync::Arc;
3
4use cloud_sdk::authentication::{
5    AsyncAuthenticatedTransport, AuthenticatedRequest, CredentialGeneration, CredentialLifetime,
6};
7use cloud_sdk::transport::{
8    AsyncResponseStaging, BoundTransport, EndpointIdentity, EndpointIdentityError,
9    ResponseCompletion, ResponseStorageSanitizer, 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, credential.lifetime)),
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    /// Atomically replaces an expiring token and its complete lifetime.
65    pub fn rotate_bearer_token_with_lifetime(
66        &self,
67        replacement: BearerToken,
68        lifetime: CredentialLifetime,
69    ) -> Result<CredentialGeneration, CredentialUpdateError> {
70        self.credentials.rotate_with_lifetime(replacement, lifetime)
71    }
72
73    /// Validates and rotates mutable bytes, clearing the complete source.
74    pub fn rotate_bearer_token_from_mut_bytes(
75        &self,
76        source: &mut [u8],
77    ) -> Result<CredentialGeneration, TokenRotationError> {
78        self.credentials.rotate_from_mut_bytes(source)
79    }
80
81    /// Validates and rotates guarded storage, which clears on return.
82    pub fn rotate_bearer_token_from_secret_buffer(
83        &self,
84        source: SecretBuffer<'_>,
85    ) -> Result<CredentialGeneration, TokenRotationError> {
86        self.credentials.rotate_from_secret_buffer(source)
87    }
88
89    /// Clears mutable input and atomically installs an expiring replacement.
90    pub fn rotate_bearer_token_from_mut_bytes_with_lifetime(
91        &self,
92        source: &mut [u8],
93        lifetime: CredentialLifetime,
94    ) -> Result<CredentialGeneration, TokenRotationError> {
95        self.credentials
96            .rotate_from_mut_bytes_with_lifetime(source, lifetime)
97    }
98
99    /// Consumes guarded input and atomically installs an expiring replacement.
100    pub fn rotate_bearer_token_from_secret_buffer_with_lifetime(
101        &self,
102        source: SecretBuffer<'_>,
103        lifetime: CredentialLifetime,
104    ) -> Result<CredentialGeneration, TokenRotationError> {
105        self.credentials
106            .rotate_from_secret_buffer_with_lifetime(source, lifetime)
107    }
108
109    /// Installs a refresh only if its captured generation is still current.
110    pub fn refresh_bearer_token(
111        &self,
112        handoff: BearerRefreshHandoff,
113        replacement: BearerToken,
114    ) -> Result<CredentialGeneration, TokenRefreshError> {
115        self.credentials.refresh(handoff, replacement)
116    }
117
118    /// Installs an expiring refresh if its time-qualified handoff is current.
119    pub fn refresh_bearer_token_with_lifetime(
120        &self,
121        handoff: BearerRefreshHandoff,
122        replacement: BearerToken,
123        lifetime: CredentialLifetime,
124    ) -> Result<CredentialGeneration, TokenRefreshError> {
125        self.credentials
126            .refresh_with_lifetime(handoff, replacement, lifetime)
127    }
128
129    /// Validates refreshed mutable bytes, clears them, and rejects stale work.
130    pub fn refresh_bearer_token_from_mut_bytes(
131        &self,
132        handoff: BearerRefreshHandoff,
133        source: &mut [u8],
134    ) -> Result<CredentialGeneration, TokenRefreshError> {
135        self.credentials.refresh_from_mut_bytes(handoff, source)
136    }
137
138    /// Consumes guarded refreshed storage and rejects stale work.
139    pub fn refresh_bearer_token_from_secret_buffer(
140        &self,
141        handoff: BearerRefreshHandoff,
142        source: SecretBuffer<'_>,
143    ) -> Result<CredentialGeneration, TokenRefreshError> {
144        self.credentials.refresh_from_secret_buffer(handoff, source)
145    }
146
147    /// Clears mutable refresh input and atomically installs its lifetime.
148    pub fn refresh_bearer_token_from_mut_bytes_with_lifetime(
149        &self,
150        handoff: BearerRefreshHandoff,
151        source: &mut [u8],
152        lifetime: CredentialLifetime,
153    ) -> Result<CredentialGeneration, TokenRefreshError> {
154        self.credentials
155            .refresh_from_mut_bytes_with_lifetime(handoff, source, lifetime)
156    }
157
158    /// Consumes guarded refresh input and atomically installs its lifetime.
159    pub fn refresh_bearer_token_from_secret_buffer_with_lifetime(
160        &self,
161        handoff: BearerRefreshHandoff,
162        source: SecretBuffer<'_>,
163        lifetime: CredentialLifetime,
164    ) -> Result<CredentialGeneration, TokenRefreshError> {
165        self.credentials
166            .refresh_from_secret_buffer_with_lifetime(handoff, source, lifetime)
167    }
168
169    async fn send_inner<'writer, 'buffer>(
170        &self,
171        authenticated: AuthenticatedRequest<'_, '_>,
172        response: AsyncResponseStaging<'writer, 'buffer>,
173    ) -> Result<ResponseCompletion, AuthenticatedTransportFailure> {
174        let endpoint_identity = self.endpoint.identity().map_err(|_| {
175            TransportFailure::not_sent(TransportError::AuthenticationEndpointMismatch)
176        })?;
177        validate_bearer_authentication(
178            endpoint_identity,
179            &self.scope,
180            authenticated.policy(),
181            self.allow_insecure_loopback,
182        )
183        .map_err(|error| TransportFailure::not_sent(map_authentication_error(error)))?;
184        let token_snapshot = self
185            .credentials
186            .snapshot()
187            .map_err(|_| TransportFailure::not_sent(TransportError::CredentialStateUnavailable))?;
188        let authorization = token_snapshot
189            .header_value()
190            .map_err(|_| TransportFailure::not_sent(TransportError::HeaderRejected))?;
191        drop(token_snapshot);
192        self.client
193            .execute_authenticated(
194                authenticated.transport_request(),
195                authenticated.response_policy(),
196                authorization,
197                response,
198            )
199            .await
200            .map_err(|failure| failure.map(TransportError::RawHttp))
201    }
202}
203
204impl AsyncAuthenticatedTransport for AsyncClient {
205    type Error = AuthenticatedTransportFailure;
206
207    async fn send_authenticated<'transport, 'request, 'policy, 'writer, 'buffer>(
208        &'transport self,
209        request: AuthenticatedRequest<'request, 'policy>,
210        response: AsyncResponseStaging<'writer, 'buffer>,
211    ) -> Result<ResponseCompletion, Self::Error>
212    where
213        'transport: 'writer,
214        'request: 'writer,
215        'policy: 'writer,
216        'buffer: 'writer,
217    {
218        self.send_inner(request, response).await
219    }
220}
221
222impl ResponseStorageSanitizer for AsyncClient {
223    fn sanitize_response_storage(&self, response_storage: &mut [u8]) {
224        sanitize_bytes(response_storage);
225    }
226}
227
228impl BoundTransport for AsyncClient {
229    fn endpoint_identity(&self) -> Result<EndpointIdentity<'_>, EndpointIdentityError> {
230        self.endpoint.identity()
231    }
232}
233
234impl fmt::Debug for AsyncClient {
235    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
236        formatter
237            .debug_struct("AsyncClient")
238            .field("endpoint", &"[redacted]")
239            .field("scope", &"[redacted]")
240            .field("credentials", &"[redacted]")
241            .finish_non_exhaustive()
242    }
243}
cloud_sdk_reqwest/asynchronous/client.rs

cloud_sdk_reqwest/asynchronous/
client.rs
