cloud_sdk_reqwest/shared/
raw.rs

1use cloud_sdk::Method;
2use cloud_sdk::transport::{
3    ContentType, HeaderSensitivity, RawResponsePolicy, ResponseHeaders, ResponseMediaPolicy,
4    StatusCode, TrailerPolicy, TransportFailure,
5};
6use core::ops::Range;
7use reqwest::header::{CONTENT_LENGTH, CONTENT_TYPE, HeaderMap, TRAILER};
8
9/// Maximum response-header fields parsed by pinned Hyper HTTP/1.
10pub const MAX_UPSTREAM_HTTP1_HEADERS: usize = 100;
11/// Maximum pinned Hyper HTTP/1 read buffer before a head-too-large failure.
12pub const MAX_UPSTREAM_HTTP1_HEAD_BYTES: usize = 64 * 1024;
13/// Maximum request body copied into raw adapter-owned staging.
14pub const MAX_RAW_REQUEST_BODY_BYTES: usize = cloud_sdk::operation::LARGE_BODY_BYTES;
15
16/// Payload-free raw HTTP execution failure.
17#[derive(Clone, Copy, Debug, Eq, PartialEq)]
18pub enum RawHttpError {
19    /// The supplied response writer was already committed.
20    ResponseAlreadyCommitted,
21    /// Endpoint and request-target composition failed.
22    TargetRejected,
23    /// The validated method could not be represented by reqwest.
24    MethodRejected,
25    /// A non-empty request body omitted `Content-Type`.
26    MissingContentType,
27    /// A request header could not be represented by reqwest.
28    HeaderRejected,
29    /// Adapter-owned request-header staging allocation failed.
30    RequestHeaderAllocationFailed,
31    /// Adapter-owned request-body staging allocation failed.
32    RequestBodyAllocationFailed,
33    /// Request-body length cannot be represented by the HTTP implementation.
34    RequestBodyTooLarge,
35    /// Construction of the exact raw request failed.
36    RequestBuildFailed,
37    /// The blocking adapter could not construct its private executor.
38    RuntimeInitializationFailed,
39    /// The blocking adapter was called from an active Tokio runtime.
40    BlockingRuntimeContext,
41    /// Connection establishment failed.
42    ConnectFailed,
43    /// A configured request or read deadline expired.
44    TimedOut,
45    /// Sending failed without a more precise payload-free classification.
46    RequestFailed,
47    /// The final response origin differed from the configured endpoint.
48    ResponseOriginChanged,
49    /// The final status was outside the core HTTP status domain.
50    InvalidStatus,
51    /// `101 Switching Protocols` is forbidden.
52    SwitchingProtocols,
53    /// More informational responses were observed than admitted.
54    TooManyInformationalResponses,
55    /// The response head exceeded pinned upstream count or byte bounds.
56    ResponseHeadTooLarge,
57    /// A response header name occurred more than once.
58    DuplicateResponseHeader,
59    /// A response declared trailers while the raw policy rejects them.
60    ResponseTrailersRejected,
61    /// A no-body response used forbidden framing.
62    InvalidNoBodyFraming,
63    /// A response content type was required but absent.
64    MissingResponseContentType,
65    /// A response content type was malformed.
66    InvalidResponseContentType,
67    /// A response content type did not match the selected status policy.
68    UnexpectedResponseContentType,
69    /// A response content type was present when forbidden.
70    ForbiddenResponseContentType,
71    /// A retained response header violated core bounds.
72    InvalidResponseHeader,
73    /// The declared or observed response body exceeded its status-class limit.
74    ResponseTooLarge,
75    /// A response exceeded the fixed chunk-observation budget.
76    ResponseChunkLimitExceeded,
77    /// Reading the response body failed.
78    ResponseReadFailed,
79    /// The core response writer rejected final commitment.
80    ResponseCommitFailed,
81}
82
83impl_static_error!(RawHttpError,
84    Self::ResponseAlreadyCommitted => "response writer is already committed",
85    Self::TargetRejected => "request target was rejected",
86    Self::MethodRejected => "request method was rejected",
87    Self::MissingContentType => "request body content type is missing",
88    Self::HeaderRejected => "request header was rejected",
89    Self::RequestHeaderAllocationFailed => "request-header allocation failed",
90    Self::RequestBodyAllocationFailed => "request-body allocation failed",
91    Self::RequestBodyTooLarge => "request body is too large",
92    Self::RequestBuildFailed => "raw request construction failed",
93    Self::RuntimeInitializationFailed => "blocking executor initialization failed",
94    Self::BlockingRuntimeContext => "blocking executor called from an async runtime",
95    Self::ConnectFailed => "connection failed",
96    Self::TimedOut => "request timed out",
97    Self::RequestFailed => "request failed",
98    Self::ResponseOriginChanged => "response origin changed",
99    Self::InvalidStatus => "response status is invalid",
100    Self::SwitchingProtocols => "switching protocols is forbidden",
101    Self::TooManyInformationalResponses => "too many informational responses",
102    Self::ResponseHeadTooLarge => "response head exceeds wire limits",
103    Self::DuplicateResponseHeader => "response header is duplicated",
104    Self::ResponseTrailersRejected => "response trailers are rejected",
105    Self::InvalidNoBodyFraming => "no-body response framing is invalid",
106    Self::MissingResponseContentType => "response content type is missing",
107    Self::InvalidResponseContentType => "response content type is invalid",
108    Self::UnexpectedResponseContentType => "response content type is not admitted",
109    Self::ForbiddenResponseContentType => "response content type is forbidden",
110    Self::InvalidResponseHeader => "retained response header is invalid",
111    Self::ResponseTooLarge => "response body exceeds its status-class limit",
112    Self::ResponseChunkLimitExceeded => "response chunk limit is exceeded",
113    Self::ResponseReadFailed => "response body read failed",
114    Self::ResponseCommitFailed => "response commitment failed",
115);
116
117/// Delivery-phased raw reqwest failure.
118pub type RawTransportFailure = TransportFailure<RawHttpError>;
119
120/// Delivery-phased authenticated reqwest failure.
121pub type AuthenticatedTransportFailure = TransportFailure<super::TransportError>;
122
123pub(crate) struct ResponseBodyBudget {
124    limit: usize,
125    len: usize,
126    chunks: usize,
127}
128
129impl ResponseBodyBudget {
130    pub(crate) const fn new(limit: usize) -> Self {
131        Self {
132            limit,
133            len: 0,
134            chunks: 0,
135        }
136    }
137
138    pub(crate) fn observe(&mut self, bytes: usize) -> Result<Range<usize>, RawHttpError> {
139        self.chunks = self
140            .chunks
141            .checked_add(1)
142            .ok_or(RawHttpError::ResponseChunkLimitExceeded)?;
143        if self.chunks > cloud_sdk::transport::MAX_RESPONSE_CHUNKS {
144            return Err(RawHttpError::ResponseChunkLimitExceeded);
145        }
146        let end = self
147            .len
148            .checked_add(bytes)
149            .ok_or(RawHttpError::ResponseTooLarge)?;
150        if end > self.limit {
151            return Err(RawHttpError::ResponseTooLarge);
152        }
153        let range = self.len..end;
154        self.len = end;
155        Ok(range)
156    }
157
158    pub(crate) const fn len(&self) -> usize {
159        self.len
160    }
161}
162
163pub(crate) fn inspect_response_head(
164    method: Method,
165    status: StatusCode,
166    source: &HeaderMap,
167    policy: RawResponsePolicy<'_>,
168    captured: &mut ResponseHeaders<'_>,
169    writer_capacity: usize,
170) -> Result<usize, RawHttpError> {
171    validate_wire_head(source)?;
172    if status.get() == 101 {
173        return Err(RawHttpError::SwitchingProtocols);
174    }
175    if status.get() < 200 {
176        return Err(RawHttpError::InvalidStatus);
177    }
178    if matches!(policy.trailer_policy(), TrailerPolicy::Reject) && source.contains_key(TRAILER) {
179        return Err(RawHttpError::ResponseTrailersRejected);
180    }
181
182    let policy_limit = policy.body_limit(status);
183    let body_forbidden = method == Method::Head || matches!(status.get(), 204 | 304);
184    if status.get() == 204 && source.contains_key(CONTENT_LENGTH) {
185        return Err(RawHttpError::InvalidNoBodyFraming);
186    }
187    validate_media(source, policy.media_policy(status))?;
188    let selected_limit = if body_forbidden {
189        0
190    } else {
191        core::cmp::min(policy_limit, writer_capacity)
192    };
193    if let Some(declared) = declared_content_length(source)? {
194        let declared = usize::try_from(declared).map_err(|_| RawHttpError::ResponseTooLarge)?;
195        if !body_forbidden && declared > selected_limit {
196            return Err(RawHttpError::ResponseTooLarge);
197        }
198    }
199
200    for name in source.keys() {
201        if !policy.admits_header(name.as_str()) {
202            continue;
203        }
204        let Some(value) = source.get(name) else {
205            return Err(RawHttpError::InvalidResponseHeader);
206        };
207        let sensitivity = if is_reviewed_public(name.as_str()) {
208            HeaderSensitivity::Public
209        } else {
210            HeaderSensitivity::Sensitive
211        };
212        captured
213            .try_push(name.as_str(), value.as_bytes(), sensitivity)
214            .map_err(|_| RawHttpError::InvalidResponseHeader)?;
215    }
216    Ok(selected_limit)
217}
218
219fn validate_wire_head(headers: &HeaderMap) -> Result<(), RawHttpError> {
220    if headers.len() > MAX_UPSTREAM_HTTP1_HEADERS {
221        return Err(RawHttpError::ResponseHeadTooLarge);
222    }
223    let mut encoded_len = 0_usize;
224    for name in headers.keys() {
225        let values = headers.get_all(name);
226        if values.iter().count() != 1 {
227            return Err(RawHttpError::DuplicateResponseHeader);
228        }
229        let Some(value) = values.iter().next() else {
230            return Err(RawHttpError::InvalidResponseHeader);
231        };
232        encoded_len = encoded_len
233            .checked_add(name.as_str().len())
234            .and_then(|length| length.checked_add(value.as_bytes().len()))
235            .and_then(|length| length.checked_add(4))
236            .ok_or(RawHttpError::ResponseHeadTooLarge)?;
237        if encoded_len > MAX_UPSTREAM_HTTP1_HEAD_BYTES {
238            return Err(RawHttpError::ResponseHeadTooLarge);
239        }
240    }
241    Ok(())
242}
243
244fn declared_content_length(headers: &HeaderMap) -> Result<Option<u64>, RawHttpError> {
245    let Some(value) = headers.get(CONTENT_LENGTH) else {
246        return Ok(None);
247    };
248    let text = value
249        .to_str()
250        .map_err(|_| RawHttpError::InvalidNoBodyFraming)?;
251    text.parse::<u64>()
252        .map(Some)
253        .map_err(|_| RawHttpError::InvalidNoBodyFraming)
254}
255
256fn validate_media(
257    headers: &HeaderMap,
258    policy: ResponseMediaPolicy<'_>,
259) -> Result<(), RawHttpError> {
260    let content_type = headers.get(CONTENT_TYPE);
261    match (policy, content_type) {
262        (ResponseMediaPolicy::Required(_), None) => Err(RawHttpError::MissingResponseContentType),
263        (ResponseMediaPolicy::Optional(_), None) | (ResponseMediaPolicy::Forbidden, None) => Ok(()),
264        (ResponseMediaPolicy::Forbidden, Some(_)) => {
265            Err(RawHttpError::ForbiddenResponseContentType)
266        }
267        (
268            ResponseMediaPolicy::Required(admitted) | ResponseMediaPolicy::Optional(admitted),
269            Some(value),
270        ) => {
271            let text = value
272                .to_str()
273                .map_err(|_| RawHttpError::InvalidResponseContentType)?;
274            let parsed =
275                ContentType::new(text).map_err(|_| RawHttpError::InvalidResponseContentType)?;
276            if admitted.iter().any(|media| parsed.matches(*media)) {
277                Ok(())
278            } else {
279                Err(RawHttpError::UnexpectedResponseContentType)
280            }
281        }
282    }
283}
284
285fn is_reviewed_public(name: &str) -> bool {
286    ["content-length", "content-type", "date"]
287        .iter()
288        .any(|candidate| name.eq_ignore_ascii_case(candidate))
289}
290
291#[cfg(test)]
292mod tests {
293    use std::format;
294
295    use cloud_sdk::transport::{
296        HeaderName, MediaType, RawResponsePolicy, ResponseHeaders, ResponseMediaPolicy, StatusCode,
297    };
298    use reqwest::header::{HeaderMap, HeaderValue};
299
300    use super::{RawHttpError, inspect_response_head};
301
302    fn policy<'a>(headers: &[HeaderName<'a>]) -> Option<RawResponsePolicy<'a>> {
303        RawResponsePolicy::new(
304            8,
305            4,
306            ResponseMediaPolicy::Required(&[MediaType::JSON]),
307            ResponseMediaPolicy::Optional(&[MediaType::JSON]),
308            headers,
309            2,
310        )
311        .ok()
312    }
313
314    #[test]
315    fn selects_status_limit_and_drops_unadmitted_headers() {
316        let admitted = HeaderName::new("content-type");
317        assert!(admitted.is_ok());
318        let Ok(admitted) = admitted else { return };
319        let admitted_headers = [admitted];
320        let Some(policy) = policy(&admitted_headers) else {
321            return;
322        };
323        let mut source = HeaderMap::new();
324        source.insert("content-type", HeaderValue::from_static("application/json"));
325        source.insert("set-cookie", HeaderValue::from_static("secret=1"));
326        source.insert("x-unknown", HeaderValue::from_static("secret"));
327        let mut storage = [0_u8; 128];
328        let mut captured = ResponseHeaders::new(&mut storage);
329        let result = inspect_response_head(
330            cloud_sdk::Method::Get,
331            StatusCode::OK,
332            &source,
333            policy,
334            &mut captured,
335            16,
336        );
337        assert_eq!(result, Ok(8));
338        assert!(captured.get("content-type").is_some());
339        assert!(captured.get("set-cookie").is_none());
340        assert!(captured.get("x-unknown").is_none());
341    }
342    #[test]
343    fn retains_incomplete_admitted_quota_metadata_for_provider_validation() {
344        let names = [
345            HeaderName::new("ratelimit-limit"),
346            HeaderName::new("ratelimit-remaining"),
347            HeaderName::new("ratelimit-reset"),
348        ];
349        let [Ok(limit), Ok(remaining), Ok(reset)] = names else {
350            return;
351        };
352        let Some(policy) = policy(&[limit, remaining, reset]) else {
353            return;
354        };
355        let mut source = HeaderMap::new();
356        source.insert("ratelimit-remaining", HeaderValue::from_static("7"));
357        let mut storage = [0_u8; 128];
358        let mut captured = ResponseHeaders::new(&mut storage);
359        let result = inspect_response_head(
360            cloud_sdk::Method::Get,
361            StatusCode::TOO_MANY_REQUESTS,
362            &source,
363            policy,
364            &mut captured,
365            16,
366        );
367        assert_eq!(result, Ok(4));
368        assert!(captured.get("ratelimit-limit").is_none());
369        assert_eq!(
370            captured
371                .get("ratelimit-remaining")
372                .map(|header| header.value()),
373            Some(b"7".as_slice())
374        );
375        assert!(captured.get("ratelimit-reset").is_none());
376    }
377    #[test]
378    fn rejects_duplicates_and_no_content_framing() {
379        let Some(policy) = policy(&[]) else { return };
380        let mut duplicate = HeaderMap::new();
381        duplicate.append("x-test", HeaderValue::from_static("one"));
382        duplicate.append("x-test", HeaderValue::from_static("two"));
383        let mut storage = [0_u8; 128];
384        let mut captured = ResponseHeaders::new(&mut storage);
385        assert_eq!(
386            inspect_response_head(
387                cloud_sdk::Method::Get,
388                StatusCode::new(400).unwrap_or(StatusCode::TOO_MANY_REQUESTS),
389                &duplicate,
390                policy,
391                &mut captured,
392                16,
393            ),
394            Err(RawHttpError::DuplicateResponseHeader)
395        );
396
397        let mut no_content = HeaderMap::new();
398        no_content.insert("content-length", HeaderValue::from_static("0"));
399        assert_eq!(
400            inspect_response_head(
401                cloud_sdk::Method::Get,
402                StatusCode::NO_CONTENT,
403                &no_content,
404                policy,
405                &mut captured,
406                16,
407            ),
408            Err(RawHttpError::InvalidNoBodyFraming)
409        );
410    }
411
412    #[test]
413    fn rejects_media_mismatch_oversized_length_and_hostile_header_count() {
414        let Some(policy) = policy(&[]) else { return };
415        let mut storage = [0_u8; 128];
416        let mut captured = ResponseHeaders::new(&mut storage);
417
418        let mut wrong_media = HeaderMap::new();
419        wrong_media.insert("content-type", HeaderValue::from_static("text/plain"));
420        assert_eq!(
421            inspect_response_head(
422                cloud_sdk::Method::Get,
423                StatusCode::OK,
424                &wrong_media,
425                policy,
426                &mut captured,
427                16,
428            ),
429            Err(RawHttpError::UnexpectedResponseContentType)
430        );
431
432        let mut oversized = HeaderMap::new();
433        oversized.insert("content-type", HeaderValue::from_static("application/json"));
434        oversized.insert("content-length", HeaderValue::from_static("9"));
435        assert_eq!(
436            inspect_response_head(
437                cloud_sdk::Method::Get,
438                StatusCode::OK,
439                &oversized,
440                policy,
441                &mut captured,
442                16,
443            ),
444            Err(RawHttpError::ResponseTooLarge)
445        );
446
447        let mut hostile = HeaderMap::new();
448        for index in 0..=super::MAX_UPSTREAM_HTTP1_HEADERS {
449            let name = format!("x-field-{index}");
450            let Ok(name) = reqwest::header::HeaderName::from_bytes(name.as_bytes()) else {
451                return;
452            };
453            hostile.insert(name, HeaderValue::from_static("value"));
454        }
455        assert_eq!(
456            inspect_response_head(
457                cloud_sdk::Method::Get,
458                StatusCode::OK,
459                &hostile,
460                policy,
461                &mut captured,
462                16,
463            ),
464            Err(RawHttpError::ResponseHeadTooLarge)
465        );
466    }
467
468    #[test]
469    fn head_and_not_modified_select_zero_body_capacity() {
470        let Some(policy) = policy(&[]) else { return };
471        let mut source = HeaderMap::new();
472        source.insert("content-type", HeaderValue::from_static("application/json"));
473        source.insert("content-length", HeaderValue::from_static("8"));
474        let mut storage = [0_u8; 128];
475        let mut captured = ResponseHeaders::new(&mut storage);
476        assert_eq!(
477            inspect_response_head(
478                cloud_sdk::Method::Head,
479                StatusCode::OK,
480                &source,
481                policy,
482                &mut captured,
483                16,
484            ),
485            Ok(0)
486        );
487        let not_modified = StatusCode::new(304).unwrap_or(StatusCode::NO_CONTENT);
488        assert_eq!(
489            inspect_response_head(
490                cloud_sdk::Method::Get,
491                not_modified,
492                &source,
493                policy,
494                &mut captured,
495                16,
496            ),
497            Ok(0)
498        );
499    }
500}
cloud_sdk_reqwest/shared/raw.rs

cloud_sdk_reqwest/shared/
raw.rs
