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, 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    AuthenticationValidationError, BearerCredential, BearerCredentialScope,
20    BearerCredentialSnapshot, BearerRefreshHandoff, BearerToken, CredentialStateError,
21    CredentialStore, CredentialUpdateError, HttpsEndpoint, TokenRefreshError, TokenRotationError,
22    TransportError, capture_response_headers, 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        if response_writer.is_committed() {
117            return Err(TransportError::ResponseCommitFailed);
118        }
119        let mut response_writer = response_writer
120            .begin_attempt()
121            .map_err(|_| TransportError::ResponseCommitFailed)?;
122        let endpoint_identity = self
123            .endpoint
124            .identity()
125            .map_err(|_| TransportError::AuthenticationEndpointMismatch)?;
126        validate_bearer_authentication(
127            endpoint_identity,
128            &self.scope,
129            authenticated.policy(),
130            self.allow_insecure_loopback,
131        )
132        .map_err(map_authentication_error)?;
133        let request = authenticated.transport_request();
134        let url = self
135            .endpoint
136            .compose(request.target())
137            .map_err(|_| TransportError::TargetRejected)?;
138        let method = map_method(request.method())?;
139        let token_snapshot = self
140            .credentials
141            .snapshot()
142            .map_err(|_| TransportError::CredentialStateUnavailable)?;
143        let authorization = token_snapshot
144            .header_value()
145            .map_err(|_| TransportError::HeaderRejected)?;
146        let mut outbound = self
147            .client
148            .request(method, url)
149            .header(AUTHORIZATION, authorization);
150
151        for header in request.headers().as_slice() {
152            let name = HeaderName::from_bytes(header.name().as_str().as_bytes())
153                .map_err(|_| TransportError::HeaderRejected)?;
154            let mut value = HeaderValue::from_str(header.value().as_str())
155                .map_err(|_| TransportError::HeaderRejected)?;
156            value.set_sensitive(matches!(
157                header.sensitivity(),
158                cloud_sdk::transport::HeaderSensitivity::Sensitive
159            ));
160            outbound = outbound.header(name, value);
161        }
162        if !request.body().is_empty() && request.headers().get("content-type").is_none() {
163            return Err(TransportError::MissingContentType);
164        }
165        if !request.body().is_empty() {
166            let body = SanitizedRequestBody::new(request.body())
167                .map_err(|_| TransportError::RequestBodyAllocationFailed)?;
168            let body_len = u64::try_from(request.body().len())
169                .map_err(|_| TransportError::RequestBodyTooLarge)?;
170            outbound = outbound.body(Body::sized(body, body_len));
171        }
172
173        let mut response = outbound.send().map_err(classify_reqwest_error)?;
174        self.endpoint
175            .verify_origin(response.url())
176            .map_err(|_| TransportError::ResponseOriginChanged)?;
177        capture_response_headers(
178            response.headers(),
179            response_writer
180                .headers_mut()
181                .map_err(|_| TransportError::ResponseCommitFailed)?,
182        )?;
183        if response.content_length().is_some_and(|length| {
184            u64::try_from(response_writer.body_capacity()).map_or(true, |cap| length > cap)
185        }) {
186            return Err(TransportError::ResponseTooLarge);
187        }
188        let status =
189            StatusCode::new(response.status().as_u16()).ok_or(TransportError::InvalidStatus)?;
190        let rate_limit = parse_rate_limit(response_writer.headers())?;
191        parse_response_content_type(response_writer.headers())?;
192        let body_len = read_response(
193            &mut response,
194            response_writer
195                .body_mut()
196                .map_err(|_| TransportError::ResponseCommitFailed)?,
197        )?;
198        let mut metadata = ResponseMetadata::EMPTY;
199        if let Some(value) = rate_limit {
200            metadata = metadata.with_rate_limit(value);
201        }
202        drop(token_snapshot);
203        response_writer
204            .commit(status, body_len, metadata)
205            .map_err(|_| TransportError::ResponseCommitFailed)
206    }
207}
208
209impl BlockingAuthenticatedTransport for BlockingClient {
210    type Error = TransportError;
211
212    fn send_authenticated(
213        &self,
214        request: AuthenticatedRequest<'_, '_>,
215        response: &mut ResponseWriter<'_>,
216    ) -> Result<(), Self::Error> {
217        self.send_inner(request, response)
218    }
219}
220
221impl ResponseStorageSanitizer for BlockingClient {
222    fn sanitize_response_storage(&self, response_storage: &mut [u8]) {
223        sanitize_bytes(response_storage);
224    }
225}
226
227impl BoundTransport for BlockingClient {
228    fn endpoint_identity(&self) -> Result<EndpointIdentity<'_>, EndpointIdentityError> {
229        self.endpoint.identity()
230    }
231}
232
233impl fmt::Debug for BlockingClient {
234    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
235        formatter
236            .debug_struct("BlockingClient")
237            .field("endpoint", &"[redacted]")
238            .field("scope", &"[redacted]")
239            .field("credentials", &"[redacted]")
240            .finish_non_exhaustive()
241    }
242}
243
244fn map_authentication_error(error: AuthenticationValidationError) -> TransportError {
245    match error {
246        AuthenticationValidationError::InsecureEndpoint => {
247            TransportError::InsecureAuthenticationEndpoint
248        }
249        AuthenticationValidationError::EndpointMismatch => {
250            TransportError::AuthenticationEndpointMismatch
251        }
252        AuthenticationValidationError::IncompletePolicy => {
253            TransportError::AuthenticationScopeRejected
254        }
255        AuthenticationValidationError::ScopeRejected => TransportError::AuthenticationScopeRejected,
256    }
257}
258
259fn read_response(response: &mut impl Read, output: &mut [u8]) -> Result<usize, TransportError> {
260    match read_bounded(response, output) {
261        Ok(len) => Ok(len),
262        Err(error) => {
263            sanitize_bytes(output);
264            Err(match error {
265                ReadBodyError::TooLarge => TransportError::ResponseTooLarge,
266                ReadBodyError::ReadFailed => TransportError::ResponseReadFailed,
267            })
268        }
269    }
270}
271
272fn map_method(method: Method) -> Result<reqwest::Method, TransportError> {
273    reqwest::Method::from_bytes(method.as_str().as_bytes())
274        .map_err(|_| TransportError::MethodRejected)
275}
276
277fn classify_reqwest_error(error: reqwest::Error) -> TransportError {
278    if error.is_timeout() {
279        TransportError::TimedOut
280    } else if error.is_connect() {
281        TransportError::ConnectFailed
282    } else {
283        TransportError::RequestFailed
284    }
285}
cloud_sdk_reqwest/blocking/client.rs

cloud_sdk_reqwest/blocking/
client.rs
