cloud_sdk/transport/
raw.rs

1//! Provider-neutral raw HTTP execution and response-wire policy.
2
3mod local_async;
4
5pub use local_async::{LocalAsyncRawHttpExecutor, drive_local_raw};
6
7use core::future::Future;
8
9use super::{
10    AsyncExecutionError, AsyncResponseStaging, HeaderName, MediaType, ResponseCompletion,
11    ResponseWriter, StatusCode, TransportRequest,
12};
13
14/// Maximum informational response heads accepted before a final response.
15pub const MAX_INFORMATIONAL_RESPONSES: u8 = 8;
16/// Maximum chunks admitted by one buffered raw response.
17pub const MAX_RESPONSE_CHUNKS: usize = 4_096;
18/// Maximum per-class body limit represented by the raw buffered contract.
19pub const MAX_RAW_RESPONSE_BODY_BYTES: usize = 64 * 1024 * 1024;
20
21/// Response media-type requirement for one status class.
22#[derive(Clone, Copy, Debug, Eq, PartialEq)]
23pub enum ResponseMediaPolicy<'a> {
24    /// A content type is required and must match one admitted essence.
25    Required(&'a [MediaType<'a>]),
26    /// A content type may be absent, but a present value must match.
27    Optional(&'a [MediaType<'a>]),
28    /// A content type and response body are forbidden.
29    Forbidden,
30}
31
32/// Response trailer handling.
33#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
34pub enum TrailerPolicy {
35    /// Reject declared or observed response trailers.
36    Reject,
37}
38
39/// Invalid raw response-wire policy.
40#[derive(Clone, Copy, Debug, Eq, PartialEq)]
41pub enum RawResponsePolicyError {
42    /// An informational-response limit exceeded the global hard ceiling.
43    InformationalLimitTooLarge,
44    /// A body limit exceeded the buffered raw-response ceiling.
45    BodyLimitTooLarge,
46    /// Required or optional media policy omitted accepted media types.
47    MissingMediaType,
48    /// A media policy contains duplicate media-type essences.
49    DuplicateMediaType,
50    /// Too many response-header names were admitted.
51    TooManyAdmittedHeaders,
52    /// Admitted response-header names contain a duplicate.
53    DuplicateAdmittedHeader,
54    /// Credential, cookie, framing, proxy, or upgrade response metadata was admitted.
55    UnsafeAdmittedHeader,
56    /// A status class forbids media and body data but has a nonzero body limit.
57    ForbiddenMediaHasBodyLimit,
58}
59
60impl_static_error!(RawResponsePolicyError,
61    Self::InformationalLimitTooLarge => "informational response limit is too large",
62    Self::BodyLimitTooLarge => "raw response body limit is too large",
63    Self::MissingMediaType => "response media policy has no accepted media type",
64    Self::DuplicateMediaType => "response media policy contains duplicate media types",
65    Self::TooManyAdmittedHeaders => "too many response headers are admitted",
66    Self::DuplicateAdmittedHeader => "an admitted response header is duplicated",
67    Self::UnsafeAdmittedHeader => "an unsafe response header was admitted",
68    Self::ForbiddenMediaHasBodyLimit => "forbidden response media has a nonzero body limit",
69);
70
71/// Complete provider-neutral admission policy for one final HTTP response.
72#[derive(Clone, Copy, Debug, Eq, PartialEq)]
73pub struct RawResponsePolicy<'a> {
74    success_body_bytes: usize,
75    error_body_bytes: usize,
76    success_media: ResponseMediaPolicy<'a>,
77    error_media: ResponseMediaPolicy<'a>,
78    admitted_headers: [Option<HeaderName<'a>>; super::MAX_RESPONSE_HEADERS],
79    admitted_header_count: usize,
80    informational_limit: u8,
81    trailer_policy: TrailerPolicy,
82}
83
84impl<'a> RawResponsePolicy<'a> {
85    /// Validates separate success/error bounds, media rules, and retained headers.
86    pub fn new(
87        success_body_bytes: usize,
88        error_body_bytes: usize,
89        success_media: ResponseMediaPolicy<'a>,
90        error_media: ResponseMediaPolicy<'a>,
91        admitted_headers: &[HeaderName<'a>],
92        informational_limit: u8,
93    ) -> Result<Self, RawResponsePolicyError> {
94        if success_body_bytes > MAX_RAW_RESPONSE_BODY_BYTES
95            || error_body_bytes > MAX_RAW_RESPONSE_BODY_BYTES
96        {
97            return Err(RawResponsePolicyError::BodyLimitTooLarge);
98        }
99        if informational_limit > MAX_INFORMATIONAL_RESPONSES {
100            return Err(RawResponsePolicyError::InformationalLimitTooLarge);
101        }
102        validate_media(success_media)?;
103        validate_media(error_media)?;
104        if (matches!(success_media, ResponseMediaPolicy::Forbidden) && success_body_bytes != 0)
105            || (matches!(error_media, ResponseMediaPolicy::Forbidden) && error_body_bytes != 0)
106        {
107            return Err(RawResponsePolicyError::ForbiddenMediaHasBodyLimit);
108        }
109        validate_headers(admitted_headers)?;
110        let mut owned_headers = [None; super::MAX_RESPONSE_HEADERS];
111        for (index, header) in admitted_headers.iter().copied().enumerate() {
112            if let Some(slot) = owned_headers.get_mut(index) {
113                *slot = Some(header);
114            }
115        }
116        Ok(Self {
117            success_body_bytes,
118            error_body_bytes,
119            success_media,
120            error_media,
121            admitted_headers: owned_headers,
122            admitted_header_count: admitted_headers.len(),
123            informational_limit,
124            trailer_policy: TrailerPolicy::Reject,
125        })
126    }
127
128    /// Returns the body limit selected by the final status class.
129    #[must_use]
130    pub const fn body_limit(self, status: StatusCode) -> usize {
131        if status.is_success() {
132            self.success_body_bytes
133        } else {
134            self.error_body_bytes
135        }
136    }
137
138    /// Returns the larger status-class bound needed for caller response storage.
139    #[must_use]
140    pub const fn max_body_bytes(self) -> usize {
141        if self.success_body_bytes > self.error_body_bytes {
142            self.success_body_bytes
143        } else {
144            self.error_body_bytes
145        }
146    }
147
148    /// Returns the media policy selected by the final status class.
149    #[must_use]
150    pub const fn media_policy(self, status: StatusCode) -> ResponseMediaPolicy<'a> {
151        if status.is_success() {
152            self.success_media
153        } else {
154            self.error_media
155        }
156    }
157
158    /// Reports whether an operation admitted retention of this response header.
159    #[must_use]
160    pub fn admits_header(self, name: &str) -> bool {
161        self.admitted_headers
162            .iter()
163            .take(self.admitted_header_count)
164            .flatten()
165            .any(|candidate| candidate.eq_ignore_ascii_case(name))
166    }
167
168    /// Returns the informational-response limit.
169    #[must_use]
170    pub const fn informational_limit(self) -> u8 {
171        self.informational_limit
172    }
173
174    /// Returns the explicit trailer policy.
175    #[must_use]
176    pub const fn trailer_policy(self) -> TrailerPolicy {
177        self.trailer_policy
178    }
179}
180
181/// Informational or final-response selection failure.
182#[derive(Clone, Copy, Debug, Eq, PartialEq)]
183pub enum InformationalResponseError {
184    /// `101 Switching Protocols` is never admitted by buffered cloud requests.
185    SwitchingProtocols,
186    /// Too many informational responses preceded the final response.
187    TooManyInformationalResponses,
188    /// A final response was still informational.
189    MissingFinalResponse,
190}
191
192impl_static_error!(InformationalResponseError,
193    Self::SwitchingProtocols => "switching protocols is forbidden",
194    Self::TooManyInformationalResponses => "too many informational responses",
195    Self::MissingFinalResponse => "final HTTP response is missing",
196);
197
198/// Small state machine for bounded informational-response handling.
199#[derive(Clone, Copy, Debug, Eq, PartialEq)]
200pub struct InformationalResponseTracker {
201    limit: u8,
202    observed: u8,
203}
204
205impl InformationalResponseTracker {
206    /// Creates a tracker from a validated response policy.
207    #[must_use]
208    pub const fn new(policy: RawResponsePolicy<'_>) -> Self {
209        Self {
210            limit: policy.informational_limit,
211            observed: 0,
212        }
213    }
214
215    /// Observes one response head and reports whether it is the final head.
216    pub fn observe(&mut self, status: StatusCode) -> Result<bool, InformationalResponseError> {
217        if status.get() == 101 {
218            return Err(InformationalResponseError::SwitchingProtocols);
219        }
220        if status.get() < 200 {
221            self.observed = self
222                .observed
223                .checked_add(1)
224                .ok_or(InformationalResponseError::TooManyInformationalResponses)?;
225            if self.observed > self.limit {
226                return Err(InformationalResponseError::TooManyInformationalResponses);
227            }
228            return Ok(false);
229        }
230        Ok(true)
231    }
232
233    /// Returns the number of admitted informational heads.
234    #[must_use]
235    pub const fn observed(self) -> u8 {
236        self.observed
237    }
238}
239
240/// Blocking execution of one already validated raw HTTP request.
241pub trait BlockingRawHttpExecutor {
242    /// Executor-specific phased failure.
243    type Error;
244
245    /// Executes exactly once without implicit authentication or retries.
246    ///
247    /// Implementations must use [`ResponseWriter::begin_attempt`]. Response
248    /// mutation and commitment are available only through the returned guard.
249    fn execute(
250        &self,
251        request: TransportRequest<'_>,
252        policy: RawResponsePolicy<'_>,
253        response: &mut ResponseWriter<'_>,
254    ) -> Result<(), Self::Error>;
255}
256
257/// Runtime-neutral Send asynchronous raw HTTP execution.
258///
259/// Implementations stage responses without commit access. Callers use
260/// [`drive_async_raw`].
261pub trait AsyncRawHttpExecutor {
262    /// Executor-specific phased failure.
263    type Error;
264
265    /// Stages exactly one response without implicit authentication or retries.
266    fn execute<'executor, 'request, 'policy, 'writer, 'buffer>(
267        &'executor self,
268        request: TransportRequest<'request>,
269        policy: RawResponsePolicy<'policy>,
270        response: AsyncResponseStaging<'writer, 'buffer>,
271    ) -> impl Future<Output = Result<ResponseCompletion, Self::Error>> + Send + 'writer
272    where
273        'executor: 'writer,
274        'request: 'writer,
275        'policy: 'writer,
276        'buffer: 'writer;
277}
278
279/// Drives one raw cross-thread async attempt and commits after `Ready(Ok)`.
280pub async fn drive_async_raw<'executor, 'request, 'policy, 'writer, 'buffer, T>(
281    executor: &'executor T,
282    request: TransportRequest<'request>,
283    policy: RawResponsePolicy<'policy>,
284    response: &'writer mut ResponseWriter<'buffer>,
285) -> Result<(), AsyncExecutionError<T::Error>>
286where
287    T: AsyncRawHttpExecutor + ?Sized,
288    'executor: 'writer,
289    'request: 'writer,
290    'policy: 'writer,
291    'buffer: 'writer,
292{
293    let mut attempt = response
294        .begin_attempt()
295        .map_err(AsyncExecutionError::Response)?;
296    let completion = executor
297        .execute(request, policy, attempt.staging())
298        .await
299        .map_err(AsyncExecutionError::Transport)?;
300    attempt
301        .commit_completion(completion)
302        .map_err(AsyncExecutionError::Response)
303}
304
305fn validate_media(policy: ResponseMediaPolicy<'_>) -> Result<(), RawResponsePolicyError> {
306    let media = match policy {
307        ResponseMediaPolicy::Required(media) | ResponseMediaPolicy::Optional(media) => media,
308        ResponseMediaPolicy::Forbidden => return Ok(()),
309    };
310    if media.is_empty() {
311        return Err(RawResponsePolicyError::MissingMediaType);
312    }
313    for (index, value) in media.iter().enumerate() {
314        if media.get(..index).is_some_and(|seen| {
315            seen.iter()
316                .any(|candidate| candidate.as_str().eq_ignore_ascii_case(value.as_str()))
317        }) {
318            return Err(RawResponsePolicyError::DuplicateMediaType);
319        }
320    }
321    Ok(())
322}
323
324fn validate_headers(headers: &[HeaderName<'_>]) -> Result<(), RawResponsePolicyError> {
325    if headers.len() > super::MAX_RESPONSE_HEADERS {
326        return Err(RawResponsePolicyError::TooManyAdmittedHeaders);
327    }
328    for (index, header) in headers.iter().enumerate() {
329        if is_unsafe_response_header(header.as_str()) {
330            return Err(RawResponsePolicyError::UnsafeAdmittedHeader);
331        }
332        if headers.get(..index).is_some_and(|seen| {
333            seen.iter()
334                .any(|candidate| candidate.eq_ignore_ascii_case(header.as_str()))
335        }) {
336            return Err(RawResponsePolicyError::DuplicateAdmittedHeader);
337        }
338    }
339    Ok(())
340}
341
342fn is_unsafe_response_header(name: &str) -> bool {
343    [
344        "authorization",
345        "connection",
346        "cookie",
347        "proxy-authenticate",
348        "proxy-authorization",
349        "set-cookie",
350        "te",
351        "trailer",
352        "transfer-encoding",
353        "upgrade",
354    ]
355    .iter()
356    .any(|candidate| name.eq_ignore_ascii_case(candidate))
357}
358
359#[cfg(test)]
360mod tests {
361    use super::{
362        InformationalResponseError, InformationalResponseTracker, RawResponsePolicy,
363        RawResponsePolicyError, ResponseMediaPolicy,
364    };
365    use crate::transport::{HeaderName, MediaType, StatusCode};
366
367    fn policy(limit: u8) -> Result<RawResponsePolicy<'static>, RawResponsePolicyError> {
368        RawResponsePolicy::new(
369            1024,
370            256,
371            ResponseMediaPolicy::Required(&[MediaType::JSON]),
372            ResponseMediaPolicy::Optional(&[MediaType::JSON]),
373            &[],
374            limit,
375        )
376    }
377
378    #[test]
379    fn selects_independent_success_and_error_limits() {
380        let Ok(policy) = policy(2) else {
381            return;
382        };
383        assert_eq!(policy.body_limit(StatusCode::OK), 1024);
384        assert_eq!(
385            policy.body_limit(StatusCode::new(400).unwrap_or(StatusCode::TOO_MANY_REQUESTS)),
386            256
387        );
388    }
389
390    #[test]
391    fn bounds_informationals_and_rejects_switching_protocols() {
392        let Ok(policy) = policy(2) else {
393            return;
394        };
395        let mut tracker = InformationalResponseTracker::new(policy);
396        let early = StatusCode::new(103).unwrap_or(StatusCode::OK);
397        assert_eq!(tracker.observe(early), Ok(false));
398        assert_eq!(tracker.observe(early), Ok(false));
399        assert_eq!(
400            tracker.observe(early),
401            Err(InformationalResponseError::TooManyInformationalResponses)
402        );
403        let switching = StatusCode::new(101).unwrap_or(StatusCode::OK);
404        assert_eq!(
405            InformationalResponseTracker::new(policy).observe(switching),
406            Err(InformationalResponseError::SwitchingProtocols)
407        );
408    }
409
410    #[test]
411    fn rejects_unsafe_and_duplicate_admitted_headers() {
412        let unsafe_header = HeaderName::new("set-cookie");
413        assert!(unsafe_header.is_ok());
414        if let Ok(unsafe_header) = unsafe_header {
415            assert!(matches!(
416                RawResponsePolicy::new(
417                    1,
418                    1,
419                    ResponseMediaPolicy::Optional(&[MediaType::JSON]),
420                    ResponseMediaPolicy::Optional(&[MediaType::JSON]),
421                    &[unsafe_header],
422                    0,
423                ),
424                Err(RawResponsePolicyError::UnsafeAdmittedHeader)
425            ));
426        }
427        let first = HeaderName::new("x-request-id");
428        let second = HeaderName::new("X-Request-ID");
429        assert!(first.is_ok() && second.is_ok());
430        if let (Ok(first), Ok(second)) = (first, second) {
431            assert!(matches!(
432                RawResponsePolicy::new(
433                    1,
434                    1,
435                    ResponseMediaPolicy::Optional(&[MediaType::JSON]),
436                    ResponseMediaPolicy::Optional(&[MediaType::JSON]),
437                    &[first, second],
438                    0,
439                ),
440                Err(RawResponsePolicyError::DuplicateAdmittedHeader)
441            ));
442        }
443    }
444}
cloud_sdk/transport/raw.rs

cloud_sdk/transport/
raw.rs
