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
120pub(crate) struct ResponseBodyBudget {
121    limit: usize,
122    len: usize,
123    chunks: usize,
124}
125
126impl ResponseBodyBudget {
127    pub(crate) const fn new(limit: usize) -> Self {
128        Self {
129            limit,
130            len: 0,
131            chunks: 0,
132        }
133    }
134
135    pub(crate) fn observe(&mut self, bytes: usize) -> Result<Range<usize>, RawHttpError> {
136        self.chunks = self
137            .chunks
138            .checked_add(1)
139            .ok_or(RawHttpError::ResponseChunkLimitExceeded)?;
140        if self.chunks > cloud_sdk::transport::MAX_RESPONSE_CHUNKS {
141            return Err(RawHttpError::ResponseChunkLimitExceeded);
142        }
143        let end = self
144            .len
145            .checked_add(bytes)
146            .ok_or(RawHttpError::ResponseTooLarge)?;
147        if end > self.limit {
148            return Err(RawHttpError::ResponseTooLarge);
149        }
150        let range = self.len..end;
151        self.len = end;
152        Ok(range)
153    }
154
155    pub(crate) const fn len(&self) -> usize {
156        self.len
157    }
158}
159
160pub(crate) fn inspect_response_head(
161    method: Method,
162    status: StatusCode,
163    source: &HeaderMap,
164    policy: RawResponsePolicy<'_>,
165    captured: &mut ResponseHeaders<'_>,
166    writer_capacity: usize,
167) -> Result<usize, RawHttpError> {
168    validate_wire_head(source)?;
169    if status.get() == 101 {
170        return Err(RawHttpError::SwitchingProtocols);
171    }
172    if status.get() < 200 {
173        return Err(RawHttpError::InvalidStatus);
174    }
175    if matches!(policy.trailer_policy(), TrailerPolicy::Reject) && source.contains_key(TRAILER) {
176        return Err(RawHttpError::ResponseTrailersRejected);
177    }
178
179    let policy_limit = policy.body_limit(status);
180    let body_forbidden = method == Method::Head || matches!(status.get(), 204 | 304);
181    if status.get() == 204 && source.contains_key(CONTENT_LENGTH) {
182        return Err(RawHttpError::InvalidNoBodyFraming);
183    }
184    validate_media(source, policy.media_policy(status))?;
185    let selected_limit = if body_forbidden {
186        0
187    } else {
188        core::cmp::min(policy_limit, writer_capacity)
189    };
190    if let Some(declared) = declared_content_length(source)? {
191        let declared = usize::try_from(declared).map_err(|_| RawHttpError::ResponseTooLarge)?;
192        if !body_forbidden && declared > selected_limit {
193            return Err(RawHttpError::ResponseTooLarge);
194        }
195    }
196
197    for name in source.keys() {
198        if !policy.admits_header(name.as_str()) {
199            continue;
200        }
201        let Some(value) = source.get(name) else {
202            return Err(RawHttpError::InvalidResponseHeader);
203        };
204        let sensitivity = if is_reviewed_public(name.as_str()) {
205            HeaderSensitivity::Public
206        } else {
207            HeaderSensitivity::Sensitive
208        };
209        captured
210            .try_push(name.as_str(), value.as_bytes(), sensitivity)
211            .map_err(|_| RawHttpError::InvalidResponseHeader)?;
212    }
213    Ok(selected_limit)
214}
215
216fn validate_wire_head(headers: &HeaderMap) -> Result<(), RawHttpError> {
217    if headers.len() > MAX_UPSTREAM_HTTP1_HEADERS {
218        return Err(RawHttpError::ResponseHeadTooLarge);
219    }
220    let mut encoded_len = 0_usize;
221    for name in headers.keys() {
222        let values = headers.get_all(name);
223        if values.iter().count() != 1 {
224            return Err(RawHttpError::DuplicateResponseHeader);
225        }
226        let Some(value) = values.iter().next() else {
227            return Err(RawHttpError::InvalidResponseHeader);
228        };
229        encoded_len = encoded_len
230            .checked_add(name.as_str().len())
231            .and_then(|length| length.checked_add(value.as_bytes().len()))
232            .and_then(|length| length.checked_add(4))
233            .ok_or(RawHttpError::ResponseHeadTooLarge)?;
234        if encoded_len > MAX_UPSTREAM_HTTP1_HEAD_BYTES {
235            return Err(RawHttpError::ResponseHeadTooLarge);
236        }
237    }
238    Ok(())
239}
240
241fn declared_content_length(headers: &HeaderMap) -> Result<Option<u64>, RawHttpError> {
242    let Some(value) = headers.get(CONTENT_LENGTH) else {
243        return Ok(None);
244    };
245    let text = value
246        .to_str()
247        .map_err(|_| RawHttpError::InvalidNoBodyFraming)?;
248    text.parse::<u64>()
249        .map(Some)
250        .map_err(|_| RawHttpError::InvalidNoBodyFraming)
251}
252
253fn validate_media(
254    headers: &HeaderMap,
255    policy: ResponseMediaPolicy<'_>,
256) -> Result<(), RawHttpError> {
257    let content_type = headers.get(CONTENT_TYPE);
258    match (policy, content_type) {
259        (ResponseMediaPolicy::Required(_), None) => Err(RawHttpError::MissingResponseContentType),
260        (ResponseMediaPolicy::Optional(_), None) | (ResponseMediaPolicy::Forbidden, None) => Ok(()),
261        (ResponseMediaPolicy::Forbidden, Some(_)) => {
262            Err(RawHttpError::ForbiddenResponseContentType)
263        }
264        (
265            ResponseMediaPolicy::Required(admitted) | ResponseMediaPolicy::Optional(admitted),
266            Some(value),
267        ) => {
268            let text = value
269                .to_str()
270                .map_err(|_| RawHttpError::InvalidResponseContentType)?;
271            let parsed =
272                ContentType::new(text).map_err(|_| RawHttpError::InvalidResponseContentType)?;
273            if admitted.iter().any(|media| parsed.matches(*media)) {
274                Ok(())
275            } else {
276                Err(RawHttpError::UnexpectedResponseContentType)
277            }
278        }
279    }
280}
281
282fn is_reviewed_public(name: &str) -> bool {
283    ["content-length", "content-type", "date"]
284        .iter()
285        .any(|candidate| name.eq_ignore_ascii_case(candidate))
286}
287
288#[cfg(test)]
289mod tests {
290    use std::format;
291
292    use cloud_sdk::transport::{
293        HeaderName, MediaType, RawResponsePolicy, ResponseHeaders, ResponseMediaPolicy, StatusCode,
294    };
295    use reqwest::header::{HeaderMap, HeaderValue};
296
297    use super::{RawHttpError, inspect_response_head};
298
299    fn policy<'a>(headers: &'a [HeaderName<'a>]) -> Option<RawResponsePolicy<'a>> {
300        RawResponsePolicy::new(
301            8,
302            4,
303            ResponseMediaPolicy::Required(&[MediaType::JSON]),
304            ResponseMediaPolicy::Optional(&[MediaType::JSON]),
305            headers,
306            2,
307        )
308        .ok()
309    }
310
311    #[test]
312    fn selects_status_limit_and_drops_unadmitted_headers() {
313        let admitted = HeaderName::new("content-type");
314        assert!(admitted.is_ok());
315        let Ok(admitted) = admitted else { return };
316        let admitted_headers = [admitted];
317        let Some(policy) = policy(&admitted_headers) else {
318            return;
319        };
320        let mut source = HeaderMap::new();
321        source.insert("content-type", HeaderValue::from_static("application/json"));
322        source.insert("set-cookie", HeaderValue::from_static("secret=1"));
323        source.insert("x-unknown", HeaderValue::from_static("secret"));
324        let mut storage = [0_u8; 128];
325        let mut captured = ResponseHeaders::new(&mut storage);
326        let result = inspect_response_head(
327            cloud_sdk::Method::Get,
328            StatusCode::OK,
329            &source,
330            policy,
331            &mut captured,
332            16,
333        );
334        assert_eq!(result, Ok(8));
335        assert!(captured.get("content-type").is_some());
336        assert!(captured.get("set-cookie").is_none());
337        assert!(captured.get("x-unknown").is_none());
338    }
339
340    #[test]
341    fn rejects_duplicates_and_no_content_framing() {
342        let Some(policy) = policy(&[]) else { return };
343        let mut duplicate = HeaderMap::new();
344        duplicate.append("x-test", HeaderValue::from_static("one"));
345        duplicate.append("x-test", HeaderValue::from_static("two"));
346        let mut storage = [0_u8; 128];
347        let mut captured = ResponseHeaders::new(&mut storage);
348        assert_eq!(
349            inspect_response_head(
350                cloud_sdk::Method::Get,
351                StatusCode::new(400).unwrap_or(StatusCode::TOO_MANY_REQUESTS),
352                &duplicate,
353                policy,
354                &mut captured,
355                16,
356            ),
357            Err(RawHttpError::DuplicateResponseHeader)
358        );
359
360        let mut no_content = HeaderMap::new();
361        no_content.insert("content-length", HeaderValue::from_static("0"));
362        assert_eq!(
363            inspect_response_head(
364                cloud_sdk::Method::Get,
365                StatusCode::NO_CONTENT,
366                &no_content,
367                policy,
368                &mut captured,
369                16,
370            ),
371            Err(RawHttpError::InvalidNoBodyFraming)
372        );
373    }
374
375    #[test]
376    fn rejects_media_mismatch_oversized_length_and_hostile_header_count() {
377        let Some(policy) = policy(&[]) else { return };
378        let mut storage = [0_u8; 128];
379        let mut captured = ResponseHeaders::new(&mut storage);
380
381        let mut wrong_media = HeaderMap::new();
382        wrong_media.insert("content-type", HeaderValue::from_static("text/plain"));
383        assert_eq!(
384            inspect_response_head(
385                cloud_sdk::Method::Get,
386                StatusCode::OK,
387                &wrong_media,
388                policy,
389                &mut captured,
390                16,
391            ),
392            Err(RawHttpError::UnexpectedResponseContentType)
393        );
394
395        let mut oversized = HeaderMap::new();
396        oversized.insert("content-type", HeaderValue::from_static("application/json"));
397        oversized.insert("content-length", HeaderValue::from_static("9"));
398        assert_eq!(
399            inspect_response_head(
400                cloud_sdk::Method::Get,
401                StatusCode::OK,
402                &oversized,
403                policy,
404                &mut captured,
405                16,
406            ),
407            Err(RawHttpError::ResponseTooLarge)
408        );
409
410        let mut hostile = HeaderMap::new();
411        for index in 0..=super::MAX_UPSTREAM_HTTP1_HEADERS {
412            let name = format!("x-field-{index}");
413            let Ok(name) = reqwest::header::HeaderName::from_bytes(name.as_bytes()) else {
414                return;
415            };
416            hostile.insert(name, HeaderValue::from_static("value"));
417        }
418        assert_eq!(
419            inspect_response_head(
420                cloud_sdk::Method::Get,
421                StatusCode::OK,
422                &hostile,
423                policy,
424                &mut captured,
425                16,
426            ),
427            Err(RawHttpError::ResponseHeadTooLarge)
428        );
429    }
430
431    #[test]
432    fn head_and_not_modified_select_zero_body_capacity() {
433        let Some(policy) = policy(&[]) else { return };
434        let mut source = HeaderMap::new();
435        source.insert("content-type", HeaderValue::from_static("application/json"));
436        source.insert("content-length", HeaderValue::from_static("8"));
437        let mut storage = [0_u8; 128];
438        let mut captured = ResponseHeaders::new(&mut storage);
439        assert_eq!(
440            inspect_response_head(
441                cloud_sdk::Method::Head,
442                StatusCode::OK,
443                &source,
444                policy,
445                &mut captured,
446                16,
447            ),
448            Ok(0)
449        );
450        let not_modified = StatusCode::new(304).unwrap_or(StatusCode::NO_CONTENT);
451        assert_eq!(
452            inspect_response_head(
453                cloud_sdk::Method::Get,
454                not_modified,
455                &source,
456                policy,
457                &mut captured,
458                16,
459            ),
460            Ok(0)
461        );
462    }
463}
cloud_sdk_reqwest/shared/raw.rs

cloud_sdk_reqwest/shared/
raw.rs
