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, StatusCode,
8    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,
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 body_len = read_response(&mut response, response_body)?;
119        let initialized = response_body
120            .get(..body_len)
121            .ok_or(TransportError::ResponseReadFailed)?;
122        let response = TransportResponse::new(status, initialized);
123        drop(token_snapshot);
124        Ok(rate_limit.map_or(response, |value| response.with_rate_limit(value)))
125    }
126}
127
128impl BlockingTransport for BlockingClient {
129    type Error = TransportError;
130
131    fn send<'buffer>(
132        &self,
133        request: TransportRequest<'_>,
134        response_body: &'buffer mut [u8],
135    ) -> Result<TransportResponse<'buffer>, Self::Error> {
136        self.send_inner(request, response_body)
137    }
138}
139
140impl BoundTransport for BlockingClient {
141    fn endpoint_identity(&self) -> Result<EndpointIdentity<'_>, EndpointIdentityError> {
142        self.endpoint.identity()
143    }
144}
145
146impl fmt::Debug for BlockingClient {
147    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
148        formatter
149            .debug_struct("BlockingClient")
150            .field("endpoint", &"[redacted]")
151            .field("credentials", &"[redacted]")
152            .finish_non_exhaustive()
153    }
154}
155
156fn read_response(response: &mut impl Read, output: &mut [u8]) -> Result<usize, TransportError> {
157    match read_bounded(response, output) {
158        Ok(len) => Ok(len),
159        Err(error) => {
160            sanitize_bytes(output);
161            Err(match error {
162                ReadBodyError::TooLarge => TransportError::ResponseTooLarge,
163                ReadBodyError::ReadFailed => TransportError::ResponseReadFailed,
164            })
165        }
166    }
167}
168
169const fn map_method(method: Method) -> reqwest::Method {
170    match method {
171        Method::Get => reqwest::Method::GET,
172        Method::Post => reqwest::Method::POST,
173        Method::Put => reqwest::Method::PUT,
174        Method::Delete => reqwest::Method::DELETE,
175    }
176}
177
178fn classify_reqwest_error(error: reqwest::Error) -> TransportError {
179    if error.is_timeout() {
180        TransportError::TimedOut
181    } else if error.is_connect() {
182        TransportError::ConnectFailed
183    } else {
184        TransportError::RequestFailed
185    }
186}
cloud_sdk_reqwest/blocking/client.rs

cloud_sdk_reqwest/blocking/
client.rs
