grove
grove/src/http.rs
at @8
1//! A small HTTP/1.1 server: one request per connection, bodies by Content-Length.
2//!
3//! grove sits behind Caddy, which handles TLS, HTTP/2 and keep-alive towards clients; this
4//! side only has to be correct and bounded.
5
6use std::io::{self, BufRead, BufReader, Read, Write};
7use std::net::{TcpListener, TcpStream};
8use std::sync::atomic::{AtomicUsize, Ordering};
9use std::sync::Arc;
10use std::time::Duration;
11
12const MAX_HEAD: usize = 64 << 10;
13pub const MAX_BODY: usize = 512 << 20;
14const MAX_CONNECTIONS: usize = 128;
15
16pub struct Request {
17 pub method: String,
18 pub path: String,
19 pub query: String,
20 pub headers: Vec<(String, String)>,
21 pub body: Vec<u8>,
22}
23
24impl Request {
25 pub fn header(&self, name: &str) -> Option<&str> {
26 self.headers.iter().find(|(k, _)| k.eq_ignore_ascii_case(name)).map(|(_, v)| v.as_str())
27 }
28}
29
30pub struct Response {
31 pub status: u16,
32 pub content_type: &'static str,
33 pub body: Vec<u8>,
34}
35
36impl Response {
37 pub fn json(status: u16, v: &crate::json::Value) -> Response {
38 Response { status, content_type: "application/json", body: v.encode().into_bytes() }
39 }
40 pub fn bytes(body: Vec<u8>) -> Response {
41 Response { status: 200, content_type: "application/octet-stream", body }
42 }
43 pub fn error(status: u16, msg: &str) -> Response {
44 Response::json(status, &crate::json::obj([("error", msg.into())]))
45 }
46}
47
48fn reason(status: u16) -> &'static str {
49 match status {
50 200 => "OK",
51 400 => "Bad Request",
52 401 => "Unauthorized",
53 403 => "Forbidden",
54 404 => "Not Found",
55 405 => "Method Not Allowed",
56 409 => "Conflict",
57 411 => "Length Required",
58 413 => "Payload Too Large",
59 500 => "Internal Server Error",
60 503 => "Service Unavailable",
61 _ => "",
62 }
63}
64
65fn bad(msg: &str) -> io::Error {
66 io::Error::new(io::ErrorKind::InvalidData, msg.to_string())
67}
68
69/// Read one request. Errors carry the status to answer with.
70fn read_request(stream: &TcpStream) -> Result<Request, (u16, String)> {
71 let mut reader = BufReader::new(stream);
72 let mut head = Vec::new();
73 loop {
74 let mut line = Vec::new();
75 let n = reader.read_until(b'\n', &mut line).map_err(|e| (400, e.to_string()))?;
76 if n == 0 {
77 return Err((400, "connection closed".into()));
78 }
79 head.extend_from_slice(&line);
80 if head.len() > MAX_HEAD {
81 return Err((413, "headers too large".into()));
82 }
83 if line == b"\r\n" || line == b"\n" {
84 break;
85 }
86 }
87 let head = String::from_utf8(head).map_err(|_| (400, "headers aren't UTF-8".to_string()))?;
88 let mut lines = head.lines();
89 let first = lines.next().unwrap_or("");
90 let mut parts = first.split(' ');
91 let (method, target, version) = (parts.next().unwrap_or(""), parts.next().unwrap_or(""), parts.next().unwrap_or(""));
92 if method.is_empty() || !target.starts_with('/') || !version.starts_with("HTTP/1.") {
93 return Err((400, "bad request line".into()));
94 }
95 let mut headers = Vec::new();
96 for line in lines {
97 if line.is_empty() {
98 break;
99 }
100 let (k, v) = line.split_once(':').ok_or((400, "bad header".to_string()))?;
101 headers.push((k.trim().to_string(), v.trim().to_string()));
102 }
103 let find = |name: &str| headers.iter().find(|(k, _)| k.eq_ignore_ascii_case(name)).map(|(_, v)| v.clone());
104 if find("transfer-encoding").is_some() {
105 return Err((411, "send a Content-Length".into()));
106 }
107 let len = match find("content-length") {
108 Some(v) => v.parse::<usize>().map_err(|_| (400, "bad Content-Length".to_string()))?,
109 None => 0,
110 };
111 if len > MAX_BODY {
112 return Err((413, "body too large".into()));
113 }
114 let mut body = vec![0u8; len];
115 reader.read_exact(&mut body).map_err(|e| (400, e.to_string()))?;
116 let (path, query) = match target.split_once('?') {
117 Some((p, q)) => (p.to_string(), q.to_string()),
118 None => (target.to_string(), String::new()),
119 };
120 Ok(Request { method: method.to_string(), path, query, headers, body })
121}
122
123fn write_response(stream: &mut TcpStream, r: &Response) -> io::Result<()> {
124 let head = format!(
125 "HTTP/1.1 {} {}\r\nContent-Type: {}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
126 r.status,
127 reason(r.status),
128 r.content_type,
129 r.body.len()
130 );
131 stream.write_all(head.as_bytes())?;
132 stream.write_all(&r.body)?;
133 stream.flush()
134}
135
136pub type Handler = Arc<dyn Fn(Request) -> Response + Send + Sync>;
137
138/// Accept connections forever, each on its own thread, at most MAX_CONNECTIONS at once.
139pub fn serve(listener: TcpListener, handler: Handler) -> io::Result<()> {
140 let active = Arc::new(AtomicUsize::new(0));
141 for stream in listener.incoming() {
142 let mut stream = match stream {
143 Ok(s) => s,
144 Err(_) => continue,
145 };
146 if active.load(Ordering::SeqCst) >= MAX_CONNECTIONS {
147 let _ = write_response(&mut stream, &Response::error(503, "busy"));
148 continue;
149 }
150 active.fetch_add(1, Ordering::SeqCst);
151 let (handler, active) = (handler.clone(), active.clone());
152 std::thread::spawn(move || {
153 let _ = stream.set_read_timeout(Some(Duration::from_secs(60)));
154 let _ = stream.set_write_timeout(Some(Duration::from_secs(60)));
155 let response = match read_request(&stream) {
156 // A bug in one request must not take the server down: answer 500 and carry on.
157 Ok(req) => std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| handler(req)))
158 .unwrap_or_else(|_| Response::error(500, "internal error")),
159 Err((status, msg)) => Response::error(status, &msg),
160 };
161 let _ = write_response(&mut stream, &response);
162 active.fetch_sub(1, Ordering::SeqCst);
163 });
164 }
165 Err(bad("listener closed"))
166}