upstream_rs.rs (10182B)
1 /* 2 This file is part of Paivana. 3 Copyright (C) 2026 Taler Systems SA 4 5 Paivana is free software; you can redistribute it and/or 6 modify it under the terms of the GNU Affero General Public License 7 as published by the Free Software Foundation; either version 8 3, or (at your option) any later version. 9 10 Paivana is distributed in the hope that it will be useful, but 11 WITHOUT ANY WARRANTY; without even the implied warranty of 12 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 13 GNU Affero General Public License for more details. 14 15 You should have received a copy of the GNU Affero General Public 16 License along with Paivana; see the file COPYING. If not, 17 write to the Free Software Foundation, Inc., 51 Franklin 18 Street, Fifth Floor, Boston, MA 02110-1301, USA. 19 */ 20 21 // upstream_rs: Rust-based upstream HTTP server used by the 22 // paivana reverse-proxy tests. Implements the same small set 23 // of canned endpoints as upstream_mhd.c, using only the std 24 // library so it can be built with just `rustc`. 25 26 use std::env; 27 use std::io::{BufRead, BufReader, Read, Write}; 28 use std::net::{TcpListener, TcpStream}; 29 use std::sync::Arc; 30 use std::thread; 31 use std::time::Duration; 32 33 const UPSTREAM: &str = "rs"; 34 35 // Largest body /large/ will serve. The buffer is built once at 36 // startup and sliced per request rather than regenerated: a 37 // byte-at-a-time fill of 64 KiB is real work, and benchmark.sh 38 // compares this server against paivana in front of it, where anything 39 // the origin spends on manufacturing the page is charged to the arm 40 // that does not have the proxy in it. 41 const LARGE_MAX: usize = 10 * 1024 * 1024; 42 43 struct Request { 44 method: String, 45 path: String, 46 headers: Vec<(String, String)>, 47 body: Vec<u8>, 48 } 49 50 fn parse_request(stream: &mut TcpStream) -> Option<Request> { 51 let mut br = BufReader::new(stream); 52 let mut line = String::new(); 53 if br.read_line(&mut line).ok()? == 0 { 54 return None; 55 } 56 let mut parts = line.trim_end().splitn(3, ' '); 57 let method = parts.next()?.to_string(); 58 let path = parts.next()?.to_string(); 59 let mut headers = Vec::new(); 60 loop { 61 let mut h = String::new(); 62 if br.read_line(&mut h).ok()? == 0 { 63 break; 64 } 65 let t = h.trim_end(); 66 if t.is_empty() { 67 break; 68 } 69 if let Some(idx) = t.find(':') { 70 let k = t[..idx].trim().to_string(); 71 let v = t[idx + 1..].trim().to_string(); 72 headers.push((k, v)); 73 } 74 } 75 // Read body if there's a Content-Length 76 let cl: usize = headers 77 .iter() 78 .find(|(k, _)| k.eq_ignore_ascii_case("Content-Length")) 79 .and_then(|(_, v)| v.parse().ok()) 80 .unwrap_or(0); 81 let mut body = vec![0u8; cl]; 82 if cl > 0 { 83 if br.read_exact(&mut body).is_err() { 84 return None; 85 } 86 } 87 Some(Request { 88 method, 89 path, 90 headers, 91 body, 92 }) 93 } 94 95 fn send_response(stream: &mut TcpStream, code: u16, reason: &str, 96 content_type: &str, body: &[u8], extra_headers: &[(&str, &str)]) { 97 let mut head = format!( 98 "HTTP/1.1 {} {}\r\nX-Upstream: {}\r\nContent-Type: {}\r\nContent-Length: {}\r\n", 99 code, 100 reason, 101 UPSTREAM, 102 content_type, 103 body.len() 104 ); 105 for (k, v) in extra_headers { 106 head.push_str(&format!("{}: {}\r\n", k, v)); 107 } 108 // See client_loop(): this server answers one request per 109 // connection and then drops the stream, so it has to say so. 110 head.push_str("Connection: close\r\n\r\n"); 111 let _ = stream.write_all(head.as_bytes()); 112 let _ = stream.write_all(body); 113 } 114 115 fn handle(req: &Request, stream: &mut TcpStream, large: &[u8]) { 116 if req.method == "OPTIONS" { 117 send_response( 118 stream, 204, "No Content", "text/plain", &[], 119 &[("Allow", "GET, POST, PUT, PATCH, DELETE, HEAD, OPTIONS")], 120 ); 121 return; 122 } 123 if req.path == "/hello" && (req.method == "GET" || req.method == "HEAD") { 124 let body = format!("Hello from {}\n", UPSTREAM); 125 let b: &[u8] = if req.method == "HEAD" { &[] } else { body.as_bytes() }; 126 // For HEAD, still advertise correct Content-Length of the would-be body. 127 if req.method == "HEAD" { 128 let head = format!( 129 "HTTP/1.1 200 OK\r\nX-Upstream: {}\r\nContent-Type: text/plain\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", 130 UPSTREAM, 131 body.len() 132 ); 133 let _ = stream.write_all(head.as_bytes()); 134 } else { 135 send_response(stream, 200, "OK", "text/plain", b, &[]); 136 } 137 return; 138 } 139 if let Some(rest) = req.path.strip_prefix("/status/") { 140 let code: u16 = rest.parse().unwrap_or(500); 141 let code = if !(100..=599).contains(&code) { 500 } else { code }; 142 let body = format!("status {}\n", code); 143 send_response(stream, code, "Status", "text/plain", body.as_bytes(), &[]); 144 return; 145 } 146 if let Some(rest) = req.path.strip_prefix("/large/") { 147 let n: usize = rest.parse().unwrap_or(0).min(LARGE_MAX); 148 send_response(stream, 200, "OK", "application/octet-stream", &large[..n], &[]); 149 return; 150 } 151 if let Some(rest) = req.path.strip_prefix("/slow/") { 152 let ms: u64 = rest.parse().unwrap_or(0).min(30000); 153 thread::sleep(Duration::from_millis(ms)); 154 send_response(stream, 200, "OK", "text/plain", b"slept\n", &[]); 155 return; 156 } 157 if req.path == "/conn-response" && req.method == "GET" { 158 // Name two of our own response headers in Connection; a 159 // conforming proxy must strip both (RFC 9110 ยง7.6.1) but keep 160 // X-Keep-Resp. X-Hop-Before is emitted *before* the 161 // Connection header that names it and X-Hop-After *after* it, 162 // so a single-pass filter cannot catch both. 163 let body = b"conn\n"; 164 let head = format!( 165 "HTTP/1.1 200 OK\r\nX-Upstream: {}\r\nX-Hop-Before: must-not-leak\r\n\ 166 Connection: X-Hop-Before, X-Hop-After\r\nX-Hop-After: must-not-leak\r\n\ 167 X-Keep-Resp: survivor\r\nContent-Type: text/plain\r\n\ 168 Content-Length: {}\r\n\r\n", 169 UPSTREAM, 170 body.len() 171 ); 172 let _ = stream.write_all(head.as_bytes()); 173 let _ = stream.write_all(body); 174 return; 175 } 176 if req.path == "/echo-headers" && req.method == "GET" { 177 let mut b = String::new(); 178 for (k, v) in &req.headers { 179 b.push_str(&format!("{}: {}\n", k, v)); 180 } 181 send_response(stream, 200, "OK", "text/plain", b.as_bytes(), &[]); 182 return; 183 } 184 if req.path == "/echo" && req.method == "POST" { 185 send_response(stream, 200, "OK", "application/octet-stream", &req.body, &[]); 186 return; 187 } 188 if req.path == "/upload" && req.method == "POST" { 189 let b = format!("Received {} bytes\n", req.body.len()); 190 send_response(stream, 200, "OK", "text/plain", b.as_bytes(), &[]); 191 return; 192 } 193 if req.path == "/put" && req.method == "PUT" { 194 let b = format!("PUT received {}\n", req.body.len()); 195 send_response(stream, 200, "OK", "text/plain", b.as_bytes(), &[]); 196 return; 197 } 198 if req.path == "/patch" && req.method == "PATCH" { 199 let b = format!("PATCH received {}\n", req.body.len()); 200 send_response(stream, 200, "OK", "text/plain", b.as_bytes(), &[]); 201 return; 202 } 203 if req.path.starts_with("/item") && req.method == "DELETE" { 204 send_response(stream, 204, "No Content", "text/plain", &[], &[]); 205 return; 206 } 207 send_response(stream, 404, "Not Found", "text/plain", b"not found\n", &[]); 208 } 209 210 fn client_loop(mut stream: TcpStream, large: &[u8]) { 211 // We can't easily loop keep-alive with our BufReader pattern without 212 // ownership gymnastics; handle one request per connection. That is 213 // why every response above carries `Connection: close': paivana 214 // puts its outbound handles on a shared curl multi handle and sets 215 // neither CURLOPT_FORBID_REUSE nor CURLOPT_FRESH_CONNECT, so a 216 // connection we told it to keep is a connection it will reuse -- 217 // and find shut. libcurl retries an idempotent request, which 218 // hides it, but a POST or PUT whose body is already partly on the 219 // wire is not retried: the transfer fails, paivana answers 502, and 220 // the suite reports an intermittent proxy bug that is really this 221 // server lying about its own connection handling. 222 if let Some(req) = parse_request(&mut stream) { 223 handle(&req, &mut stream, large); 224 } 225 } 226 227 fn main() { 228 // An argument that is not a port must be an error, not a silent 229 // fallback to the default: the driver would then wait five 230 // seconds for a port nothing ever bound and blame that port. 231 let port: u16 = match env::args().nth(1) { 232 None => 8404, 233 Some(s) => match s.parse::<u16>() { 234 Ok(p) if p >= 1 => p, 235 _ => { 236 eprintln!("invalid port {:?}", s); 237 std::process::exit(1); 238 } 239 }, 240 }; 241 // Filled before the listener exists: bind() already starts queueing 242 // connections, and the readiness probe the test driver uses is a 243 // successful connect, so a fill after the bind would be time the 244 // driver has been told the server is ready for. 245 let large: Arc<Vec<u8>> = 246 Arc::new((0..LARGE_MAX).map(|i| b'A' + ((i % 26) as u8)).collect()); 247 // Loopback only: this server echoes an arbitrary POST body back 248 // and hands out 10 MiB on request, and has no business being 249 // reachable from the network for the duration of `make check'. 250 let listener = TcpListener::bind(("127.0.0.1", port)).expect("bind failed"); 251 eprintln!("upstream_rs listening on port {}", port); 252 for stream in listener.incoming() { 253 match stream { 254 Ok(s) => { 255 let large = Arc::clone(&large); 256 thread::spawn(move || client_loop(s, &large)); 257 } 258 Err(_) => continue, 259 } 260 } 261 }