cloud_sdk_reqwest/blocking/
client.rs

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

cloud_sdk_reqwest/blocking/
client.rs
