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::authentication::{
7    AuthenticatedRequest, BlockingAuthenticatedTransport, CredentialGeneration,
8};
9use cloud_sdk::transport::{
10    BoundTransport, EndpointIdentity, EndpointIdentityError, ResponseAttempt, ResponseMetadata,
11    ResponseStorageSanitizer, ResponseWriter, StatusCode,
12};
13use cloud_sdk_sanitization::{SecretBuffer, sanitize_bytes};
14use reqwest::blocking::{Body, Client};
15use reqwest::header::{AUTHORIZATION, HeaderName, HeaderValue};
16
17use super::body::{ReadBodyError, SanitizedRequestBody, read_bounded};
18use crate::shared::{
19    BearerCredential, BearerCredentialScope, BearerCredentialSnapshot, BearerRefreshHandoff,
20    BearerToken, CredentialStateError, CredentialStore, CredentialUpdateError, HttpsEndpoint,
21    TokenRefreshError, TokenRotationError, TransportError, capture_response_headers,
22    map_authentication_error, parse_rate_limit, parse_response_content_type,
23    validate_bearer_authentication,
24};
25
26/// Hardened provider-neutral reqwest blocking bearer transport.
27#[derive(Clone)]
28pub struct BlockingClient {
29    client: Client,
30    endpoint: HttpsEndpoint,
31    scope: Arc<BearerCredentialScope>,
32    credentials: Arc<CredentialStore>,
33    allow_insecure_loopback: bool,
34}
35
36impl BlockingClient {
37    pub(super) fn new(
38        client: Client,
39        endpoint: HttpsEndpoint,
40        credential: BearerCredential,
41        allow_insecure_loopback: bool,
42    ) -> Self {
43        Self {
44            client,
45            endpoint,
46            scope: Arc::new(credential.scope),
47            credentials: Arc::new(CredentialStore::new(credential.token)),
48            allow_insecure_loopback,
49        }
50    }
51
52    /// Captures the current generation without exposing token bytes.
53    pub fn credential_snapshot(&self) -> Result<BearerCredentialSnapshot, CredentialStateError> {
54        self.credentials.snapshot()
55    }
56
57    /// Atomically replaces the bearer token used by newly started requests.
58    ///
59    /// In-flight requests retain their previous snapshot. The immutable scope
60    /// cannot change during rotation.
61    pub fn rotate_bearer_token(
62        &self,
63        replacement: BearerToken,
64    ) -> Result<CredentialGeneration, CredentialUpdateError> {
65        self.credentials.rotate(replacement)
66    }
67
68    /// Validates and rotates mutable bytes, clearing the complete source.
69    pub fn rotate_bearer_token_from_mut_bytes(
70        &self,
71        source: &mut [u8],
72    ) -> Result<CredentialGeneration, TokenRotationError> {
73        self.credentials.rotate_from_mut_bytes(source)
74    }
75
76    /// Validates and rotates guarded storage, which clears on return.
77    pub fn rotate_bearer_token_from_secret_buffer(
78        &self,
79        source: SecretBuffer<'_>,
80    ) -> Result<CredentialGeneration, TokenRotationError> {
81        self.credentials.rotate_from_secret_buffer(source)
82    }
83
84    /// Installs a refresh only if its captured generation is still current.
85    pub fn refresh_bearer_token(
86        &self,
87        handoff: BearerRefreshHandoff,
88        replacement: BearerToken,
89    ) -> Result<CredentialGeneration, TokenRefreshError> {
90        self.credentials.refresh(handoff, replacement)
91    }
92
93    /// Validates refreshed mutable bytes, clears them, and rejects stale work.
94    pub fn refresh_bearer_token_from_mut_bytes(
95        &self,
96        handoff: BearerRefreshHandoff,
97        source: &mut [u8],
98    ) -> Result<CredentialGeneration, TokenRefreshError> {
99        self.credentials.refresh_from_mut_bytes(handoff, source)
100    }
101
102    /// Consumes guarded refreshed storage and rejects stale work.
103    pub fn refresh_bearer_token_from_secret_buffer(
104        &self,
105        handoff: BearerRefreshHandoff,
106        source: SecretBuffer<'_>,
107    ) -> Result<CredentialGeneration, TokenRefreshError> {
108        self.credentials.refresh_from_secret_buffer(handoff, source)
109    }
110
111    fn send_inner(
112        &self,
113        authenticated: AuthenticatedRequest<'_, '_>,
114        response_writer: &mut ResponseWriter<'_>,
115    ) -> Result<(), TransportError> {
116        let mut response_attempt = response_writer
117            .begin_attempt()
118            .map_err(|_| TransportError::ResponseCommitFailed)?;
119        let endpoint_identity = self
120            .endpoint
121            .identity()
122            .map_err(|_| TransportError::AuthenticationEndpointMismatch)?;
123        validate_bearer_authentication(
124            endpoint_identity,
125            &self.scope,
126            authenticated.policy(),
127            self.allow_insecure_loopback,
128        )
129        .map_err(map_authentication_error)?;
130        let token_snapshot = self
131            .credentials
132            .snapshot()
133            .map_err(|_| TransportError::CredentialStateUnavailable)?;
134        let authorization = token_snapshot
135            .header_value()
136            .map_err(|_| TransportError::HeaderRejected)?;
137        drop(token_snapshot);
138        execute(
139            &self.client,
140            &self.endpoint,
141            authorization,
142            authenticated,
143            &mut response_attempt,
144        )
145    }
146}
147
148pub(super) fn execute(
149    client: &Client,
150    endpoint: &HttpsEndpoint,
151    authorization: HeaderValue,
152    authenticated: AuthenticatedRequest<'_, '_>,
153    response_writer: &mut ResponseAttempt<'_, '_>,
154) -> Result<(), TransportError> {
155    let request = authenticated.transport_request();
156    let url = endpoint
157        .compose(request.target())
158        .map_err(|_| TransportError::TargetRejected)?;
159    let method = map_method(request.method())?;
160    let mut outbound = client
161        .request(method, url)
162        .header(AUTHORIZATION, authorization);
163
164    for header in request.headers().as_slice() {
165        let name = HeaderName::from_bytes(header.name().as_str().as_bytes())
166            .map_err(|_| TransportError::HeaderRejected)?;
167        let mut value = HeaderValue::from_str(header.value().as_str())
168            .map_err(|_| TransportError::HeaderRejected)?;
169        value.set_sensitive(matches!(
170            header.sensitivity(),
171            cloud_sdk::transport::HeaderSensitivity::Sensitive
172        ));
173        outbound = outbound.header(name, value);
174    }
175    if !request.body().is_empty() && request.headers().get("content-type").is_none() {
176        return Err(TransportError::MissingContentType);
177    }
178    if !request.body().is_empty() {
179        let body = SanitizedRequestBody::new(request.body())
180            .map_err(|_| TransportError::RequestBodyAllocationFailed)?;
181        let body_len =
182            u64::try_from(request.body().len()).map_err(|_| TransportError::RequestBodyTooLarge)?;
183        outbound = outbound.body(Body::sized(body, body_len));
184    }
185
186    let mut response = outbound.send().map_err(classify_reqwest_error)?;
187    endpoint
188        .verify_origin(response.url())
189        .map_err(|_| TransportError::ResponseOriginChanged)?;
190    capture_response_headers(
191        response.headers(),
192        response_writer
193            .headers_mut()
194            .map_err(|_| TransportError::ResponseCommitFailed)?,
195    )?;
196    if response.content_length().is_some_and(|length| {
197        u64::try_from(response_writer.body_capacity()).map_or(true, |cap| length > cap)
198    }) {
199        return Err(TransportError::ResponseTooLarge);
200    }
201    let status =
202        StatusCode::new(response.status().as_u16()).ok_or(TransportError::InvalidStatus)?;
203    let rate_limit = parse_rate_limit(response_writer.headers())?;
204    parse_response_content_type(response_writer.headers())?;
205    let body_len = read_response(
206        &mut response,
207        response_writer
208            .body_mut()
209            .map_err(|_| TransportError::ResponseCommitFailed)?,
210    )?;
211    let mut metadata = ResponseMetadata::EMPTY;
212    if let Some(value) = rate_limit {
213        metadata = metadata.with_rate_limit(value);
214    }
215    response_writer
216        .commit(status, body_len, metadata)
217        .map_err(|_| TransportError::ResponseCommitFailed)
218}
219
220impl BlockingAuthenticatedTransport for BlockingClient {
221    type Error = TransportError;
222
223    fn send_authenticated(
224        &self,
225        request: AuthenticatedRequest<'_, '_>,
226        response: &mut ResponseWriter<'_>,
227    ) -> Result<(), Self::Error> {
228        self.send_inner(request, response)
229    }
230}
231
232impl ResponseStorageSanitizer for BlockingClient {
233    fn sanitize_response_storage(&self, response_storage: &mut [u8]) {
234        sanitize_bytes(response_storage);
235    }
236}
237
238impl BoundTransport for BlockingClient {
239    fn endpoint_identity(&self) -> Result<EndpointIdentity<'_>, EndpointIdentityError> {
240        self.endpoint.identity()
241    }
242}
243
244impl fmt::Debug for BlockingClient {
245    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
246        formatter
247            .debug_struct("BlockingClient")
248            .field("endpoint", &"[redacted]")
249            .field("scope", &"[redacted]")
250            .field("credentials", &"[redacted]")
251            .finish_non_exhaustive()
252    }
253}
254
255fn read_response(response: &mut impl Read, output: &mut [u8]) -> Result<usize, TransportError> {
256    match read_bounded(response, output) {
257        Ok(len) => Ok(len),
258        Err(error) => {
259            sanitize_bytes(output);
260            Err(match error {
261                ReadBodyError::TooLarge => TransportError::ResponseTooLarge,
262                ReadBodyError::ReadFailed => TransportError::ResponseReadFailed,
263            })
264        }
265    }
266}
267
268fn map_method(method: Method) -> Result<reqwest::Method, TransportError> {
269    reqwest::Method::from_bytes(method.as_str().as_bytes())
270        .map_err(|_| TransportError::MethodRejected)
271}
272
273fn classify_reqwest_error(error: reqwest::Error) -> TransportError {
274    if error.is_timeout() {
275        TransportError::TimedOut
276    } else if error.is_connect() {
277        TransportError::ConnectFailed
278    } else {
279        TransportError::RequestFailed
280    }
281}
cloud_sdk_reqwest/blocking/client.rs

cloud_sdk_reqwest/blocking/
client.rs
