io_http/rfc9112/
send.rs

1//! I/O-free coroutine sending an HTTP/1.1 request and receiving its
2//! response ([RFC 9112]).
3//!
4//! Serialises the request, drives the read/parse cycle through
5//! [`Http11HeadersRead`] for the head and selects a body-reading
6//! strategy from the response headers:
7//!
8//! | Strategy     | Trigger                      |
9//! |--------------|------------------------------|
10//! | Chunked      | `Transfer-Encoding: chunked` |
11//! | Fixed-length | `Content-Length: <n>`        |
12//! | Read-to-EOF  | Neither header present       |
13//!
14//! # Example
15//!
16//! ```rust,no_run
17//! use std::{io::{Read, Write}, net::TcpStream};
18//!
19//! use io_http::{
20//!     coroutine::*,
21//!     rfc9110::{request::HttpRequest, send::HttpSendYield},
22//!     rfc9112::send::Http11Send,
23//! };
24//! use url::Url;
25//!
26//! let url = Url::parse("http://example.com/").unwrap();
27//! let request = HttpRequest::get(url.clone())
28//!     .header("Host", url.host_str().unwrap())
29//!     .header("Connection", "close");
30//!
31//! let mut stream = TcpStream::connect("example.com:80").unwrap();
32//! let mut send = Http11Send::new(request);
33//! let mut arg: Option<&[u8]> = None;
34//! let mut buf = [0u8; 4096];
35//!
36//! let (response, keep_alive) = loop {
37//!     match send.resume(arg.take()) {
38//!         HttpCoroutineState::Complete(Ok(out)) => break (out.response, out.keep_alive),
39//!         HttpCoroutineState::Complete(Err(err)) => panic!("{err}"),
40//!         HttpCoroutineState::Yielded(HttpSendYield::WantsRead) => {
41//!             let n = stream.read(&mut buf).unwrap();
42//!             arg = Some(&buf[..n]);
43//!         }
44//!         HttpCoroutineState::Yielded(HttpSendYield::WantsWrite(bytes)) => {
45//!             stream.write_all(&bytes).unwrap();
46//!         }
47//!         HttpCoroutineState::Yielded(HttpSendYield::WantsRedirect { url, .. }) => {
48//!             panic!("redirect to {url}");
49//!         }
50//!     }
51//! };
52//!
53//! println!("{}", *response.status);
54//! # let _ = keep_alive;
55//! ```
56//!
57//! [RFC 9112]: https://www.rfc-editor.org/rfc/rfc9112
58
59use core::mem;
60
61use alloc::{borrow::ToOwned, string::String, vec::Vec};
62
63use httparse::Error as HttparseError;
64use log::{debug, trace};
65use thiserror::Error;
66use url::Url;
67
68use crate::{
69    coroutine::*,
70    http_try,
71    rfc1945::version::HTTP_10,
72    rfc9110::{
73        headers::{HTTP_CONTENT_LENGTH, HTTP_LOCATION, HTTP_TRANSFER_ENCODING},
74        request::HttpRequest,
75        response::HttpResponse,
76        send::{HTTP_DEFAULT_MAX_BODY_LEN, HttpSendOutput, HttpSendYield},
77    },
78    rfc9112::{
79        chunk::{Http11ChunksRead, Http11ChunksReadError},
80        read_headers::{HTTP_DEFAULT_MAX_HEADERS_LEN, Http11HeadersRead, Http11HeadersReadError},
81    },
82};
83
84/// Failure causes during the HTTP/1.1 send flow.
85#[derive(Debug, Error)]
86pub enum Http11SendError {
87    /// The stream reached EOF before the response was complete.
88    #[error("HTTP/1.1 send failed: reached unexpected EOF")]
89    Eof,
90    /// The response head could not be parsed.
91    #[error("HTTP/1.1 send failed: parse response headers: {0}")]
92    ParseResponseHeaders(HttparseError),
93    /// The content length header value is not a valid integer.
94    #[error("HTTP/1.1 send failed: invalid content length `{0}`")]
95    InvalidContentLength(String),
96    /// The chunked-body decoder failed.
97    #[error("HTTP/1.1 send failed: {0}")]
98    ReadChunks(#[from] Http11ChunksReadError),
99    /// The response head exceeds [`Http11Send::max_headers_len`].
100    #[error("HTTP/1.1 send failed: response head exceeds {0} bytes")]
101    HeadersTooLarge(usize),
102    /// The response body exceeds [`Http11Send::max_body_len`].
103    #[error("HTTP/1.1 send failed: response body exceeds {0} bytes")]
104    BodyTooLarge(usize),
105}
106
107impl From<Http11HeadersReadError> for Http11SendError {
108    fn from(err: Http11HeadersReadError) -> Self {
109        match err {
110            Http11HeadersReadError::Eof => Self::Eof,
111            Http11HeadersReadError::ParseResponseHeaders(e) => Self::ParseResponseHeaders(e),
112            Http11HeadersReadError::TooLarge(limit) => Self::HeadersTooLarge(limit),
113        }
114    }
115}
116
117#[derive(Debug)]
118enum State {
119    ReadHeaders(Http11HeadersRead),
120    BodyChunks(Http11ChunksRead),
121    BodyLength(usize),
122    BodyEof,
123}
124
125/// I/O-free coroutine to send an HTTP/1.1 request and receive its response.
126#[derive(Debug)]
127pub struct Http11Send {
128    /// Maximum size in bytes of the response head. Defaults to
129    /// [`HTTP_DEFAULT_MAX_HEADERS_LEN`].
130    pub max_headers_len: usize,
131    /// Maximum size in bytes of the response body, whatever its
132    /// framing (fixed-length, chunked or read-to-EOF). Defaults to
133    /// [`HTTP_DEFAULT_MAX_BODY_LEN`].
134    pub max_body_len: usize,
135    request_url: Url,
136    state: State,
137    wants_write: Option<Vec<u8>>,
138    is_conn_closed: bool,
139    response: Option<HttpResponse>,
140    buf: Vec<u8>,
141}
142
143impl Http11Send {
144    /// Creates a new coroutine that will send the given request and
145    /// receive its response.
146    pub fn new(req: HttpRequest) -> Self {
147        debug!("prepare request to send");
148        trace!("{req:?}");
149
150        let request_url = req.url.clone();
151        let bytes = req.to_http_11_vec();
152
153        Self {
154            max_headers_len: HTTP_DEFAULT_MAX_HEADERS_LEN,
155            max_body_len: HTTP_DEFAULT_MAX_BODY_LEN,
156            request_url,
157            state: State::ReadHeaders(Http11HeadersRead::default()),
158            wants_write: Some(bytes),
159            is_conn_closed: false,
160            response: None,
161            buf: Vec::new(),
162        }
163    }
164
165    fn finish(
166        &self,
167        response: HttpResponse,
168        remaining: Vec<u8>,
169    ) -> HttpCoroutineState<HttpSendYield, Result<HttpSendOutput, Http11SendError>> {
170        let keep_alive = !self.is_conn_closed;
171
172        if response.status.is_redirection()
173            && let Some(location) = response.header(HTTP_LOCATION)
174            && let Ok(url) = self.request_url.join(location)
175        {
176            let same_scheme = self.request_url.scheme() == url.scheme();
177            let same_host =
178                self.request_url.host() == url.host() && self.request_url.port() == url.port();
179            let same_origin = same_scheme && same_host;
180
181            return HttpCoroutineState::Yielded(HttpSendYield::WantsRedirect {
182                url,
183                response,
184                keep_alive,
185                same_origin,
186            });
187        }
188
189        HttpCoroutineState::Complete(Ok(HttpSendOutput {
190            response,
191            remaining,
192            keep_alive,
193        }))
194    }
195}
196
197impl HttpCoroutine for Http11Send {
198    type Yield = HttpSendYield;
199    type Return = Result<HttpSendOutput, Http11SendError>;
200
201    fn resume(&mut self, mut arg: Option<&[u8]>) -> HttpCoroutineState<Self::Yield, Self::Return> {
202        loop {
203            if let Some(bytes) = self.wants_write.take() {
204                return HttpCoroutineState::Yielded(HttpSendYield::WantsWrite(bytes));
205            }
206
207            if let State::ReadHeaders(rh) = &mut self.state {
208                rh.max_len = self.max_headers_len;
209            }
210
211            match &mut self.state {
212                State::ReadHeaders(rh) => match rh.resume(arg.take()) {
213                    HttpCoroutineState::Yielded(HttpYield::WantsRead) => {
214                        return HttpCoroutineState::Yielded(HttpSendYield::WantsRead);
215                    }
216                    HttpCoroutineState::Yielded(HttpYield::WantsWrite(_)) => {
217                        unreachable!("Http11HeadersRead never writes");
218                    }
219                    HttpCoroutineState::Complete(Err(err)) => {
220                        return HttpCoroutineState::Complete(Err(err.into()));
221                    }
222                    HttpCoroutineState::Complete(Ok(out)) => {
223                        let mut response = out.response;
224                        let is_http10 = response.version == HTTP_10;
225                        self.is_conn_closed = !out.keep_alive;
226                        let status = *response.status;
227
228                        if status == 204 || status == 304 {
229                            return self.finish(response, out.remaining);
230                        }
231
232                        if !is_http10 {
233                            let chunked = response
234                                .header(HTTP_TRANSFER_ENCODING)
235                                .is_some_and(|enc| enc.eq_ignore_ascii_case("chunked"));
236                            if chunked {
237                                let mut chunks = Http11ChunksRead::default();
238                                chunks.max_body_len = self.max_body_len;
239
240                                // NOTE: an empty slice means EOF to the
241                                // decoder, so nothing pre-read is None.
242                                let remaining = &out.remaining;
243                                let arg = (!remaining.is_empty()).then_some(remaining.as_slice());
244
245                                match chunks.resume(arg) {
246                                    HttpCoroutineState::Complete(Ok(chunk_out)) => {
247                                        response.body = chunk_out.body;
248                                        return self.finish(response, chunk_out.remaining);
249                                    }
250                                    HttpCoroutineState::Yielded(HttpYield::WantsRead) => {
251                                        self.response = Some(response);
252                                        debug!("reading chunked body");
253                                        self.state = State::BodyChunks(chunks);
254                                        return HttpCoroutineState::Yielded(
255                                            HttpSendYield::WantsRead,
256                                        );
257                                    }
258                                    HttpCoroutineState::Yielded(HttpYield::WantsWrite(_)) => {
259                                        unreachable!("Http11ChunksRead never writes");
260                                    }
261                                    HttpCoroutineState::Complete(Err(err)) => {
262                                        return HttpCoroutineState::Complete(Err(err.into()));
263                                    }
264                                }
265                            }
266                        }
267
268                        if let Some(len_str) = response.header(HTTP_CONTENT_LENGTH) {
269                            let len_str = len_str.trim();
270                            let is_digits = len_str.bytes().all(|b| b.is_ascii_digit());
271
272                            let Some(len) =
273                                is_digits.then(|| len_str.parse::<usize>().ok()).flatten()
274                            else {
275                                let err = Http11SendError::InvalidContentLength(len_str.to_owned());
276                                return HttpCoroutineState::Complete(Err(err));
277                            };
278
279                            if len > self.max_body_len {
280                                let err = Http11SendError::BodyTooLarge(self.max_body_len);
281                                return HttpCoroutineState::Complete(Err(err));
282                            }
283
284                            self.buf = out.remaining;
285                            self.response = Some(response);
286                            debug!("reading body of length {len}");
287                            self.state = State::BodyLength(len);
288                            continue;
289                        }
290
291                        self.buf = out.remaining;
292                        self.response = Some(response);
293                        debug!("reading body until connection closes");
294                        self.state = State::BodyEof;
295                    }
296                },
297                State::BodyChunks(chunks) => {
298                    let chunk_out = http_try!(chunks, arg.take());
299                    let mut response = self.response.take().expect("response missing");
300                    response.body = chunk_out.body;
301                    return self.finish(response, chunk_out.remaining);
302                }
303                State::BodyLength(len) => {
304                    match arg.take() {
305                        Some(&[]) if *len > self.buf.len() => {
306                            return HttpCoroutineState::Complete(Err(Http11SendError::Eof));
307                        }
308                        Some(data) => self.buf.extend_from_slice(data),
309                        None => {}
310                    }
311
312                    if *len > self.buf.len() {
313                        trace!("received incomplete body {len}/{}", self.buf.len());
314                        return HttpCoroutineState::Yielded(HttpSendYield::WantsRead);
315                    }
316
317                    let body = self.buf.drain(..*len).collect();
318                    let remaining = mem::take(&mut self.buf);
319                    let mut response = self.response.take().expect("response missing");
320                    response.body = body;
321                    return self.finish(response, remaining);
322                }
323                State::BodyEof => match arg.take() {
324                    _ if self.buf.len() > self.max_body_len => {
325                        let err = Http11SendError::BodyTooLarge(self.max_body_len);
326                        return HttpCoroutineState::Complete(Err(err));
327                    }
328                    Some(&[]) => {
329                        let buf = mem::take(&mut self.buf);
330                        let mut response = self.response.take().expect("response missing");
331                        response.body = buf;
332                        return self.finish(response, Vec::new());
333                    }
334                    Some(data) if data.len() > self.max_body_len.saturating_sub(self.buf.len()) => {
335                        let err = Http11SendError::BodyTooLarge(self.max_body_len);
336                        return HttpCoroutineState::Complete(Err(err));
337                    }
338                    Some(data) => {
339                        self.buf.extend_from_slice(data);
340                        return HttpCoroutineState::Yielded(HttpSendYield::WantsRead);
341                    }
342                    None => {
343                        return HttpCoroutineState::Yielded(HttpSendYield::WantsRead);
344                    }
345                },
346            }
347        }
348    }
349}
350
351#[cfg(test)]
352mod tests {
353    use crate::rfc9112::send::*;
354
355    #[test]
356    fn body_chunks_completes() {
357        let req = HttpRequest::get("https://example.com".try_into().unwrap());
358        let mut coroutine = Http11Send::new(req);
359
360        let bytes = expect_wants_write(&mut coroutine, None);
361        assert_eq!(
362            bytes,
363            b"GET / HTTP/1.1\r\nhost: example.com\r\ncontent-length: 0\r\n\r\n"
364        );
365
366        expect_wants_read(&mut coroutine, None);
367
368        let reply = b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nhello\r\n6\r\n world\r\n0\r\n\r\n";
369        let out = expect_complete_ok(&mut coroutine, Some(reply));
370        assert_eq!(out.response.version, "HTTP/1.1");
371        assert_eq!(*out.response.status, 200);
372        assert_eq!(out.response.body, b"hello world");
373        assert!(out.remaining.is_empty());
374        assert!(out.keep_alive);
375    }
376
377    #[test]
378    fn body_length_completes() {
379        let req = HttpRequest::get("https://example.com".try_into().unwrap());
380        let mut coroutine = Http11Send::new(req);
381
382        expect_wants_write(&mut coroutine, None);
383        expect_wants_read(&mut coroutine, None);
384
385        let reply = b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nhello";
386        let out = expect_complete_ok(&mut coroutine, Some(reply));
387        assert_eq!(*out.response.status, 200);
388        assert_eq!(out.response.body, b"hello");
389        assert!(out.keep_alive);
390    }
391
392    #[test]
393    fn body_eof_completes() {
394        let req = HttpRequest::get("https://example.com".try_into().unwrap());
395        let mut coroutine = Http11Send::new(req);
396
397        expect_wants_write(&mut coroutine, None);
398        expect_wants_read(&mut coroutine, None);
399        expect_wants_read(&mut coroutine, Some(b"HTTP/1.1 200 OK\r\n\r\nhello "));
400        expect_wants_read(&mut coroutine, Some(b"world"));
401
402        let out = expect_complete_ok(&mut coroutine, Some(b""));
403        assert_eq!(out.response.body, b"hello world");
404    }
405
406    #[test]
407    fn invalid_content_length_errors() {
408        let req = HttpRequest::get("https://example.com".try_into().unwrap());
409        let mut coroutine = Http11Send::new(req);
410
411        expect_wants_write(&mut coroutine, None);
412        expect_wants_read(&mut coroutine, None);
413
414        let reply = b"HTTP/1.1 200 OK\r\nContent-Length: notanumber\r\n\r\n";
415        let err = expect_complete_err(&mut coroutine, Some(reply));
416        let Http11SendError::InvalidContentLength(s) = err else {
417            panic!("expected InvalidContentLength, got {err:?}");
418        };
419        assert_eq!(s, "notanumber");
420    }
421
422    #[test]
423    fn redirect_yields_wants_redirect() {
424        let req = HttpRequest::get("http://example.com/old".try_into().unwrap());
425        let mut coroutine = Http11Send::new(req);
426
427        expect_wants_write(&mut coroutine, None);
428        expect_wants_read(&mut coroutine, None);
429
430        let reply =
431            b"HTTP/1.1 301 Moved Permanently\r\nLocation: /new\r\nContent-Length: 0\r\n\r\n";
432        match coroutine.resume(Some(reply)) {
433            HttpCoroutineState::Yielded(HttpSendYield::WantsRedirect {
434                url,
435                same_origin,
436                keep_alive,
437                ..
438            }) => {
439                assert_eq!(url.path(), "/new");
440                assert!(same_origin);
441                assert!(keep_alive);
442            }
443            state => panic!("expected WantsRedirect, got {state:?}"),
444        }
445    }
446
447    fn expect_wants_write(cor: &mut Http11Send, arg: Option<&[u8]>) -> Vec<u8> {
448        match cor.resume(arg) {
449            HttpCoroutineState::Yielded(HttpSendYield::WantsWrite(bytes)) => bytes,
450            state => panic!("expected WantsWrite, got {state:?}"),
451        }
452    }
453
454    fn expect_wants_read(cor: &mut Http11Send, arg: Option<&[u8]>) {
455        match cor.resume(arg) {
456            HttpCoroutineState::Yielded(HttpSendYield::WantsRead) => {}
457            state => panic!("expected WantsRead, got {state:?}"),
458        }
459    }
460
461    fn expect_complete_ok(cor: &mut Http11Send, arg: Option<&[u8]>) -> HttpSendOutput {
462        match cor.resume(arg) {
463            HttpCoroutineState::Complete(Ok(out)) => out,
464            state => panic!("expected Complete(Ok), got {state:?}"),
465        }
466    }
467
468    fn expect_complete_err(cor: &mut Http11Send, arg: Option<&[u8]>) -> Http11SendError {
469        match cor.resume(arg) {
470            HttpCoroutineState::Complete(Err(err)) => err,
471            state => panic!("expected Complete(Err), got {state:?}"),
472        }
473    }
474}
io_http/rfc9112/send.rs

io_http/rfc9112/
send.rs
