cloud_sdk_reqwest/blocking/
client.rs

1use core::fmt;
2use std::io::Read;
3
4use cloud_sdk::Method;
5use cloud_sdk::transport::{BlockingTransport, StatusCode, TransportRequest, TransportResponse};
6use cloud_sdk_sanitization::sanitize_bytes;
7use reqwest::blocking::{Body, Client};
8use reqwest::header::{ACCEPT, AUTHORIZATION, CONTENT_TYPE, HeaderValue};
9
10use super::body::{ReadBodyError, SanitizedRequestBody, read_bounded};
11use crate::shared::{BearerToken, HttpsEndpoint, TransportError, parse_rate_limit};
12
13/// Hardened provider-neutral reqwest blocking transport.
14pub struct BlockingClient {
15    client: Client,
16    endpoint: HttpsEndpoint,
17    token: BearerToken,
18}
19
20impl BlockingClient {
21    pub(super) const fn new(client: Client, endpoint: HttpsEndpoint, token: BearerToken) -> Self {
22        Self {
23            client,
24            endpoint,
25            token,
26        }
27    }
28
29    fn send_inner<'buffer>(
30        &mut self,
31        request: TransportRequest<'_>,
32        response_body: &'buffer mut [u8],
33    ) -> Result<TransportResponse<'buffer>, TransportError> {
34        sanitize_bytes(response_body);
35        let url = self
36            .endpoint
37            .compose(request.target())
38            .map_err(|_| TransportError::TargetRejected)?;
39        let method = map_method(request.method());
40        let authorization = self
41            .token
42            .header_value()
43            .map_err(|_| TransportError::HeaderRejected)?;
44        let mut outbound = self
45            .client
46            .request(method, url)
47            .header(AUTHORIZATION, authorization)
48            .header(ACCEPT, HeaderValue::from_static("application/json"));
49
50        if let Some(content_type) = request.content_type() {
51            let value = HeaderValue::from_str(content_type.as_str())
52                .map_err(|_| TransportError::HeaderRejected)?;
53            outbound = outbound.header(CONTENT_TYPE, value);
54        } else if !request.body().is_empty() {
55            return Err(TransportError::MissingContentType);
56        }
57
58        if !request.body().is_empty() {
59            let body = SanitizedRequestBody::new(request.body())
60                .map_err(|_| TransportError::RequestBodyAllocationFailed)?;
61            let body_len = u64::try_from(request.body().len())
62                .map_err(|_| TransportError::RequestBodyTooLarge)?;
63            outbound = outbound.body(Body::sized(body, body_len));
64        }
65
66        let mut response = outbound.send().map_err(classify_reqwest_error)?;
67        self.endpoint
68            .verify_origin(response.url())
69            .map_err(|_| TransportError::ResponseOriginChanged)?;
70        if response.content_length().is_some_and(|length| {
71            u64::try_from(response_body.len()).map_or(true, |cap| length > cap)
72        }) {
73            return Err(TransportError::ResponseTooLarge);
74        }
75        let status =
76            StatusCode::new(response.status().as_u16()).ok_or(TransportError::InvalidStatus)?;
77        let rate_limit = parse_rate_limit(response.headers())?;
78        let body_len = read_response(&mut response, response_body)?;
79        let initialized = response_body
80            .get(..body_len)
81            .ok_or(TransportError::ResponseReadFailed)?;
82        let response = TransportResponse::new(status, initialized);
83        Ok(rate_limit.map_or(response, |value| response.with_rate_limit(value)))
84    }
85}
86
87impl BlockingTransport for BlockingClient {
88    type Error = TransportError;
89
90    fn send<'buffer>(
91        &mut self,
92        request: TransportRequest<'_>,
93        response_body: &'buffer mut [u8],
94    ) -> Result<TransportResponse<'buffer>, Self::Error> {
95        self.send_inner(request, response_body)
96    }
97}
98
99impl fmt::Debug for BlockingClient {
100    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
101        formatter
102            .debug_struct("BlockingClient")
103            .field("endpoint", &"[redacted]")
104            .field("token", &"[redacted]")
105            .finish_non_exhaustive()
106    }
107}
108
109fn read_response(response: &mut impl Read, output: &mut [u8]) -> Result<usize, TransportError> {
110    match read_bounded(response, output) {
111        Ok(len) => Ok(len),
112        Err(error) => {
113            sanitize_bytes(output);
114            Err(match error {
115                ReadBodyError::TooLarge => TransportError::ResponseTooLarge,
116                ReadBodyError::ReadFailed => TransportError::ResponseReadFailed,
117            })
118        }
119    }
120}
121
122const fn map_method(method: Method) -> reqwest::Method {
123    match method {
124        Method::Get => reqwest::Method::GET,
125        Method::Post => reqwest::Method::POST,
126        Method::Put => reqwest::Method::PUT,
127        Method::Delete => reqwest::Method::DELETE,
128    }
129}
130
131fn classify_reqwest_error(error: reqwest::Error) -> TransportError {
132    if error.is_timeout() {
133        TransportError::TimedOut
134    } else if error.is_connect() {
135        TransportError::ConnectFailed
136    } else {
137        TransportError::RequestFailed
138    }
139}
cloud_sdk_reqwest/blocking/client.rs

cloud_sdk_reqwest/blocking/
client.rs
