rama_http/service/web/endpoint/extract/body/
octet_stream.rs

1use super::BytesRejection;
2use crate::Request;
3use crate::body::util::BodyExt;
4use crate::service::web::extract::{FromRequest, OptionalFromRequest};
5use crate::utils::macros::{composite_http_rejection, define_http_rejection};
6use rama_core::bytes::Bytes;
7use rama_http_types::{HeaderMap, header};
8use rama_utils::macros::impl_deref;
9
10/// Wrapper used to extract `application/octet-stream` payloads from request bodies.
11///
12/// The request `Content-Type` must be `application/octet-stream`, or absent —
13/// per RFC 9110 §8.3 a receiver MAY treat a missing `Content-Type` as
14/// `application/octet-stream`. Any other content type is rejected with
15/// `415 Unsupported Media Type`. The full body is collected into [`Bytes`].
16///
17/// # Performance
18///
19/// `OctetStream` collects the entire body into a single contiguous buffer
20/// before the handler runs — convenient for small uploads, but unsuitable for
21/// large or streaming payloads. For those, use the [`Body`](super::Body)
22/// extractor and consume frame-by-frame; cap the request body size with the
23/// existing body-limit machinery to bound memory use.
24///
25/// For producing octet-stream responses, see
26/// [`response::OctetStream`](crate::service::web::endpoint::response::OctetStream).
27#[derive(Debug, Clone)]
28pub struct OctetStream(pub Bytes);
29
30impl_deref!(OctetStream: Bytes);
31
32impl From<Bytes> for OctetStream {
33    fn from(value: Bytes) -> Self {
34        Self(value)
35    }
36}
37
38impl From<OctetStream> for Bytes {
39    fn from(value: OctetStream) -> Self {
40        value.0
41    }
42}
43
44define_http_rejection! {
45    #[status = UNSUPPORTED_MEDIA_TYPE]
46    #[body = "OctetStream requests must have `Content-Type: application/octet-stream` (or no Content-Type)"]
47    /// Rejection type for [`OctetStream`]
48    /// used if the `Content-Type` header is present and its value is not
49    /// `application/octet-stream`.
50    pub struct InvalidOctetStreamContentType;
51}
52
53composite_http_rejection! {
54    /// Rejection used for [`OctetStream`]
55    ///
56    /// Contains one variant for each way the [`OctetStream`] extractor
57    /// can fail.
58    pub enum OctetStreamRejection {
59        InvalidOctetStreamContentType,
60        BytesRejection,
61    }
62}
63
64impl FromRequest for OctetStream {
65    type Rejection = OctetStreamRejection;
66
67    async fn from_request(req: Request) -> Result<Self, Self::Rejection> {
68        match content_type_match(req.headers()) {
69            ContentTypeMatch::Match | ContentTypeMatch::Absent => {}
70            ContentTypeMatch::Mismatch => return Err(InvalidOctetStreamContentType.into()),
71        }
72
73        match req.into_body().collect().await {
74            Ok(c) => Ok(Self(c.to_bytes())),
75            Err(err) => Err(BytesRejection::from_err(err).into()),
76        }
77    }
78}
79
80impl OptionalFromRequest for OctetStream {
81    type Rejection = OctetStreamRejection;
82
83    /// Mirrors the non-optional [`FromRequest`] acceptance policy: any
84    /// indication of a body (`Content-Type`, `Content-Length > 0`, or
85    /// `Transfer-Encoding`) delegates to `FromRequest`. Only when the
86    /// request clearly carries no body do we return `None`.
87    ///
88    /// This avoids the pitfall of silently dropping a body the
89    /// non-optional extractor would happily parse — important now that the
90    /// non-optional path treats a missing `Content-Type` as
91    /// `application/octet-stream` per RFC 9110 §8.3.
92    async fn from_request(req: Request) -> Result<Option<Self>, Self::Rejection> {
93        if request_has_body(req.headers()) {
94            let v = <Self as FromRequest>::from_request(req).await?;
95            Ok(Some(v))
96        } else {
97            Ok(None)
98        }
99    }
100}
101
102fn request_has_body(headers: &HeaderMap) -> bool {
103    if headers.get(header::CONTENT_TYPE).is_some() {
104        return true;
105    }
106    if headers.get(header::TRANSFER_ENCODING).is_some() {
107        return true;
108    }
109    headers
110        .get(header::CONTENT_LENGTH)
111        .and_then(|v| v.to_str().ok())
112        .and_then(|s| s.parse::<u64>().ok())
113        .is_some_and(|n| n > 0)
114}
115
116enum ContentTypeMatch {
117    Match,
118    Absent,
119    Mismatch,
120}
121
122fn content_type_match(headers: &HeaderMap) -> ContentTypeMatch {
123    let Some(value) = headers.get(header::CONTENT_TYPE) else {
124        return ContentTypeMatch::Absent;
125    };
126    let parsed = value
127        .to_str()
128        .ok()
129        .and_then(|s| s.parse::<crate::mime::Mime>().ok());
130    match parsed {
131        Some(mime)
132            if mime.type_() == crate::mime::APPLICATION
133                && mime.subtype() == crate::mime::OCTET_STREAM =>
134        {
135            ContentTypeMatch::Match
136        }
137        _ => ContentTypeMatch::Mismatch,
138    }
139}
140
141#[cfg(test)]
142mod test {
143    use super::*;
144    use crate::StatusCode;
145    use crate::service::web::WebService;
146    use rama_core::Service;
147
148    #[tokio::test]
149    async fn test_octet_stream() {
150        let service =
151            WebService::default().with_post("/", async |OctetStream(body): OctetStream| {
152                assert_eq!(body.as_ref(), b"\x00\x01\x02\x03");
153            });
154
155        let req = rama_http_types::Request::builder()
156            .method(rama_http_types::Method::POST)
157            .header(
158                rama_http_types::header::CONTENT_TYPE,
159                "application/octet-stream",
160            )
161            .body(vec![0u8, 1, 2, 3].into())
162            .unwrap();
163        let resp = service.serve(req).await.unwrap();
164        assert_eq!(resp.status(), StatusCode::OK);
165    }
166
167    #[tokio::test]
168    async fn test_octet_stream_missing_content_type() {
169        let service = WebService::default()
170            .with_post("/", async |OctetStream(_): OctetStream| StatusCode::OK);
171
172        let req = rama_http_types::Request::builder()
173            .method(rama_http_types::Method::POST)
174            .header(rama_http_types::header::CONTENT_TYPE, "text/plain")
175            .body(vec![0u8, 1, 2, 3].into())
176            .unwrap();
177        let resp = service.serve(req).await.unwrap();
178        assert_eq!(resp.status(), StatusCode::UNSUPPORTED_MEDIA_TYPE);
179    }
180
181    #[tokio::test]
182    async fn test_octet_stream_optional_present() {
183        let service = WebService::default().with_post("/", async |body: Option<OctetStream>| {
184            let OctetStream(b) = body.expect("body present");
185            assert_eq!(b.as_ref(), b"data");
186        });
187
188        let req = rama_http_types::Request::builder()
189            .method(rama_http_types::Method::POST)
190            .header(
191                rama_http_types::header::CONTENT_TYPE,
192                "application/octet-stream",
193            )
194            .body("data".into())
195            .unwrap();
196        let resp = service.serve(req).await.unwrap();
197        assert_eq!(resp.status(), StatusCode::OK);
198    }
199
200    #[tokio::test]
201    async fn test_octet_stream_accepts_absent_content_type() {
202        // Per RFC 9110 §8.3 a receiver MAY treat a missing Content-Type as
203        // application/octet-stream. We do.
204        let service =
205            WebService::default().with_post("/", async |OctetStream(body): OctetStream| {
206                assert_eq!(body.as_ref(), b"raw");
207            });
208
209        let req = rama_http_types::Request::builder()
210            .method(rama_http_types::Method::POST)
211            .body("raw".into())
212            .unwrap();
213        let resp = service.serve(req).await.unwrap();
214        assert_eq!(resp.status(), StatusCode::OK);
215    }
216
217    #[tokio::test]
218    async fn test_octet_stream_optional_no_content_type_with_body() {
219        // Regression: `Option<OctetStream>` previously returned `None` when
220        // no Content-Type was set, even though the non-optional extractor
221        // would happily parse the same request. Now the optional variant
222        // mirrors the acceptance policy and treats a body indicated by
223        // Content-Length as present.
224        let service = WebService::default().with_post("/", async |body: Option<OctetStream>| {
225            let OctetStream(b) = body.expect("body present");
226            assert_eq!(b.as_ref(), b"data");
227        });
228
229        let req = rama_http_types::Request::builder()
230            .method(rama_http_types::Method::POST)
231            .header(rama_http_types::header::CONTENT_LENGTH, "4")
232            .body("data".into())
233            .unwrap();
234        let resp = service.serve(req).await.unwrap();
235        assert_eq!(resp.status(), StatusCode::OK);
236    }
237
238    #[tokio::test]
239    async fn test_octet_stream_optional_absent() {
240        let service = WebService::default().with_post("/", async |body: Option<OctetStream>| {
241            assert!(body.is_none());
242            StatusCode::OK
243        });
244
245        let req = rama_http_types::Request::builder()
246            .method(rama_http_types::Method::POST)
247            .body(rama_http_types::Body::empty())
248            .unwrap();
249        let resp = service.serve(req).await.unwrap();
250        assert_eq!(resp.status(), StatusCode::OK);
251    }
252}
rama_http/service/web/endpoint/extract/body/octet_stream.rs

rama_http/service/web/endpoint/extract/body/
octet_stream.rs
