Oregami
Repositories/oxedyne/fe2o3

oxedyne/fe2o3/fe2o3_net/src/http/header.rs

20.7 KiB, 89 runs

created by r1870400018:577, which is this file's identity for as long as the history lasts, whatever it is later renamed to

download · who wrote it · its history

1use crate::{
2 constant,
3 http::{
4 fields::{
5 HeaderField,
6 HeaderFields,
7 HeaderFieldValue,
8 HeaderName,
9 },
10 loc::HttpLocator,
11 status::HttpStatus,
12 },
13};
14
15use oxedyne_fe2o3_core::prelude::*;
16
17use std::{
18 fmt,
19 str::FromStr,
20};
21// `Pin` is only the async `HttpHeader::read`'s.
22#[cfg(feature = "async")]
23use std::pin::Pin;
24
25use strum::{
26 Display,
27 EnumString,
28};
29#[cfg(feature = "async")]
30use tokio::{
31 io::{
32 AsyncRead,
33 AsyncReadExt,
34 //AsyncWriteExt,
35 },
36};
37
38#[derive(Debug)]
39pub enum HttpVersion {
40 Http1_1,
41 Http2_0,
42 Http3_0,
43}
44
45impl fmt::Display for HttpVersion {
46 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
47 match self {
48 Self::Http1_1 => write!(f, "HTTP/1.1"),
49 Self::Http2_0 => write!(f, "HTTP/2"),
50 Self::Http3_0 => write!(f, "HTTP/3"),
51 }
52 }
53}
54
55impl FromStr for HttpVersion {
56 type Err = Error<ErrTag>;
57
58 fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
59 Ok(match s {
60 "HTTP/1.1" => Self::Http1_1,
61 "HTTP/2" => Self::Http2_0,
62 "HTTP/3" => Self::Http3_0,
63 _ => return Err(err!(
64 "Unrecognised HTTP version {}.", s;
65 IO, Network, Unknown, Input))
66 })
67 }
68}
69
70#[derive(Clone, Copy, Debug, Display, EnumString, PartialEq)]
71pub enum HttpMethod {
72 CONNECT,
73 DELETE,
74 GET,
75 HEAD,
76 OPTIONS,
77 PATCH,
78 POST,
79 PUT,
80 TRACE,
81}
82
83impl HttpMethod {
84 pub fn body_required(&self) -> bool {
85 match self {
86 Self::POST |
87 Self::PUT |
88 Self::PATCH => true,
89 _ => false,
90 }
91 }
92}
93
94#[derive(Clone, Debug)]
95pub enum HttpHeadline {
96 Request {
97 method: HttpMethod,
98 loc: HttpLocator,
99 },
100 Response {
101 status: HttpStatus,
102 }
103}
104
105impl HttpHeadline {
106
107 /// Used when the caller may not know if the message is a request or a response.
108 pub fn parse(
109 line: &str,
110 is_request: Option<bool>,
111 )
112 -> Outcome<(Self, HttpVersion)>
113 {
114 match is_request {
115 Some(true) => Self::parse_request(line),
116 Some(false) => Self::parse_response(line),
117 None => {
118 let mut parts = line.split_whitespace();
119 if let Some(first_str) = parts.next() {
120 match HttpMethod::from_str(first_str) {
121 Ok(method) => if let Some(loc_str) = parts.next() {
122 if let Some(version_str) = parts.next() {
123 return Ok((
124 HttpHeadline::Request {
125 method,
126 loc: res!(HttpLocator::new(loc_str)),
127 },
128 res!(HttpVersion::from_str(version_str)),
129 ));
130 }
131 },
132 Err(_) => match HttpVersion::from_str(first_str) {
133 Ok(version) => if let Some(code_str) = parts.next() {
134 return Ok((
135 HttpHeadline::Response {
136 status: res!(HttpStatus::from_str(code_str)),
137 },
138 version,
139 ));
140 },
141 Err(_) => return Err(err!(
142 "HTTP message headline '{}' begins with an unrecognised word.", line;
143 IO, Network, Invalid, Input)),
144 },
145 }
146 }
147 Err(err!(
148 "HTTP request headline '{}' invalid, expected at least 3 components.", line;
149 IO, Network, Missing, Input))
150 },
151 }
152 }
153
154 /// Used when the caller knows the message is a request.
155 pub fn parse_request(line: &str) -> Outcome<(Self, HttpVersion)> {
156 let mut parts = line.split_whitespace();
157 if let Some(method_str) = parts.next() {
158 if let Some(loc_str) = parts.next() {
159 if let Some(version_str) = parts.next() {
160 return Ok((
161 HttpHeadline::Request {
162 method: res!(HttpMethod::from_str(method_str)),
163 loc: res!(HttpLocator::new(loc_str)),
164 },
165 res!(HttpVersion::from_str(version_str)),
166 ));
167 }
168 }
169 }
170 Err(err!(
171 "HTTP request headline '{}' invalid, expected at least 3 components.", line;
172 IO, Network, Missing, Input))
173 }
174
175 /// Used when the caller knows the message is a response.
176 pub fn parse_response(line: &str) -> Outcome<(Self, HttpVersion)> {
177 let mut parts = line.split_whitespace();
178 if let Some(version_str) = parts.next() {
179 if let Some(code_str) = parts.next() {
180 return Ok((
181 HttpHeadline::Response {
182 status: res!(HttpStatus::from_str(code_str)),
183 },
184 res!(HttpVersion::from_str(version_str)),
185 ));
186 }
187 }
188 Err(err!(
189 "HTTP response headline '{}' invalid, expected at least 3 components.", line;
190 IO, Network, Missing, Input))
191 }
192
193}
194
195#[derive(Debug)]
196pub struct HttpHeader {
197 pub version: HttpVersion,
198 pub headline: HttpHeadline,
199 pub fields: HeaderFields,
200}
201
202impl Default for HttpHeader {
203 fn default() -> Self {
204 Self {
205 version: HttpVersion::Http1_1,
206 headline: HttpHeadline::Request {
207 method: HttpMethod::GET,
208 loc: HttpLocator::default(),
209 },
210 fields: HeaderFields::default(),
211 }
212 }
213}
214
215impl fmt::Display for HttpHeader {
216 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
217 match &self.headline {
218 HttpHeadline::Request { method, loc } => {
219 ok!(write!(f, "{} {} {}\r\n", method, loc, self.version));
220 },
221 HttpHeadline::Response { status } => {
222 ok!(write!(f, "{} {} {}\r\n", self.version, status, status.desc()));
223 },
224 }
225 for (k, header_field_values) in self.fields.iter() {
226 for header_field_value in header_field_values {
227 // A field that renders to an empty value tells the peer nothing,
228 // so it is left off the wire rather than sent as a bare name.
229 if header_field_value.is_wire_empty() {
230 continue;
231 }
232 ok!(write!(f, "{}: {}\r\n", k, header_field_value));
233 }
234 }
235 write!(f, "\r\n")
236 }
237}
238
239impl HttpHeader {
240 pub fn as_vec(&self) -> Vec<u8> {
241 fmt!("{}", self).into_bytes()
242 }
243
244 /// Find the `CRLF CRLF` that ends the header block, scanning from `from`.
245 ///
246 /// The search runs over the accumulated header bytes rather than the latest
247 /// read, and the caller resumes it three bytes back from the join, so a
248 /// terminator split across two reads in any of its three places is still
249 /// found.
250 #[cfg(feature = "async")]
251 fn find_terminator(bytes: &[u8], from: usize) -> Option<usize> {
252 bytes[from..]
253 .windows(4)
254 .position(|w| w == b"\r\n\r\n")
255 .map(|pos| from + pos)
256 }
257
258 #[cfg(feature = "async")]
259 pub async fn read<
260 'a,
261 const CHUNK_SIZE: usize,
262 R: AsyncRead + Unpin,
263 >(
264 mut stream: Pin<&mut R>,
265 remnant: &Vec<u8>,
266 is_request: Option<bool>,
267 limits: Option<&crate::http::msg::ReadLimits>,
268 )
269 -> Outcome<Option<(Self, Vec<u8>, usize)>>
270 {
271 //trace!("Entered HttpHeader::read");
272 let mut header_bytes = Vec::new();
273 let mut buf = [0u8; CHUNK_SIZE];
274
275 let max_header_bytes = limits.and_then(|l| l.max_header_bytes);
276 let header_timeout = limits.and_then(|l| l.header_read_timeout);
277 let started = std::time::Instant::now();
278
279 // The remnant of the previous message may already hold a complete header.
280 header_bytes.extend_from_slice(&remnant);
281 if let Some(pos) = Self::find_terminator(&header_bytes, 0) {
282 let rest = header_bytes[pos + 4..].to_vec();
283 header_bytes.truncate(pos);
284 let (header_str, content_length) = res!(Self::parse_header_str(&header_bytes));
285 return Ok(Some((
286 res!(Self::parse(header_str, is_request)),
287 rest,
288 content_length,
289 )));
290 }
291
292 if let Some(lim) = max_header_bytes {
293 if header_bytes.len() > lim {
294 return Err(err!(
295 "HTTP header carry-over from previous request ({} \
296 bytes) already exceeds the configured header limit \
297 of {} bytes.", header_bytes.len(), lim;
298 IO, Network, Input, TooBig));
299 }
300 }
301
302 // Read from the stream until the header is complete.
303 loop {
304 // Slowloris guard: bound the total wall-clock time the
305 // reader will spend accumulating header bytes. The
306 // elapsed check runs on every loop iteration so a client
307 // that manages to ship one byte before the deadline
308 // still gets evicted on the next iteration.
309 let bytes_read = match header_timeout {
310 Some(budget) => {
311 let elapsed = started.elapsed();
312 if elapsed >= budget {
313 return Err(err!(
314 "HTTP header read timed out after {:?} \
315 (limit {:?}).", elapsed, budget;
316 IO, Network, Input, Timeout));
317 }
318 let remaining = budget - elapsed;
319 let result = tokio::time::timeout(
320 remaining,
321 stream.as_mut().read(&mut buf),
322 ).await;
323 match result {
324 Ok(Ok(bytes_read)) => bytes_read,
325 Ok(Err(e)) if e.kind() == tokio::io::ErrorKind::UnexpectedEof => {
326 warn!("UnexpectedEof treated as connection closure.");
327 return Ok(None);
328 },
329 Ok(Err(e)) => return Err(e.into()),
330 Err(_) => return Err(err!(
331 "HTTP header read timed out after {:?} with \
332 {} bytes accumulated so far.",
333 budget, header_bytes.len();
334 IO, Network, Input, Timeout)),
335 }
336 },
337 // No deadline configured.
338 None => match stream.as_mut().read(&mut buf).await {
339 Ok(bytes_read) => bytes_read,
340 Err(e) if e.kind() == tokio::io::ErrorKind::UnexpectedEof => {
341 warn!("UnexpectedEof treated as connection closure.");
342 return Ok(None);
343 },
344 Err(e) => return Err(e.into()),
345 },
346 };
347
348 // The peer closed before completing its header block.
349 if bytes_read == 0 {
350 return Ok(None);
351 }
352 trace!("Successfully read {} bytes into buf of size {}", bytes_read, CHUNK_SIZE);
353
354 if let Some(lim) = max_header_bytes {
355 if header_bytes.len().saturating_add(bytes_read) > lim {
356 return Err(err!(
357 "HTTP header exceeds configured limit of \
358 {} bytes.", lim;
359 IO, Network, Input, TooBig));
360 }
361 }
362
363 // A header block larger than one chunk arrives over several reads,
364 // and its terminator can straddle the join between any two of them.
365 // The search therefore runs over the accumulated bytes, resuming
366 // three back from the join so a `CRLF CRLF` split 1/3, 2/2 or 3/1 is
367 // still seen. Searching only the latest read, as this once did, lost
368 // every such message in silence.
369 let scan_from = header_bytes.len().saturating_sub(3);
370 header_bytes.extend_from_slice(&buf[..bytes_read]);
371
372 if let Some(pos) = Self::find_terminator(&header_bytes, scan_from) {
373 trace!("FOUND at pos = {}", pos);
374 let rest = header_bytes[pos + 4..].to_vec();
375 header_bytes.truncate(pos);
376 let (header_str, content_length) = res!(Self::parse_header_str(&header_bytes));
377 return Ok(Some((
378 res!(Self::parse(header_str, is_request)),
379 rest,
380 content_length,
381 )));
382 }
383 }
384 }
385
386 #[cfg(feature = "async")]
387 fn parse_header_str(header_bytes: &[u8]) -> Outcome<(String, usize)> {
388 let header_str = match std::str::from_utf8(header_bytes) {
389 Ok(s) => s.to_string(),
390 Err(e) => return Err(err!(e,
391 "Invalid UTF-8 sequence in header bytes.";
392 IO, Network, Invalid, Input)),
393 };
394
395 let header = res!(Self::parse(header_str.clone(), None));
396
397 let content_length = match header.fields.get_one(&HeaderName::ContentLength) {
398 Some(HeaderFieldValue::ContentLength(n)) => *n,
399 _ => 0,
400 };
401
402 Ok((header_str, content_length))
403 }
404
405
406 pub fn parse(
407 header_str: String,
408 is_request: Option<bool>,
409 )
410 -> Outcome<Self>
411 {
412 // Parse the headline.
413 let mut header = Self::default();
414 let mut lines = header_str.lines();
415 let (headline, version) = match lines.next() {
416 Some(line) => res!(HttpHeadline::parse(line, is_request)),
417 None => return Err(err!(
418 "HTTP request missing headline.";
419 IO, Network, Missing, Input)),
420 };
421
422 header.version = version;
423 header.headline = headline;
424
425 // Parse the fields.
426 let mut i: u16 = 1;
427 let mut ml: u8 = 1;
428 let mut current_header = String::new();
429 for line in lines {
430
431 // Accommodate multi-line headers.
432 let is_a_continuation = line.starts_with(' ') || line.starts_with('\t');
433 if ml == 1 || is_a_continuation {
434 current_header.push_str(line.trim_start());
435 ml += 1;
436 if ml < constant::HTTP_HEADER_MAX_MULTILINES {
437 if is_a_continuation {
438 continue;
439 }
440 } else {
441 return Err(err!(
442 "The HTTP header '{}' has stretched across {} lines, exceeding \
443 the limit for this server.", current_header,
444 constant::HTTP_HEADER_MAX_MULTILINES;
445 IO, Network, Invalid, Input));
446 }
447 }
448
449 if !current_header.is_empty() {
450 // Check for exceeding max fields
451 if i > constant::HTTP_HEADER_MAX_FIELDS {
452 return Err(err!(
453 "Number of header fields exceeds limit of {}.",
454 constant::HTTP_HEADER_MAX_FIELDS;
455 IO, Network, Invalid, Input));
456 }
457 let hf = res!(HeaderField::new(&current_header, Some(i)));
458 header.fields.insert(hf.name, hf.value, Some(i));
459 i += 1;
460 ml = 1;
461 current_header = String::new();
462 }
463 }
464
465 Ok(header)
466 }
467
468 pub fn get_a_field_value(&self, nam: &HeaderName) -> Option<&HeaderFieldValue> {
469 self.fields.get_one(&nam)
470 }
471
472 pub fn get_the_field_value(&self, nam: &HeaderName) -> Outcome<&HeaderFieldValue> {
473 self.fields.get_the_one(&nam)
474 }
475}
476
477
478#[cfg(all(test, feature = "async"))]
479mod reader_tests {
480 use super::*;
481
482 use std::{
483 collections::VecDeque,
484 io,
485 task::{
486 Context,
487 Poll,
488 },
489 };
490
491 use tokio::io::ReadBuf;
492
493 /// A reader that hands back the chunks it was given, one per read, so a test
494 /// can split the wire bytes wherever it likes.
495 ///
496 /// A `Cursor` returns everything it holds in a single read, and so cannot
497 /// present a header that arrives in pieces. That is exactly how the
498 /// terminator bug survived: every test fed the reader a stream that could not
499 /// exhibit it.
500 struct Scripted {
501 chunks: VecDeque<Vec<u8>>,
502 }
503
504 impl Scripted {
505 fn new(chunks: Vec<Vec<u8>>) -> Self {
506 Self { chunks: chunks.into() }
507 }
508
509 /// Dribble the bytes out `n` at a time.
510 fn dribble(bytes: &[u8], n: usize) -> Self {
511 Self::new(bytes.chunks(n).map(|c| c.to_vec()).collect())
512 }
513 }
514
515 impl AsyncRead for Scripted {
516 fn poll_read(
517 mut self: Pin<&mut Self>,
518 _cx: &mut Context<'_>,
519 buf: &mut ReadBuf<'_>,
520 )
521 -> Poll<io::Result<()>>
522 {
523 match self.chunks.pop_front() {
524 Some(chunk) => {
525 let n = std::cmp::min(chunk.len(), buf.remaining());
526 buf.put_slice(&chunk[..n]);
527 if n < chunk.len() {
528 self.chunks.push_front(chunk[n..].to_vec());
529 }
530 Poll::Ready(Ok(()))
531 },
532 // Nothing left: a read of zero bytes, as from a closed peer.
533 None => Poll::Ready(Ok(())),
534 }
535 }
536 }
537
538 fn read_header(
539 mut stream: Scripted,
540 is_request: Option<bool>,
541 )
542 -> Outcome<Option<(HttpHeader, Vec<u8>, usize)>>
543 {
544 let rt = res!(tokio::runtime::Runtime::new());
545 rt.block_on(HttpHeader::read::<{ constant::HTTP_DEFAULT_HEADER_CHUNK_SIZE }, _>(
546 Pin::new(&mut stream),
547 &Vec::new(),
548 is_request,
549 None,
550 ))
551 }
552
553 /// The `CRLF CRLF` that ends a header block can straddle the join between two
554 /// reads. Searching only the latest read never sees it, and the message is
555 /// then lost in silence.
556 #[test]
557 fn test_a_terminator_split_between_two_reads_is_still_found() -> Outcome<()> {
558 let wire: &[u8] = b"GET /a HTTP/1.1\r\nhost: example.test\r\n\r\n";
559 let term = wire.len() - 4; // Where the terminator begins.
560 for split in 1..=3 { // Split 1/3, then 2/2, then 3/1.
561 let at = term + split;
562 let stream = Scripted::new(vec![
563 wire[..at].to_vec(),
564 wire[at..].to_vec(),
565 ]);
566 let (header, remnant, content_length) = match res!(read_header(stream, Some(true))) {
567 Some(triple) => triple,
568 None => return Err(err!(
569 "A header whose terminator was split {}/{} across two reads \
570 was read as no message at all.", split, 4 - split;
571 Test, Missing)),
572 };
573 assert_eq!(header.as_vec(), wire.to_vec());
574 assert!(remnant.is_empty());
575 assert_eq!(content_length, 0);
576 }
577 Ok(())
578 }
579
580 /// A header block bigger than one read has to be assembled across several.
581 /// `HTTP_DEFAULT_HEADER_CHUNK_SIZE` is 1500, so this one cannot arrive whole
582 /// however the stream behaves.
583 #[test]
584 fn test_a_header_larger_than_one_read_is_assembled() -> Outcome<()> {
585 let pad = "p".repeat(2_000);
586 let wire = fmt!(
587 "GET /a HTTP/1.1\r\nhost: example.test\r\nx-pad: {}\r\n\r\nBODY", pad);
588 let expect = fmt!(
589 "GET /a HTTP/1.1\r\nhost: example.test\r\nx-pad: {}\r\n\r\n", pad);
590 for n in 1..=3 { // One, two and three bytes at a time.
591 let stream = Scripted::dribble(wire.as_bytes(), n);
592 let (header, remnant, _) = match res!(read_header(stream, Some(true))) {
593 Some(triple) => triple,
594 None => return Err(err!(
595 "A {} byte header arriving {} bytes at a time was read as no \
596 message at all.", wire.len(), n;
597 Test, Missing)),
598 };
599 assert_eq!(header.as_vec(), expect.clone().into_bytes());
600 // Only the body bytes that shared the terminator's read are in hand.
601 assert!(b"BODY".starts_with(&remnant[..]));
602 }
603 Ok(())
604 }
605}