cloud_sdk_reqwest/asynchronous/
client.rs

1use core::fmt;
2use std::sync::Arc;
3
4use cloud_sdk::Method;
5use cloud_sdk::transport::{
6    AsyncTransport, BoundTransport, EndpointIdentity, EndpointIdentityError,
7    ResponseStorageSanitizer, StatusCode, TransportRequest, TransportResponse,
8};
9use cloud_sdk_sanitization::{SecretBuffer, sanitize_bytes};
10use reqwest::header::{ACCEPT, AUTHORIZATION, CONTENT_TYPE, HeaderValue};
11use reqwest::{Body, Client};
12
13use crate::shared::{
14    BearerToken, CredentialStateError, CredentialStore, HttpsEndpoint, TokenRotationError,
15    TransportError, parse_rate_limit, parse_response_content_type,
16};
17
18use super::body::SanitizedBuffer;
19
20/// Hardened provider-neutral reqwest asynchronous transport.
21///
22/// The adapter uses reqwest's Tokio-based execution internally but does not
23/// install or own a runtime. Callers must poll it from a compatible executor.
24#[derive(Clone)]
25pub struct AsyncClient {
26    client: Client,
27    endpoint: HttpsEndpoint,
28    credentials: Arc<CredentialStore>,
29}
30
31impl AsyncClient {
32    pub(super) fn new(client: Client, endpoint: HttpsEndpoint, token: BearerToken) -> Self {
33        Self {
34            client,
35            endpoint,
36            credentials: Arc::new(CredentialStore::new(token)),
37        }
38    }
39
40    /// Atomically replaces the bearer token used by newly started requests.
41    ///
42    /// In-flight requests retain their previous snapshot. No credential lock
43    /// is held across network I/O or `.await`.
44    pub fn rotate_bearer_token(
45        &self,
46        replacement: BearerToken,
47    ) -> Result<(), CredentialStateError> {
48        self.credentials.rotate(replacement)
49    }
50
51    /// Validates and rotates from mutable bytes, clearing the complete source
52    /// on success or failure. Rejected input leaves the active token unchanged.
53    pub fn rotate_bearer_token_from_mut_bytes(
54        &self,
55        source: &mut [u8],
56    ) -> Result<(), TokenRotationError> {
57        self.credentials.rotate_from_mut_bytes(source)
58    }
59
60    /// Validates and rotates from guarded storage. Dropping the consumed guard
61    /// clears the complete source on success or failure.
62    pub fn rotate_bearer_token_from_secret_buffer(
63        &self,
64        source: SecretBuffer<'_>,
65    ) -> Result<(), TokenRotationError> {
66        self.credentials.rotate_from_secret_buffer(source)
67    }
68
69    async fn send_inner<'buffer>(
70        &self,
71        request: TransportRequest<'_>,
72        response_body: &'buffer mut [u8],
73    ) -> Result<TransportResponse<'buffer>, TransportError> {
74        sanitize_bytes(response_body);
75        let url = self
76            .endpoint
77            .compose(request.target())
78            .map_err(|_| TransportError::TargetRejected)?;
79        let token_snapshot = self
80            .credentials
81            .snapshot()
82            .map_err(|_| TransportError::CredentialStateUnavailable)?;
83        let authorization = token_snapshot
84            .header_value()
85            .map_err(|_| TransportError::HeaderRejected)?;
86        let mut outbound = self
87            .client
88            .request(map_method(request.method()), url)
89            .header(AUTHORIZATION, authorization)
90            .header(ACCEPT, HeaderValue::from_static("application/json"));
91
92        if let Some(content_type) = request.content_type() {
93            let value = HeaderValue::from_str(content_type.as_str())
94                .map_err(|_| TransportError::HeaderRejected)?;
95            outbound = outbound.header(CONTENT_TYPE, value);
96        } else if !request.body().is_empty() {
97            return Err(TransportError::MissingContentType);
98        }
99
100        if !request.body().is_empty() {
101            let body = SanitizedBuffer::copy_from(request.body())
102                .map_err(|_| TransportError::RequestBodyAllocationFailed)?;
103            let _ = u64::try_from(request.body().len())
104                .map_err(|_| TransportError::RequestBodyTooLarge)?;
105            outbound = outbound.body(Body::from(body.into_bytes()));
106        }
107
108        let mut response = outbound.send().await.map_err(classify_reqwest_error)?;
109        self.endpoint
110            .verify_origin(response.url())
111            .map_err(|_| TransportError::ResponseOriginChanged)?;
112        if response.content_length().is_some_and(|length| {
113            u64::try_from(response_body.len()).map_or(true, |cap| length > cap)
114        }) {
115            return Err(TransportError::ResponseTooLarge);
116        }
117        let status =
118            StatusCode::new(response.status().as_u16()).ok_or(TransportError::InvalidStatus)?;
119        let rate_limit = parse_rate_limit(response.headers())?;
120        let content_type = parse_response_content_type(response.headers())?;
121        let buffered = read_response(&mut response, response_body.len()).await?;
122        let body_len = buffered.len();
123        let initialized = response_body
124            .get_mut(..body_len)
125            .ok_or(TransportError::ResponseReadFailed)?;
126        initialized.copy_from_slice(buffered.as_ref());
127        let response = TransportResponse::new(status, initialized);
128        let response = content_type.map_or(response, |value| response.with_content_type(value));
129        drop(token_snapshot);
130        Ok(rate_limit.map_or(response, |value| response.with_rate_limit(value)))
131    }
132}
133
134impl AsyncTransport for AsyncClient {
135    type Error = TransportError;
136
137    async fn send<'transport, 'request, 'buffer>(
138        &'transport self,
139        request: TransportRequest<'request>,
140        response_body: &'buffer mut [u8],
141    ) -> Result<TransportResponse<'buffer>, Self::Error>
142    where
143        'request: 'transport,
144        'buffer: 'transport,
145    {
146        self.send_inner(request, response_body).await
147    }
148}
149
150impl ResponseStorageSanitizer for AsyncClient {
151    fn sanitize_response_storage(&self, response_storage: &mut [u8]) {
152        sanitize_bytes(response_storage);
153    }
154}
155
156impl BoundTransport for AsyncClient {
157    fn endpoint_identity(&self) -> Result<EndpointIdentity<'_>, EndpointIdentityError> {
158        self.endpoint.identity()
159    }
160}
161
162impl fmt::Debug for AsyncClient {
163    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
164        formatter
165            .debug_struct("AsyncClient")
166            .field("endpoint", &"[redacted]")
167            .field("credentials", &"[redacted]")
168            .finish_non_exhaustive()
169    }
170}
171
172async fn read_response(
173    response: &mut reqwest::Response,
174    limit: usize,
175) -> Result<SanitizedBuffer, TransportError> {
176    let mut buffered = SanitizedBuffer::with_capacity(limit)
177        .map_err(|_| TransportError::ResponseBodyAllocationFailed)?;
178    loop {
179        let chunk = response
180            .chunk()
181            .await
182            .map_err(|_| TransportError::ResponseReadFailed)?;
183        let Some(chunk) = chunk else { break };
184        buffered
185            .extend_bounded(&chunk, limit)
186            .map_err(|_| TransportError::ResponseTooLarge)?;
187    }
188    Ok(buffered)
189}
190
191const fn map_method(method: Method) -> reqwest::Method {
192    match method {
193        Method::Get => reqwest::Method::GET,
194        Method::Post => reqwest::Method::POST,
195        Method::Put => reqwest::Method::PUT,
196        Method::Delete => reqwest::Method::DELETE,
197    }
198}
199
200fn classify_reqwest_error(error: reqwest::Error) -> TransportError {
201    if error.is_timeout() {
202        TransportError::TimedOut
203    } else if error.is_connect() {
204        TransportError::ConnectFailed
205    } else {
206        TransportError::RequestFailed
207    }
208}
cloud_sdk_reqwest/asynchronous/client.rs

cloud_sdk_reqwest/asynchronous/
client.rs
