/rust/git/checkouts/micro-http-22be4cdcbef12607/ef43cef/src/connection.rs
Line | Count | Source (jump to first uncovered line) |
1 | | // Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved. |
2 | | // SPDX-License-Identifier: Apache-2.0 |
3 | | |
4 | | use std::collections::VecDeque; |
5 | | use std::fs::File; |
6 | | use std::io::{Read, Write}; |
7 | | use std::os::unix::io::FromRawFd; |
8 | | |
9 | | use crate::common::ascii::{CR, CRLF_LEN, LF}; |
10 | | use crate::common::Body; |
11 | | pub use crate::common::{ConnectionError, HttpHeaderError, RequestError}; |
12 | | use crate::headers::Headers; |
13 | | use crate::request::{find, Request, RequestLine}; |
14 | | use crate::response::{Response, StatusCode}; |
15 | | use crate::server::MAX_PAYLOAD_SIZE; |
16 | | use vmm_sys_util::sock_ctrl_msg::ScmSocket; |
17 | | |
18 | | const BUFFER_SIZE: usize = 1024; |
19 | | const SCM_MAX_FD: usize = 253; |
20 | | |
21 | | /// Describes the state machine of an HTTP connection. |
22 | | enum ConnectionState { |
23 | | WaitingForRequestLine, |
24 | | WaitingForHeaders, |
25 | | WaitingForBody, |
26 | | RequestReady, |
27 | | } |
28 | | |
29 | | /// A wrapper over a HTTP Connection. |
30 | | pub struct HttpConnection<T> { |
31 | | /// A partial request that is still being received. |
32 | | pending_request: Option<Request>, |
33 | | /// Stream implementing `Read` and `Write`, capable of sending and |
34 | | /// receiving bytes. |
35 | | stream: T, |
36 | | /// The state of the connection regarding the current request that |
37 | | /// is being processed. |
38 | | state: ConnectionState, |
39 | | /// Buffer where we store the bytes we read from the stream. |
40 | | buffer: [u8; BUFFER_SIZE], |
41 | | /// The index in the buffer from where we have to start reading in |
42 | | /// the next `try_read` call. |
43 | | read_cursor: usize, |
44 | | /// Contains all bytes pertaining to the body of the request that |
45 | | /// is currently being processed. |
46 | | body_vec: Vec<u8>, |
47 | | /// Represents how many bytes from the body of the request are still |
48 | | /// to be read. |
49 | | body_bytes_to_be_read: u32, |
50 | | /// A queue of all requests that have been fully received and parsed. |
51 | | parsed_requests: VecDeque<Request>, |
52 | | /// A queue of requests that are waiting to be sent. |
53 | | response_queue: VecDeque<Response>, |
54 | | /// A buffer containing the bytes of a response that is currently |
55 | | /// being sent. |
56 | | response_buffer: Option<Vec<u8>>, |
57 | | /// The list of files that has been received and which must be associated |
58 | | /// with the pending request. |
59 | | files: Vec<File>, |
60 | | /// Optional payload max size. |
61 | | payload_max_size: usize, |
62 | | } |
63 | | |
64 | | impl<T: Read + Write + ScmSocket> HttpConnection<T> { |
65 | | /// Creates an empty connection. |
66 | 0 | pub fn new(stream: T) -> Self { |
67 | 0 | Self { |
68 | 0 | pending_request: None, |
69 | 0 | stream, |
70 | 0 | state: ConnectionState::WaitingForRequestLine, |
71 | 0 | buffer: [0; BUFFER_SIZE], |
72 | 0 | read_cursor: 0, |
73 | 0 | body_vec: vec![], |
74 | 0 | body_bytes_to_be_read: 0, |
75 | 0 | parsed_requests: VecDeque::new(), |
76 | 0 | response_queue: VecDeque::new(), |
77 | 0 | response_buffer: None, |
78 | 0 | files: Vec::new(), |
79 | 0 | payload_max_size: MAX_PAYLOAD_SIZE, |
80 | 0 | } |
81 | 0 | } |
82 | | |
83 | | /// This function sets the limit for PUT/PATCH requests. It overwrites the |
84 | | /// default limit of 0.05MiB with the one allowed by server. |
85 | 0 | pub fn set_payload_max_size(&mut self, request_payload_max_size: usize) { |
86 | 0 | self.payload_max_size = request_payload_max_size; |
87 | 0 | } |
88 | | |
89 | | /// Tries to read new bytes from the stream and automatically update the request. |
90 | | /// Meant to be used only with non-blocking streams and an `EPOLL` structure. |
91 | | /// Should be called whenever an `EPOLLIN` event is signaled. |
92 | | /// |
93 | | /// # Errors |
94 | | /// `StreamError` is returned when an IO operation fails. |
95 | | /// `ConnectionClosed` is returned when a client prematurely closes the connection. |
96 | | /// `ParseError` is returned when a parsing operation fails. |
97 | 0 | pub fn try_read(&mut self) -> Result<(), ConnectionError> { |
98 | | // Read some bytes from the stream, which will be appended to what is already |
99 | | // present in the buffer from a previous call of `try_read`. There are already |
100 | | // `read_cursor` bytes present in the buffer. |
101 | 0 | let end_cursor = self.read_bytes()?; |
102 | | |
103 | 0 | let mut line_start_index = 0; |
104 | | loop { |
105 | 0 | match self.state { |
106 | | ConnectionState::WaitingForRequestLine => { |
107 | 0 | if !self.parse_request_line(&mut line_start_index, end_cursor)? { |
108 | 0 | return Ok(()); |
109 | 0 | } |
110 | | } |
111 | | ConnectionState::WaitingForHeaders => { |
112 | 0 | if !self.parse_headers(&mut line_start_index, end_cursor)? { |
113 | 0 | return Ok(()); |
114 | 0 | } |
115 | | } |
116 | | ConnectionState::WaitingForBody => { |
117 | 0 | if !self.parse_body(&mut line_start_index, end_cursor)? { |
118 | 0 | return Ok(()); |
119 | 0 | } |
120 | | } |
121 | 0 | ConnectionState::RequestReady => { |
122 | 0 | // This request is ready to be passed for handling. |
123 | 0 | // Update the state machine to expect a new request and push this request into |
124 | 0 | // the `parsed_requests` queue. |
125 | 0 | self.state = ConnectionState::WaitingForRequestLine; |
126 | 0 | self.body_bytes_to_be_read = 0; |
127 | 0 | let mut pending_request = self.pending_request.take().unwrap(); |
128 | 0 | pending_request.files = self.files.drain(..).collect(); |
129 | 0 | self.parsed_requests.push_back(pending_request); |
130 | 0 | } |
131 | | }; |
132 | | } |
133 | 0 | } |
134 | | |
135 | | /// Reads a maximum of 1024 bytes from the stream into `buffer`. |
136 | | /// The return value represents the end index of what we have just appended. |
137 | | /// |
138 | | /// # Errors |
139 | | /// `StreamError` is returned if any error occurred while reading the stream. |
140 | | /// `ConnectionClosed` is returned if the client closed the connection. |
141 | | /// `Overflow` is returned if an arithmetic overflow occurs while parsing the request. |
142 | 0 | fn read_bytes(&mut self) -> Result<usize, ConnectionError> { |
143 | 0 | if self.read_cursor >= BUFFER_SIZE { |
144 | 0 | return Err(ConnectionError::ParseError(RequestError::Overflow)); |
145 | 0 | } |
146 | | // Append new bytes to what we already have in the buffer. |
147 | | // The slice access is safe, the index is checked above. |
148 | 0 | let (bytes_read, new_files) = self.recv_with_fds()?; |
149 | | |
150 | | // Update the internal list of files that must be associated with the |
151 | | // request. |
152 | 0 | self.files.extend(new_files); |
153 | 0 |
|
154 | 0 | // If the read returned 0 then the client has closed the connection. |
155 | 0 | if bytes_read == 0 { |
156 | 0 | return Err(ConnectionError::ConnectionClosed); |
157 | 0 | } |
158 | 0 | bytes_read |
159 | 0 | .checked_add(self.read_cursor) |
160 | 0 | .ok_or(ConnectionError::ParseError(RequestError::Overflow)) |
161 | 0 | } |
162 | | |
163 | | /// Receive data along with optional files descriptors. |
164 | | /// It is a wrapper around the same function from vmm-sys-util. |
165 | | /// |
166 | | /// # Errors |
167 | | /// `StreamError` is returned if any error occurred while reading the stream. |
168 | 0 | fn recv_with_fds(&mut self) -> Result<(usize, Vec<File>), ConnectionError> { |
169 | 0 | let buf = &mut self.buffer[self.read_cursor..]; |
170 | 0 | // We must allocate the maximum number of receivable file descriptors |
171 | 0 | // if don't want to miss any of them. Allocating a too small number |
172 | 0 | // would lead to the incapacity of receiving the file descriptors. |
173 | 0 | let mut fds = [0; SCM_MAX_FD]; |
174 | 0 | let mut iovecs = [libc::iovec { |
175 | 0 | iov_base: buf.as_mut_ptr() as *mut libc::c_void, |
176 | 0 | iov_len: buf.len(), |
177 | 0 | }]; |
178 | | |
179 | | // SAFETY: Safe because we have mutably borrowed buf and it's safe to write |
180 | | // arbitrary data to a slice. |
181 | 0 | let (read_count, fd_count) = unsafe { |
182 | 0 | self.stream |
183 | 0 | .recv_with_fds(&mut iovecs, &mut fds) |
184 | 0 | .map_err(ConnectionError::StreamReadError)? |
185 | | }; |
186 | | |
187 | 0 | Ok(( |
188 | 0 | read_count, |
189 | 0 | fds.iter() |
190 | 0 | .take(fd_count) |
191 | 0 | .map(|fd| { |
192 | 0 | // SAFETY: Safe because all fds are owned by us after they have been |
193 | 0 | // received through the socket. |
194 | 0 | unsafe { File::from_raw_fd(*fd) } |
195 | 0 | }) |
196 | 0 | .collect(), |
197 | 0 | )) |
198 | 0 | } |
199 | | |
200 | | /// Parses bytes in `buffer` for a valid request line. |
201 | | /// Returns `false` if there are no more bytes to be parsed in the buffer. |
202 | | /// |
203 | | /// # Errors |
204 | | /// `ParseError` is returned if unable to parse request line or line longer than BUFFER_SIZE. |
205 | 0 | fn parse_request_line( |
206 | 0 | &mut self, |
207 | 0 | start: &mut usize, |
208 | 0 | end: usize, |
209 | 0 | ) -> Result<bool, ConnectionError> { |
210 | 0 | if end < *start { |
211 | 0 | return Err(ConnectionError::ParseError(RequestError::Underflow)); |
212 | 0 | } |
213 | 0 | if end > self.buffer.len() { |
214 | 0 | return Err(ConnectionError::ParseError(RequestError::Overflow)); |
215 | 0 | } |
216 | 0 | // The slice access is safe because `end` is checked to be smaller than the buffer size |
217 | 0 | // and larger than `start`. |
218 | 0 | match find(&self.buffer[*start..end], &[CR, LF]) { |
219 | 0 | Some(line_end_index) => { |
220 | 0 | // The unchecked addition `start + line_end_index` is safe because `line_end_index` |
221 | 0 | // is returned by `find` and thus guaranteed to be in-bounds. This also makes the |
222 | 0 | // slice access safe. |
223 | 0 | let line = &self.buffer[*start..(*start + line_end_index)]; |
224 | 0 |
|
225 | 0 | // The unchecked addition is safe because of the previous `find()`. |
226 | 0 | *start = *start + line_end_index + CRLF_LEN; |
227 | | |
228 | | // Form the request with a valid request line, which is the bare minimum |
229 | | // for a valid request. |
230 | 0 | self.pending_request = Some(Request { |
231 | 0 | request_line: RequestLine::try_from(line) |
232 | 0 | .map_err(ConnectionError::ParseError)?, |
233 | 0 | headers: Headers::default(), |
234 | 0 | body: None, |
235 | 0 | files: Vec::new(), |
236 | 0 | }); |
237 | 0 | self.state = ConnectionState::WaitingForHeaders; |
238 | 0 | Ok(true) |
239 | | } |
240 | | None => { |
241 | | // The request line is longer than BUFFER_SIZE bytes, so the request is invalid. |
242 | 0 | if end == BUFFER_SIZE && *start == 0 { |
243 | 0 | return Err(ConnectionError::ParseError(RequestError::InvalidRequest)); |
244 | | } else { |
245 | | // Move the incomplete request line to the beginning of the buffer and wait |
246 | | // for the next `try_read` call to complete it. |
247 | | // This can only happen if another request was sent before this one, as the |
248 | | // limit for the length of a request line in this implementation is 1024 bytes. |
249 | 0 | self.shift_buffer_left(*start, end) |
250 | 0 | .map_err(ConnectionError::ParseError)?; |
251 | | } |
252 | 0 | Ok(false) |
253 | | } |
254 | | } |
255 | 0 | } |
256 | | |
257 | | /// Parses bytes in `buffer` for header fields. |
258 | | /// Returns `false` if there are no more bytes to be parsed in the buffer. |
259 | | /// |
260 | | /// # Errors |
261 | | /// `ParseError` is returned if unable to parse header or line longer than BUFFER_SIZE. |
262 | 0 | fn parse_headers( |
263 | 0 | &mut self, |
264 | 0 | line_start_index: &mut usize, |
265 | 0 | end_cursor: usize, |
266 | 0 | ) -> Result<bool, ConnectionError> { |
267 | 0 | if end_cursor > self.buffer.len() { |
268 | 0 | return Err(ConnectionError::ParseError(RequestError::Overflow)); |
269 | 0 | } |
270 | 0 | if end_cursor < *line_start_index { |
271 | 0 | return Err(ConnectionError::ParseError(RequestError::Underflow)); |
272 | 0 | } |
273 | 0 | // Safe to access the slice as the bounds are checked above. |
274 | 0 | match find(&self.buffer[*line_start_index..end_cursor], &[CR, LF]) { |
275 | | // `line_start_index` points to the end of the most recently found CR LF |
276 | | // sequence. That means that if we found the next CR LF sequence at this index, |
277 | | // they are, in fact, a CR LF CR LF sequence, which marks the end of the header |
278 | | // fields, per HTTP specification. |
279 | | |
280 | | // We have found the end of the header. |
281 | | Some(0) => { |
282 | | // The current state is `WaitingForHeaders`, ensuring a valid request formed from a |
283 | | // request line. |
284 | 0 | let request = self |
285 | 0 | .pending_request |
286 | 0 | .as_mut() |
287 | 0 | .ok_or(ConnectionError::ParseError( |
288 | 0 | RequestError::HeadersWithoutPendingRequest, |
289 | 0 | ))?; |
290 | 0 | if request.headers.content_length() == 0 { |
291 | 0 | self.state = ConnectionState::RequestReady; |
292 | 0 | } else { |
293 | 0 | if request.headers.content_length() as usize > self.payload_max_size { |
294 | 0 | return Err(ConnectionError::ParseError( |
295 | 0 | RequestError::SizeLimitExceeded( |
296 | 0 | self.payload_max_size, |
297 | 0 | request.headers.content_length() as usize, |
298 | 0 | ), |
299 | 0 | )); |
300 | 0 | } |
301 | 0 | if request.headers.expect() { |
302 | 0 | // Send expect. |
303 | 0 | let expect_response = |
304 | 0 | Response::new(request.http_version(), StatusCode::Continue); |
305 | 0 | self.response_queue.push_back(expect_response); |
306 | 0 | } |
307 | | |
308 | 0 | self.body_bytes_to_be_read = request.headers.content_length(); |
309 | 0 | request.body = Some(Body::new(vec![])); |
310 | 0 | self.state = ConnectionState::WaitingForBody; |
311 | | } |
312 | | |
313 | | // Update the index for the next header. |
314 | 0 | *line_start_index = line_start_index |
315 | 0 | .checked_add(CRLF_LEN) |
316 | 0 | .ok_or(ConnectionError::ParseError(RequestError::Overflow))?; |
317 | 0 | Ok(true) |
318 | | } |
319 | | // We have found the end of a header line. |
320 | 0 | Some(relative_line_end_index) => { |
321 | 0 | let request = self |
322 | 0 | .pending_request |
323 | 0 | .as_mut() |
324 | 0 | .ok_or(ConnectionError::ParseError( |
325 | 0 | RequestError::HeadersWithoutPendingRequest, |
326 | 0 | ))?; |
327 | | // The `line_end_index` relative to the whole buffer. |
328 | 0 | let line_end_index = relative_line_end_index |
329 | 0 | .checked_add(*line_start_index) |
330 | 0 | .ok_or(ConnectionError::ParseError(RequestError::Overflow))?; |
331 | | |
332 | | // Get the line slice and parse it. |
333 | | // The slice access is safe because `line_end_index` is a sum of `line_end_index` |
334 | | // and something else, and `line_end_index` itself is guaranteed to be within |
335 | | // `self.buffer`'s bounds by the `find()`. |
336 | 0 | let line = &self.buffer[*line_start_index..line_end_index]; |
337 | 0 | match request.headers.parse_header_line(line) { |
338 | | // If a header is unsupported we ignore it. |
339 | | Ok(_) |
340 | 0 | | Err(RequestError::HeaderError(HttpHeaderError::UnsupportedValue(_, _))) => {} |
341 | | // If parsing the header invalidates the request, we propagate |
342 | | // the error. |
343 | 0 | Err(e) => return Err(ConnectionError::ParseError(e)), |
344 | | }; |
345 | | |
346 | | // Update the `line_start_index` to where we finished parsing. |
347 | 0 | *line_start_index = line_end_index |
348 | 0 | .checked_add(CRLF_LEN) |
349 | 0 | .ok_or(ConnectionError::ParseError(RequestError::Overflow))?; |
350 | 0 | Ok(true) |
351 | | } |
352 | | // If we have an incomplete header line. |
353 | | None => { |
354 | | // If we have parsed BUFFER_SIZE bytes and still haven't found the header |
355 | | // line end sequence. |
356 | 0 | if *line_start_index == 0 && end_cursor == BUFFER_SIZE { |
357 | | // Header line is longer than BUFFER_SIZE bytes, so it is invalid. |
358 | 0 | let utf8_string = String::from_utf8_lossy(&self.buffer); |
359 | 0 | return Err(ConnectionError::ParseError(RequestError::HeaderError( |
360 | 0 | HttpHeaderError::SizeLimitExceeded(utf8_string.to_string()), |
361 | 0 | ))); |
362 | 0 | } |
363 | 0 | // Move the incomplete header line from the end of the buffer to |
364 | 0 | // the beginning, so that we can append the rest of the line and |
365 | 0 | // parse it in the next `try_read` call. |
366 | 0 | self.shift_buffer_left(*line_start_index, end_cursor) |
367 | 0 | .map_err(ConnectionError::ParseError)?; |
368 | 0 | Ok(false) |
369 | | } |
370 | | } |
371 | 0 | } |
372 | | |
373 | | /// Parses bytes in `buffer` to be put into the request body, if there should be one. |
374 | | /// Returns `false` if there are no more bytes to be parsed in the buffer. |
375 | | /// |
376 | | /// # Errors |
377 | | /// `ParseError` is returned when the body is larger than the specified content-length. |
378 | 0 | fn parse_body( |
379 | 0 | &mut self, |
380 | 0 | line_start_index: &mut usize, |
381 | 0 | end_cursor: usize, |
382 | 0 | ) -> Result<bool, ConnectionError> { |
383 | 0 | // If what we have just read is not enough to complete the request and |
384 | 0 | // there are more bytes pertaining to the body of the request. |
385 | 0 | if end_cursor > self.buffer.len() { |
386 | 0 | return Err(ConnectionError::ParseError(RequestError::Overflow)); |
387 | 0 | } |
388 | 0 | let start_to_end = end_cursor |
389 | 0 | .checked_sub(*line_start_index) |
390 | 0 | .ok_or(ConnectionError::ParseError(RequestError::Underflow))? |
391 | | as u32; |
392 | 0 | if self.body_bytes_to_be_read > start_to_end { |
393 | | // Append everything that we read to our current incomplete body and update |
394 | | // `body_bytes_to_be_read`. |
395 | | // The slice access is safe, otherwise `checked_sub` would have failed. |
396 | 0 | self.body_vec |
397 | 0 | .extend_from_slice(&self.buffer[*line_start_index..end_cursor]); |
398 | 0 | // Safe to subtract directly as the `if` condition prevents underflow. |
399 | 0 | self.body_bytes_to_be_read -= start_to_end; |
400 | | |
401 | | // Clear the buffer and reset the starting index. |
402 | 0 | for i in 0..BUFFER_SIZE { |
403 | 0 | self.buffer[i] = 0; |
404 | 0 | } |
405 | 0 | self.read_cursor = 0; |
406 | 0 |
|
407 | 0 | return Ok(false); |
408 | 0 | } |
409 | | |
410 | | // Append only the remaining necessary bytes to the body of the request. |
411 | 0 | let line_end = line_start_index |
412 | 0 | .checked_add(self.body_bytes_to_be_read as usize) |
413 | 0 | .ok_or(ConnectionError::ParseError(RequestError::Overflow))?; |
414 | | // The slice access is safe as `line_end` is a sum of `line_start_index` + something else. |
415 | 0 | self.body_vec |
416 | 0 | .extend_from_slice(&self.buffer[*line_start_index..line_end]); |
417 | 0 | *line_start_index = line_end; |
418 | 0 | self.body_bytes_to_be_read = 0; |
419 | | |
420 | 0 | let request = self |
421 | 0 | .pending_request |
422 | 0 | .as_mut() |
423 | 0 | .ok_or(ConnectionError::ParseError( |
424 | 0 | RequestError::BodyWithoutPendingRequest, |
425 | 0 | ))?; |
426 | | // If there are no more bytes to be read for this request. |
427 | | // Assign the body of the request. |
428 | 0 | let placeholder: Vec<_> = self |
429 | 0 | .body_vec |
430 | 0 | .drain(..request.headers.content_length() as usize) |
431 | 0 | .collect(); |
432 | 0 | request.body = Some(Body::new(placeholder)); |
433 | 0 |
|
434 | 0 | // If we read more bytes than we should have into the body of the request. |
435 | 0 | if !self.body_vec.is_empty() { |
436 | 0 | return Err(ConnectionError::ParseError(RequestError::InvalidRequest)); |
437 | 0 | } |
438 | 0 |
|
439 | 0 | self.state = ConnectionState::RequestReady; |
440 | 0 | Ok(true) |
441 | 0 | } |
442 | | |
443 | | /// Tries to write the first available response to the provided stream. |
444 | | /// Meant to be used only with non-blocking streams and an `EPOLL` structure. |
445 | | /// Should be called whenever an `EPOLLOUT` event is signaled. If no bytes |
446 | | /// were written to the stream or error occurred while trying to write to stream, |
447 | | /// we will discard all responses from response_queue because there is no way |
448 | | /// to deliver it to client. |
449 | | /// |
450 | | /// # Errors |
451 | | /// `StreamError` is returned when an IO operation fails. |
452 | | /// `ConnectionClosed` is returned when trying to write on a closed connection. |
453 | | /// `InvalidWrite` is returned when trying to write on a connection with an |
454 | | /// empty outgoing buffer. |
455 | 0 | pub fn try_write(&mut self) -> Result<(), ConnectionError> { |
456 | 0 | if self.response_buffer.is_none() { |
457 | 0 | if let Some(response) = self.response_queue.pop_front() { |
458 | 0 | let mut response_buffer_vec: Vec<u8> = Vec::new(); |
459 | 0 | response |
460 | 0 | .write_all(&mut response_buffer_vec) |
461 | 0 | .map_err(ConnectionError::StreamWriteError)?; |
462 | 0 | self.response_buffer = Some(response_buffer_vec); |
463 | | } else { |
464 | 0 | return Err(ConnectionError::InvalidWrite); |
465 | | } |
466 | 0 | } |
467 | | |
468 | 0 | let mut response_fully_written = false; |
469 | 0 | let mut connection_closed = false; |
470 | | |
471 | 0 | if let Some(response_buffer_vec) = self.response_buffer.as_mut() { |
472 | 0 | let bytes_to_be_written = response_buffer_vec.len(); |
473 | 0 | match self.stream.write(response_buffer_vec.as_slice()) { |
474 | 0 | Ok(0) => connection_closed = true, |
475 | 0 | Ok(bytes_written) => { |
476 | 0 | if bytes_written != bytes_to_be_written { |
477 | 0 | response_buffer_vec.drain(..bytes_written); |
478 | 0 | } else { |
479 | 0 | response_fully_written = true; |
480 | 0 | } |
481 | | } |
482 | 0 | Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {} |
483 | 0 | Err(_) => connection_closed = true, |
484 | | } |
485 | 0 | } |
486 | | |
487 | 0 | if connection_closed { |
488 | 0 | self.clear_write_buffer(); |
489 | 0 | return Err(ConnectionError::ConnectionClosed); |
490 | 0 | } else if response_fully_written { |
491 | 0 | self.response_buffer.take(); |
492 | 0 | } |
493 | | |
494 | 0 | Ok(()) |
495 | 0 | } |
496 | | |
497 | | /// Discards all pending writes from the connection. |
498 | 0 | pub fn clear_write_buffer(&mut self) { |
499 | 0 | self.response_queue.clear(); |
500 | 0 | self.response_buffer.take(); |
501 | 0 | } |
502 | | |
503 | | /// Send a response back to the source of a request. |
504 | 0 | pub fn enqueue_response(&mut self, response: Response) { |
505 | 0 | self.response_queue.push_back(response); |
506 | 0 | } |
507 | | |
508 | 0 | fn shift_buffer_left( |
509 | 0 | &mut self, |
510 | 0 | line_start_index: usize, |
511 | 0 | end_cursor: usize, |
512 | 0 | ) -> Result<(), RequestError> { |
513 | 0 | if end_cursor > self.buffer.len() { |
514 | 0 | return Err(RequestError::Overflow); |
515 | 0 | } |
516 | | // We don't want to shift something that is already at the beginning. |
517 | 0 | let delta_bytes = end_cursor |
518 | 0 | .checked_sub(line_start_index) |
519 | 0 | .ok_or(RequestError::Underflow)?; |
520 | 0 | if line_start_index != 0 { |
521 | | // Move the bytes from `line_start_index` to the beginning of the buffer. |
522 | 0 | for cursor in 0..delta_bytes { |
523 | 0 | // The unchecked addition is safe, guaranteed by the result of the substraction |
524 | 0 | // above. |
525 | 0 | // The slice access is safe, as `line_start_index + cursor` is <= `end_cursor`, |
526 | 0 | // checked at the start of the function. |
527 | 0 | self.buffer[cursor] = self.buffer[line_start_index + cursor]; |
528 | 0 | } |
529 | | |
530 | | // Clear the rest of the buffer. |
531 | 0 | for cursor in delta_bytes..end_cursor { |
532 | 0 | self.buffer[cursor] = 0; |
533 | 0 | } |
534 | 0 | } |
535 | | |
536 | | // Update `read_cursor`. |
537 | 0 | self.read_cursor = delta_bytes; |
538 | 0 | Ok(()) |
539 | 0 | } |
540 | | |
541 | | /// Returns the first parsed request in the queue or `None` if the queue |
542 | | /// is empty. |
543 | 0 | pub fn pop_parsed_request(&mut self) -> Option<Request> { |
544 | 0 | self.parsed_requests.pop_front() |
545 | 0 | } |
546 | | |
547 | | /// Returns `true` if there are bytes waiting to be written into the stream. |
548 | 0 | pub fn pending_write(&self) -> bool { |
549 | 0 | self.response_buffer.is_some() || !self.response_queue.is_empty() |
550 | 0 | } |
551 | | } |
552 | | |
553 | | #[cfg(test)] |
554 | | mod tests { |
555 | | use std::io::{Seek, SeekFrom}; |
556 | | use std::net::Shutdown; |
557 | | use std::os::unix::io::IntoRawFd; |
558 | | use std::os::unix::net::UnixStream; |
559 | | |
560 | | use super::*; |
561 | | use crate::common::{Method, Version}; |
562 | | use crate::server::MAX_PAYLOAD_SIZE; |
563 | | |
564 | | use vmm_sys_util::tempfile::TempFile; |
565 | | |
566 | | #[test] |
567 | | fn test_try_read_expect() { |
568 | | // Test request with `Expect` header. |
569 | | let (mut sender, receiver) = UnixStream::pair().unwrap(); |
570 | | receiver.set_nonblocking(true).expect("Can't modify socket"); |
571 | | let mut conn = HttpConnection::new(receiver); |
572 | | sender |
573 | | .write_all( |
574 | | b"PATCH http://localhost/home HTTP/1.1\r\n\ |
575 | | Expect: 100-continue\r\n\ |
576 | | Content-Length: 26\r\n\ |
577 | | Transfer-Encoding: chunked\r\n\r\n", |
578 | | ) |
579 | | .unwrap(); |
580 | | assert!(conn.try_read().is_ok()); |
581 | | |
582 | | sender.write_all(b"this is not\n\r\na json \nbody").unwrap(); |
583 | | conn.try_read().unwrap(); |
584 | | let request = conn.pop_parsed_request().unwrap(); |
585 | | |
586 | | let expected_request = Request { |
587 | | request_line: RequestLine::new(Method::Patch, "http://localhost/home", Version::Http11), |
588 | | headers: Headers::new(26, true, true), |
589 | | body: Some(Body::new(b"this is not\n\r\na json \nbody".to_vec())), |
590 | | files: Vec::new(), |
591 | | }; |
592 | | |
593 | | assert_eq!(request, expected_request); |
594 | | } |
595 | | |
596 | | #[test] |
597 | | fn test_try_read_long_headers() { |
598 | | // Long request headers. |
599 | | let (mut sender, receiver) = UnixStream::pair().unwrap(); |
600 | | receiver.set_nonblocking(true).expect("Can't modify socket"); |
601 | | let mut conn = HttpConnection::new(receiver); |
602 | | sender |
603 | | .write_all( |
604 | | b"PATCH http://localhost/home HTTP/1.1\r\n\ |
605 | | Expect: 100-continue\r\n\ |
606 | | Transfer-Encoding: chunked\r\n", |
607 | | ) |
608 | | .unwrap(); |
609 | | |
610 | | for i in 0..90 { |
611 | | sender.write_all(b"Custom-Header-Testing: 1").unwrap(); |
612 | | sender.write_all(i.to_string().as_bytes()).unwrap(); |
613 | | sender.write_all(b"\r\n").unwrap(); |
614 | | } |
615 | | sender |
616 | | .write_all(b"Content-Length: 26\r\n\r\nthis is not\n\r\na json \nbody") |
617 | | .unwrap(); |
618 | | assert!(conn.try_read().is_ok()); |
619 | | assert!(conn.try_read().is_ok()); |
620 | | assert!(conn.try_read().is_ok()); |
621 | | let request = conn.pop_parsed_request().unwrap(); |
622 | | |
623 | | let expected_request = Request { |
624 | | request_line: RequestLine::new(Method::Patch, "http://localhost/home", Version::Http11), |
625 | | headers: Headers::new(26, true, true), |
626 | | body: Some(Body::new(b"this is not\n\r\na json \nbody".to_vec())), |
627 | | files: Vec::new(), |
628 | | }; |
629 | | assert_eq!(request, expected_request); |
630 | | } |
631 | | |
632 | | #[test] |
633 | | fn test_try_read_split_ending() { |
634 | | // Long request with '\r\n' on BUFFER_SIZEth and 1025th positions in the request. |
635 | | let (mut sender, receiver) = UnixStream::pair().unwrap(); |
636 | | receiver.set_nonblocking(true).expect("Can't modify socket"); |
637 | | let mut conn = HttpConnection::new(receiver); |
638 | | sender |
639 | | .write_all( |
640 | | b"PATCH http://localhost/home HTTP/1.1\r\n\ |
641 | | Expect: 100-continue\r\n\ |
642 | | Transfer-Encoding: chunked\r\n", |
643 | | ) |
644 | | .unwrap(); |
645 | | |
646 | | for i in 0..32 { |
647 | | sender.write_all(b"Custom-Header-Testing: 1").unwrap(); |
648 | | sender.write_all(i.to_string().as_bytes()).unwrap(); |
649 | | sender.write_all(b"\r\n").unwrap(); |
650 | | } |
651 | | sender |
652 | | .write_all(b"Head: aaaaa\r\nContent-Length: 26\r\n\r\nthis is not\n\r\na json \nbody") |
653 | | .unwrap(); |
654 | | assert!(conn.try_read().is_ok()); |
655 | | conn.try_read().unwrap(); |
656 | | let request = conn.pop_parsed_request().unwrap(); |
657 | | let expected_request = Request { |
658 | | request_line: RequestLine::new(Method::Patch, "http://localhost/home", Version::Http11), |
659 | | headers: Headers::new(26, true, true), |
660 | | body: Some(Body::new(b"this is not\n\r\na json \nbody".to_vec())), |
661 | | files: Vec::new(), |
662 | | }; |
663 | | assert_eq!(request, expected_request); |
664 | | } |
665 | | |
666 | | #[test] |
667 | | fn test_try_read_invalid_request() { |
668 | | // Invalid request. |
669 | | let (mut sender, receiver) = UnixStream::pair().unwrap(); |
670 | | receiver.set_nonblocking(true).expect("Can't modify socket"); |
671 | | let mut conn = HttpConnection::new(receiver); |
672 | | sender |
673 | | .write_all( |
674 | | b"PATCH http://localhost/home HTTP/1.1\r\n\ |
675 | | Expect: 100-continue\r\n\ |
676 | | Transfer-Encoding: chunked\r\n", |
677 | | ) |
678 | | .unwrap(); |
679 | | |
680 | | for i in 0..40 { |
681 | | sender.write_all(b"Custom-Header-Testing: 1").unwrap(); |
682 | | sender.write_all(i.to_string().as_bytes()).unwrap(); |
683 | | sender.write_all(b"\r\n").unwrap(); |
684 | | } |
685 | | sender |
686 | | .write_all(b"Content-Length: alpha\r\n\r\nthis is not\n\r\na json \nbody") |
687 | | .unwrap(); |
688 | | assert!(conn.try_read().is_ok()); |
689 | | let request_error = conn.try_read().unwrap_err(); |
690 | | assert_eq!( |
691 | | request_error, |
692 | | ConnectionError::ParseError(RequestError::HeaderError(HttpHeaderError::InvalidValue( |
693 | | "Content-Length".to_string(), |
694 | | " alpha".to_string() |
695 | | ))) |
696 | | ); |
697 | | } |
698 | | |
699 | | #[test] |
700 | | fn test_try_read_long_request_body() { |
701 | | // Long request body. |
702 | | let (mut sender, receiver) = UnixStream::pair().unwrap(); |
703 | | receiver.set_nonblocking(true).expect("Can't modify socket"); |
704 | | let mut conn = HttpConnection::new(receiver); |
705 | | sender |
706 | | .write_all( |
707 | | b"PATCH http://localhost/home HTTP/1.1\r\n\ |
708 | | Expect: 100-continue\r\n\ |
709 | | Transfer-Encoding: chunked\r\n\ |
710 | | Content-Length: 1400\r\n\r\n", |
711 | | ) |
712 | | .unwrap(); |
713 | | |
714 | | let mut request_body: Vec<u8> = Vec::with_capacity(1400); |
715 | | for _ in 0..100 { |
716 | | request_body.write_all(b"This is a test").unwrap(); |
717 | | } |
718 | | sender.write_all(request_body.as_slice()).unwrap(); |
719 | | assert!(conn.try_read().is_ok()); |
720 | | conn.try_read().unwrap(); |
721 | | let request = conn.pop_parsed_request().unwrap(); |
722 | | let expected_request = Request { |
723 | | request_line: RequestLine::new(Method::Patch, "http://localhost/home", Version::Http11), |
724 | | headers: Headers::new(1400, true, true), |
725 | | body: Some(Body::new(request_body)), |
726 | | files: Vec::new(), |
727 | | }; |
728 | | |
729 | | assert_eq!(request, expected_request); |
730 | | } |
731 | | |
732 | | #[test] |
733 | | fn test_try_read_large_req_line() { |
734 | | // Request line longer than BUFFER_SIZE bytes. |
735 | | let (mut sender, receiver) = UnixStream::pair().unwrap(); |
736 | | receiver.set_nonblocking(true).expect("Can't modify socket"); |
737 | | let mut conn = HttpConnection::new(receiver); |
738 | | sender.write_all(b"PATCH http://localhost/home").unwrap(); |
739 | | |
740 | | let mut request_body: Vec<u8> = Vec::with_capacity(1400); |
741 | | for _ in 0..200 { |
742 | | request_body.write_all(b"/home").unwrap(); |
743 | | } |
744 | | sender.write_all(request_body.as_slice()).unwrap(); |
745 | | assert_eq!( |
746 | | conn.try_read().unwrap_err(), |
747 | | ConnectionError::ParseError(RequestError::InvalidRequest) |
748 | | ); |
749 | | } |
750 | | |
751 | | #[test] |
752 | | fn test_try_read_large_header_line() { |
753 | | // Header line longer than BUFFER_SIZE bytes. |
754 | | let (mut sender, receiver) = UnixStream::pair().unwrap(); |
755 | | receiver.set_nonblocking(true).expect("Can't modify socket"); |
756 | | let mut conn = HttpConnection::new(receiver); |
757 | | sender |
758 | | .write_all(b"PATCH http://localhost/home HTTP/1.1\r\nhead: ") |
759 | | .unwrap(); |
760 | | |
761 | | let mut request_body: Vec<u8> = Vec::with_capacity(1030); |
762 | | for _ in 0..86 { |
763 | | request_body.write_all(b"abcdefghijkl").unwrap(); |
764 | | } |
765 | | request_body.write_all(b"\r\n\r\n").unwrap(); |
766 | | sender.write_all(request_body.as_slice()).unwrap(); |
767 | | assert!(conn.try_read().is_ok()); |
768 | | |
769 | | let expected_msg = &format!("head: {}", String::from_utf8(request_body).unwrap())[..1024]; |
770 | | assert_eq!( |
771 | | conn.try_read().unwrap_err(), |
772 | | ConnectionError::ParseError(RequestError::HeaderError( |
773 | | HttpHeaderError::SizeLimitExceeded(expected_msg.to_string()) |
774 | | )) |
775 | | ); |
776 | | } |
777 | | |
778 | | #[test] |
779 | | fn test_try_read_no_body_request() { |
780 | | // Request without body. |
781 | | let (mut sender, receiver) = UnixStream::pair().unwrap(); |
782 | | receiver.set_nonblocking(true).expect("Can't modify socket"); |
783 | | let mut conn = HttpConnection::new(receiver); |
784 | | sender |
785 | | .write_all( |
786 | | b"PATCH http://localhost/home HTTP/1.1\r\n\ |
787 | | Expect: 100-continue\r\n\ |
788 | | Transfer-Encoding: chunked\r\n\r\n", |
789 | | ) |
790 | | .unwrap(); |
791 | | conn.try_read().unwrap(); |
792 | | let request = conn.pop_parsed_request().unwrap(); |
793 | | let expected_request = Request { |
794 | | request_line: RequestLine::new(Method::Patch, "http://localhost/home", Version::Http11), |
795 | | headers: Headers::new(0, true, true), |
796 | | body: None, |
797 | | files: Vec::new(), |
798 | | }; |
799 | | assert_eq!(request, expected_request); |
800 | | } |
801 | | |
802 | | #[test] |
803 | | fn test_try_read_segmented_req_line() { |
804 | | // Segmented request line. |
805 | | let (mut sender, receiver) = UnixStream::pair().unwrap(); |
806 | | receiver.set_nonblocking(true).expect("Can't modify socket"); |
807 | | let mut conn = HttpConnection::new(receiver); |
808 | | sender.write_all(b"PATCH http://local").unwrap(); |
809 | | assert!(conn.try_read().is_ok()); |
810 | | |
811 | | sender.write_all(b"host/home HTTP/1.1\r\n\r\n").unwrap(); |
812 | | |
813 | | conn.try_read().unwrap(); |
814 | | let request = conn.pop_parsed_request().unwrap(); |
815 | | let expected_request = Request { |
816 | | request_line: RequestLine::new(Method::Patch, "http://localhost/home", Version::Http11), |
817 | | headers: Headers::new(0, false, false), |
818 | | body: None, |
819 | | files: Vec::new(), |
820 | | }; |
821 | | assert_eq!(request, expected_request); |
822 | | } |
823 | | |
824 | | #[test] |
825 | | fn test_try_read_long_req_line_b2b() { |
826 | | // Long request line after another request. |
827 | | let (mut sender, receiver) = UnixStream::pair().unwrap(); |
828 | | receiver.set_nonblocking(true).expect("Can't modify socket"); |
829 | | let mut conn = HttpConnection::new(receiver); |
830 | | // Req line 23 + 10*x + 13 = 36 + 10* x 984 free in first try read |
831 | | sender |
832 | | .write_all(b"PATCH http://localhost/home HTTP/1.1\r\n\r\nPATCH http://localhost/") |
833 | | .unwrap(); |
834 | | |
835 | | let mut request_line: Vec<u8> = Vec::with_capacity(980); |
836 | | for _ in 0..98 { |
837 | | request_line.write_all(b"localhost/").unwrap(); |
838 | | } |
839 | | request_line.write_all(b" HTTP/1.1\r\n\r\n").unwrap(); |
840 | | sender.write_all(request_line.as_slice()).unwrap(); |
841 | | |
842 | | conn.try_read().unwrap(); |
843 | | let request = conn.pop_parsed_request().unwrap(); |
844 | | let expected_request = Request { |
845 | | request_line: RequestLine::new(Method::Patch, "http://localhost/home", Version::Http11), |
846 | | headers: Headers::new(0, false, false), |
847 | | body: None, |
848 | | files: Vec::new(), |
849 | | }; |
850 | | assert_eq!(request, expected_request); |
851 | | |
852 | | conn.try_read().unwrap(); |
853 | | let request = conn.pop_parsed_request().unwrap(); |
854 | | let mut expected_request_as_bytes = Vec::new(); |
855 | | expected_request_as_bytes |
856 | | .write_all(b"http://localhost/") |
857 | | .unwrap(); |
858 | | expected_request_as_bytes.append(request_line.as_mut()); |
859 | | let expected_request = Request { |
860 | | request_line: RequestLine::new( |
861 | | Method::Patch, |
862 | | std::str::from_utf8(&expected_request_as_bytes[..997]).unwrap(), |
863 | | Version::Http11, |
864 | | ), |
865 | | headers: Headers::new(0, false, false), |
866 | | body: None, |
867 | | files: Vec::new(), |
868 | | }; |
869 | | assert_eq!(request, expected_request); |
870 | | } |
871 | | |
872 | | #[test] |
873 | | fn test_try_read_double_request() { |
874 | | // Double request in a single read. |
875 | | let (mut sender, receiver) = UnixStream::pair().unwrap(); |
876 | | receiver.set_nonblocking(true).expect("Can't modify socket"); |
877 | | let mut conn = HttpConnection::new(receiver); |
878 | | sender |
879 | | .write_all( |
880 | | b"PATCH http://localhost/home HTTP/1.1\r\n\ |
881 | | Transfer-Encoding: chunked\r\n\ |
882 | | Content-Length: 26\r\n\r\nthis is not\n\r\na json \nbody", |
883 | | ) |
884 | | .unwrap(); |
885 | | sender |
886 | | .write_all( |
887 | | b"PUT http://farhost/away HTTP/1.1\r\nContent-Length: 23\r\n\r\nthis is another request", |
888 | | ) |
889 | | .unwrap(); |
890 | | |
891 | | let expected_request_first = Request { |
892 | | request_line: RequestLine::new(Method::Patch, "http://localhost/home", Version::Http11), |
893 | | headers: Headers::new(26, false, true), |
894 | | body: Some(Body::new(b"this is not\n\r\na json \nbody".to_vec())), |
895 | | files: Vec::new(), |
896 | | }; |
897 | | |
898 | | conn.try_read().unwrap(); |
899 | | let request_first = conn.pop_parsed_request().unwrap(); |
900 | | let request_second = conn.pop_parsed_request().unwrap(); |
901 | | |
902 | | let expected_request_second = Request { |
903 | | request_line: RequestLine::new(Method::Put, "http://farhost/away", Version::Http11), |
904 | | headers: Headers::new(23, false, false), |
905 | | body: Some(Body::new(b"this is another request".to_vec())), |
906 | | files: Vec::new(), |
907 | | }; |
908 | | assert_eq!(request_first, expected_request_first); |
909 | | assert_eq!(request_second, expected_request_second); |
910 | | } |
911 | | |
912 | | #[test] |
913 | | fn test_try_read_connection_closed() { |
914 | | // Connection abruptly closed. |
915 | | let (mut sender, receiver) = UnixStream::pair().unwrap(); |
916 | | receiver.set_nonblocking(true).expect("Can't modify socket"); |
917 | | let mut conn = HttpConnection::new(receiver); |
918 | | sender |
919 | | .write_all( |
920 | | b"PATCH http://localhost/home HTTP/1.1\r\n\ |
921 | | Transfer-Encoding: chunked\r\n\ |
922 | | Content-Len", |
923 | | ) |
924 | | .unwrap(); |
925 | | |
926 | | conn.try_read().unwrap(); |
927 | | sender.shutdown(std::net::Shutdown::Both).unwrap(); |
928 | | |
929 | | assert_eq!( |
930 | | conn.try_read().unwrap_err(), |
931 | | ConnectionError::ConnectionClosed |
932 | | ); |
933 | | } |
934 | | |
935 | | #[test] |
936 | | fn test_enqueue_response() { |
937 | | // Response without body. |
938 | | let (sender, mut receiver) = UnixStream::pair().unwrap(); |
939 | | receiver.set_nonblocking(true).expect("Can't modify socket"); |
940 | | let mut conn = HttpConnection::new(sender); |
941 | | |
942 | | let response = Response::new(Version::Http11, StatusCode::OK); |
943 | | let mut expected_response: Vec<u8> = vec![]; |
944 | | response.write_all(&mut expected_response).unwrap(); |
945 | | |
946 | | conn.enqueue_response(response); |
947 | | assert!(conn.try_write().is_ok()); |
948 | | |
949 | | let mut response_buffer = vec![0u8; expected_response.len()]; |
950 | | receiver.read_exact(&mut response_buffer).unwrap(); |
951 | | assert_eq!(response_buffer, expected_response); |
952 | | |
953 | | // Response with body. |
954 | | let (sender, mut receiver) = UnixStream::pair().unwrap(); |
955 | | receiver.set_nonblocking(true).expect("Can't modify socket"); |
956 | | let mut conn = HttpConnection::new(sender); |
957 | | let mut response = Response::new(Version::Http11, StatusCode::OK); |
958 | | let mut body: Vec<u8> = vec![]; |
959 | | body.write_all(br#"{ "json": "body", "hello": "world" }"#) |
960 | | .unwrap(); |
961 | | response.set_body(Body::new(body)); |
962 | | let mut expected_response: Vec<u8> = vec![]; |
963 | | response.write_all(&mut expected_response).unwrap(); |
964 | | |
965 | | conn.enqueue_response(response); |
966 | | assert!(conn.try_write().is_ok()); |
967 | | |
968 | | let mut response_buffer = vec![0u8; expected_response.len()]; |
969 | | receiver.read_exact(&mut response_buffer).unwrap(); |
970 | | assert_eq!(response_buffer, expected_response); |
971 | | } |
972 | | |
973 | | #[test] |
974 | | fn test_try_read_negative_content_len() { |
975 | | // Request with negative `Content-Length` header. |
976 | | let (mut sender, receiver) = UnixStream::pair().unwrap(); |
977 | | receiver.set_nonblocking(true).expect("Can't modify socket"); |
978 | | let mut conn = HttpConnection::new(receiver); |
979 | | sender |
980 | | .write_all( |
981 | | b"PUT http://localhost/home HTTP/1.1\r\n\ |
982 | | Content-Length: -1\r\n\r\n", |
983 | | ) |
984 | | .unwrap(); |
985 | | assert_eq!( |
986 | | conn.try_read().unwrap_err(), |
987 | | ConnectionError::ParseError(RequestError::HeaderError(HttpHeaderError::InvalidValue( |
988 | | "Content-Length".to_string(), |
989 | | " -1".to_string() |
990 | | ))) |
991 | | ); |
992 | | } |
993 | | |
994 | | #[test] |
995 | | fn test_payload_size_limit() { |
996 | | let (mut sender, receiver) = UnixStream::pair().unwrap(); |
997 | | receiver.set_nonblocking(true).expect("Can't modify socket"); |
998 | | let mut conn = HttpConnection::new(receiver); |
999 | | conn.set_payload_max_size(5); |
1000 | | sender |
1001 | | .write_all( |
1002 | | b"PUT http://localhost/home HTTP/1.1\r\n\ |
1003 | | Content-Length: 51200\r\n\r\naaaaaa", |
1004 | | ) |
1005 | | .unwrap(); |
1006 | | assert_eq!( |
1007 | | conn.try_read().unwrap_err(), |
1008 | | ConnectionError::ParseError(RequestError::SizeLimitExceeded(5, MAX_PAYLOAD_SIZE)) |
1009 | | ); |
1010 | | } |
1011 | | |
1012 | | #[test] |
1013 | | fn test_read_bytes() { |
1014 | | let (mut sender, receiver) = UnixStream::pair().unwrap(); |
1015 | | receiver.set_nonblocking(true).expect("Can't modify socket"); |
1016 | | let mut conn = HttpConnection::new(receiver); |
1017 | | |
1018 | | // Cursor positioned at buffer end. Read should fail. |
1019 | | conn.read_cursor = BUFFER_SIZE; |
1020 | | sender.write_all(b"hello\0").unwrap(); |
1021 | | assert_eq!( |
1022 | | conn.read_bytes().unwrap_err(), |
1023 | | ConnectionError::ParseError(RequestError::Overflow) |
1024 | | ); |
1025 | | |
1026 | | // Cursor positioned before buffer end. Partial read should succeed. |
1027 | | conn.read_cursor = BUFFER_SIZE - 3; |
1028 | | sender.write_all(b"hello\0").unwrap(); |
1029 | | assert_eq!(conn.read_bytes(), Ok(BUFFER_SIZE)); |
1030 | | |
1031 | | // Read the remaining 9 bytes - 3 left from the first "hello" and the 2nd full "hello". |
1032 | | conn.read_cursor = 0; |
1033 | | assert_eq!(conn.read_bytes(), Ok(9)); |
1034 | | sender.shutdown(Shutdown::Write).unwrap(); |
1035 | | assert_eq!( |
1036 | | conn.read_bytes().unwrap_err(), |
1037 | | ConnectionError::ConnectionClosed |
1038 | | ); |
1039 | | } |
1040 | | |
1041 | | #[test] |
1042 | | fn test_read_bytes_with_files() { |
1043 | | let (sender, receiver) = UnixStream::pair().unwrap(); |
1044 | | receiver.set_nonblocking(true).expect("Can't modify socket"); |
1045 | | let mut conn = HttpConnection::new(receiver); |
1046 | | |
1047 | | // Create 3 files, edit the content and rewind back to the start. |
1048 | | let mut file1 = TempFile::new().unwrap().into_file(); |
1049 | | let mut file2 = TempFile::new().unwrap().into_file(); |
1050 | | let mut file3 = TempFile::new().unwrap().into_file(); |
1051 | | file1.write_all(b"foo").unwrap(); |
1052 | | file1.seek(SeekFrom::Start(0)).unwrap(); |
1053 | | file2.write_all(b"bar").unwrap(); |
1054 | | file2.seek(SeekFrom::Start(0)).unwrap(); |
1055 | | file3.write_all(b"foobar").unwrap(); |
1056 | | file3.seek(SeekFrom::Start(0)).unwrap(); |
1057 | | |
1058 | | // Send 2 file descriptors along with 3 bytes of data. |
1059 | | assert_eq!( |
1060 | | sender.send_with_fds( |
1061 | | &[[1, 2, 3].as_ref()], |
1062 | | &[file1.into_raw_fd(), file2.into_raw_fd()] |
1063 | | ), |
1064 | | Ok(3) |
1065 | | ); |
1066 | | |
1067 | | // Check we receive the right amount of data along with the right |
1068 | | // amount of file descriptors. |
1069 | | assert_eq!(conn.read_bytes(), Ok(3)); |
1070 | | assert_eq!(conn.files.len(), 2); |
1071 | | |
1072 | | // Check the content of the data received |
1073 | | assert_eq!(conn.buffer[0], 1); |
1074 | | assert_eq!(conn.buffer[1], 2); |
1075 | | assert_eq!(conn.buffer[2], 3); |
1076 | | |
1077 | | // Check the file descriptors are usable by checking the content that |
1078 | | // can be read. |
1079 | | let mut buf = [0; 10]; |
1080 | | assert_eq!(conn.files[0].read(&mut buf).unwrap(), 3); |
1081 | | assert_eq!(&buf[..3], b"foo"); |
1082 | | assert_eq!(conn.files[1].read(&mut buf).unwrap(), 3); |
1083 | | assert_eq!(&buf[..3], b"bar"); |
1084 | | |
1085 | | // Send the 3rd file descriptor along with 1 byte of data. |
1086 | | assert_eq!( |
1087 | | sender.send_with_fds(&[[10].as_ref()], &[file3.into_raw_fd()]), |
1088 | | Ok(1) |
1089 | | ); |
1090 | | |
1091 | | // Check the amount of data along with the amount of file descriptors |
1092 | | // are updated. |
1093 | | assert_eq!(conn.read_bytes(), Ok(1)); |
1094 | | assert_eq!(conn.files.len(), 3); |
1095 | | |
1096 | | // Check the content of the new data received |
1097 | | assert_eq!(conn.buffer[0], 10); |
1098 | | |
1099 | | // Check the latest file descriptor is usable by checking the content |
1100 | | // that can be read. |
1101 | | let mut buf = [0; 10]; |
1102 | | assert_eq!(conn.files[2].read(&mut buf).unwrap(), 6); |
1103 | | assert_eq!(&buf[..6], b"foobar"); |
1104 | | |
1105 | | sender.shutdown(Shutdown::Write).unwrap(); |
1106 | | assert_eq!( |
1107 | | conn.read_bytes().unwrap_err(), |
1108 | | ConnectionError::ConnectionClosed |
1109 | | ); |
1110 | | } |
1111 | | |
1112 | | #[test] |
1113 | | fn test_shift_buffer_left() { |
1114 | | let (_, receiver) = UnixStream::pair().unwrap(); |
1115 | | let mut conn = HttpConnection::new(receiver); |
1116 | | |
1117 | | assert_eq!( |
1118 | | conn.shift_buffer_left(0, conn.buffer.len() + 1) |
1119 | | .unwrap_err(), |
1120 | | RequestError::Overflow |
1121 | | ); |
1122 | | assert_eq!( |
1123 | | conn.shift_buffer_left(1, 0).unwrap_err(), |
1124 | | RequestError::Underflow |
1125 | | ); |
1126 | | assert!(conn.shift_buffer_left(1, conn.buffer.len()).is_ok()); |
1127 | | } |
1128 | | |
1129 | | #[test] |
1130 | | fn test_parse_request_line() { |
1131 | | let (_, receiver) = UnixStream::pair().unwrap(); |
1132 | | let mut conn = HttpConnection::new(receiver); |
1133 | | |
1134 | | // Error case: end past buffer end. |
1135 | | assert_eq!( |
1136 | | conn.parse_request_line(&mut 0, conn.buffer.len() + 1) |
1137 | | .unwrap_err(), |
1138 | | ConnectionError::ParseError(RequestError::Overflow) |
1139 | | ); |
1140 | | |
1141 | | // Error case: start is past end. |
1142 | | assert_eq!( |
1143 | | conn.parse_request_line(&mut 1, 0).unwrap_err(), |
1144 | | ConnectionError::ParseError(RequestError::Underflow) |
1145 | | ); |
1146 | | |
1147 | | // Error case: the request line is longer than BUFFER_SIZE. |
1148 | | assert_eq!( |
1149 | | conn.parse_request_line(&mut 0, BUFFER_SIZE).unwrap_err(), |
1150 | | ConnectionError::ParseError(RequestError::InvalidRequest) |
1151 | | ); |
1152 | | |
1153 | | // OK case. |
1154 | | assert_eq!(conn.parse_request_line(&mut 1, BUFFER_SIZE), Ok(false)); |
1155 | | |
1156 | | // Error case: invalid content. |
1157 | | conn.buffer[0..8].copy_from_slice(b"foo\r\nbar"); |
1158 | | assert_eq!( |
1159 | | conn.parse_request_line(&mut 0, BUFFER_SIZE).unwrap_err(), |
1160 | | ConnectionError::ParseError(RequestError::InvalidRequest) |
1161 | | ); |
1162 | | |
1163 | | // OK case. |
1164 | | conn.buffer[0..29].copy_from_slice(b"GET http://foo/bar HTTP/1.1\r\n"); |
1165 | | assert_eq!(conn.parse_request_line(&mut 0, BUFFER_SIZE), Ok(true)); |
1166 | | } |
1167 | | |
1168 | | #[test] |
1169 | | fn test_parse_headers() { |
1170 | | let (_, receiver) = UnixStream::pair().unwrap(); |
1171 | | let mut conn = HttpConnection::new(receiver); |
1172 | | |
1173 | | // Error case: end_cursor past buffer end. |
1174 | | assert_eq!( |
1175 | | conn.parse_headers(&mut 0, conn.buffer.len() + 1) |
1176 | | .unwrap_err(), |
1177 | | ConnectionError::ParseError(RequestError::Overflow) |
1178 | | ); |
1179 | | |
1180 | | // Error case: line_start_index is past end_cursor. |
1181 | | assert_eq!( |
1182 | | conn.parse_headers(&mut 1, 0).unwrap_err(), |
1183 | | ConnectionError::ParseError(RequestError::Underflow) |
1184 | | ); |
1185 | | |
1186 | | // Error case: no request pending. |
1187 | | // CRLF can be at the start of the buffer... |
1188 | | conn.buffer[0] = CR; |
1189 | | conn.buffer[1] = LF; |
1190 | | assert_eq!( |
1191 | | conn.parse_headers(&mut 0, BUFFER_SIZE).unwrap_err(), |
1192 | | ConnectionError::ParseError(RequestError::HeadersWithoutPendingRequest) |
1193 | | ); |
1194 | | // ...or somewhere in the middle. |
1195 | | conn.buffer[0] = 0; |
1196 | | conn.buffer[1] = CR; |
1197 | | conn.buffer[2] = LF; |
1198 | | assert_eq!( |
1199 | | conn.parse_headers(&mut 0, BUFFER_SIZE).unwrap_err(), |
1200 | | ConnectionError::ParseError(RequestError::HeadersWithoutPendingRequest) |
1201 | | ); |
1202 | | |
1203 | | // Error case: invalid header. |
1204 | | conn.pending_request = Some(Request { |
1205 | | request_line: RequestLine::new(Method::Get, "http://foo/bar", Version::Http11), |
1206 | | headers: Headers::new(0, true, true), |
1207 | | body: None, |
1208 | | files: Vec::new(), |
1209 | | }); |
1210 | | assert_eq!( |
1211 | | conn.parse_headers(&mut 0, BUFFER_SIZE).unwrap_err(), |
1212 | | ConnectionError::ParseError(RequestError::HeaderError(HttpHeaderError::InvalidFormat( |
1213 | | "\0".to_string() |
1214 | | ))) |
1215 | | ); |
1216 | | |
1217 | | // OK case: incomplete header line. |
1218 | | let hdr = b"Custom-Header-Testing: 1"; |
1219 | | conn.buffer[..hdr.len()].copy_from_slice(hdr); |
1220 | | assert_eq!(conn.parse_headers(&mut 0, hdr.len()), Ok(false)); |
1221 | | |
1222 | | // OK case: complete header line. |
1223 | | let hdr = b"Custom-Header-Testing: 1\r\n"; |
1224 | | conn.buffer[..hdr.len()].copy_from_slice(hdr); |
1225 | | assert_eq!(conn.parse_headers(&mut 0, hdr.len()), Ok(true)); |
1226 | | |
1227 | | // OK case: complete header line, end of header. |
1228 | | let hdr = b"\r\n"; |
1229 | | conn.buffer[..hdr.len()].copy_from_slice(hdr); |
1230 | | assert_eq!(conn.parse_headers(&mut 0, hdr.len()), Ok(true)); |
1231 | | } |
1232 | | |
1233 | | #[test] |
1234 | | fn test_parse_body() { |
1235 | | let (_, receiver) = UnixStream::pair().unwrap(); |
1236 | | let mut conn = HttpConnection::new(receiver); |
1237 | | |
1238 | | // Error case: end_cursor past buffer end. |
1239 | | assert_eq!( |
1240 | | conn.parse_body(&mut 0usize, conn.buffer.len() + 1) |
1241 | | .unwrap_err(), |
1242 | | ConnectionError::ParseError(RequestError::Overflow) |
1243 | | ); |
1244 | | |
1245 | | // Error case: line_start_index is past end_cursor. |
1246 | | assert_eq!( |
1247 | | conn.parse_body(&mut 1usize, 0usize).unwrap_err(), |
1248 | | ConnectionError::ParseError(RequestError::Underflow) |
1249 | | ); |
1250 | | |
1251 | | // OK case: consume the buffer. |
1252 | | conn.body_bytes_to_be_read = 1; |
1253 | | assert_eq!(conn.parse_body(&mut 0usize, 0usize), Ok(false)); |
1254 | | |
1255 | | // Error case: there's more body to be parsed, but no pending request set. |
1256 | | assert_eq!( |
1257 | | conn.parse_body(&mut 0, BUFFER_SIZE).unwrap_err(), |
1258 | | ConnectionError::ParseError(RequestError::BodyWithoutPendingRequest) |
1259 | | ); |
1260 | | |
1261 | | // Error case: read more bytes than we should have into the body of the request. |
1262 | | conn.pending_request = Some(Request { |
1263 | | request_line: RequestLine::new(Method::Get, "http://foo/bar", Version::Http11), |
1264 | | headers: Headers::new(0, true, true), |
1265 | | body: None, |
1266 | | files: Vec::new(), |
1267 | | }); |
1268 | | conn.body_vec = vec![0xde, 0xad, 0xbe, 0xef]; |
1269 | | assert_eq!( |
1270 | | conn.parse_body(&mut 0, BUFFER_SIZE).unwrap_err(), |
1271 | | ConnectionError::ParseError(RequestError::InvalidRequest) |
1272 | | ); |
1273 | | |
1274 | | // OK case. |
1275 | | conn.body_vec.clear(); |
1276 | | assert_eq!(conn.parse_body(&mut 0, BUFFER_SIZE), Ok(true)); |
1277 | | } |
1278 | | } |