paivana

upstream_rs.rs (7239B)

---

 /*
   This file is part of Paivana.
   Copyright (C) 2026 Taler Systems SA
 
   Paivana is free software; you can redistribute it and/or
   modify it under the terms of the GNU General Public License
   as published by the Free Software Foundation; either version
   3, or (at your option) any later version.
 
   Paivana is distributed in the hope that it will be useful, but
   WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.
 
   You should have received a copy of the GNU General Public
   License along with Paivana; see the file COPYING.  If not,
   write to the Free Software Foundation, Inc., 51 Franklin
   Street, Fifth Floor, Boston, MA 02110-1301, USA.
 */
 
 // upstream_rs: Rust-based upstream HTTP server used by the
 // paivana reverse-proxy tests.  Implements the same small set
 // of canned endpoints as upstream_mhd.c, using only the std
 // library so it can be built with just `rustc`.
 
 use std::env;
 use std::io::{BufRead, BufReader, Read, Write};
 use std::net::{TcpListener, TcpStream};
 use std::thread;
 use std::time::Duration;
 
 const UPSTREAM: &str = "rs";
 
 struct Request {
     method: String,
     path: String,
     headers: Vec<(String, String)>,
     body: Vec<u8>,
 }
 
 fn parse_request(stream: &mut TcpStream) -> Option<Request> {
     let mut br = BufReader::new(stream);
     let mut line = String::new();
     if br.read_line(&mut line).ok()? == 0 {
         return None;
     }
     let mut parts = line.trim_end().splitn(3, ' ');
     let method = parts.next()?.to_string();
     let path = parts.next()?.to_string();
     let mut headers = Vec::new();
     loop {
         let mut h = String::new();
         if br.read_line(&mut h).ok()? == 0 {
             break;
         }
         let t = h.trim_end();
         if t.is_empty() {
             break;
         }
         if let Some(idx) = t.find(':') {
             let k = t[..idx].trim().to_string();
             let v = t[idx + 1..].trim().to_string();
             headers.push((k, v));
         }
     }
     // Read body if there's a Content-Length
     let cl: usize = headers
         .iter()
         .find(|(k, _)| k.eq_ignore_ascii_case("Content-Length"))
         .and_then(|(_, v)| v.parse().ok())
         .unwrap_or(0);
     let mut body = vec![0u8; cl];
     if cl > 0 {
         if br.read_exact(&mut body).is_err() {
             return None;
         }
     }
     Some(Request {
         method,
         path,
         headers,
         body,
     })
 }
 
 fn send_response(stream: &mut TcpStream, code: u16, reason: &str,
                  content_type: &str, body: &[u8], extra_headers: &[(&str, &str)]) {
     let mut head = format!(
         "HTTP/1.1 {} {}\r\nX-Upstream: {}\r\nContent-Type: {}\r\nContent-Length: {}\r\n",
         code,
         reason,
         UPSTREAM,
         content_type,
         body.len()
     );
     for (k, v) in extra_headers {
         head.push_str(&format!("{}: {}\r\n", k, v));
     }
     head.push_str("Connection: keep-alive\r\n\r\n");
     let _ = stream.write_all(head.as_bytes());
     let _ = stream.write_all(body);
 }
 
 fn handle(req: &Request, stream: &mut TcpStream) {
     if req.method == "OPTIONS" {
         send_response(
             stream, 204, "No Content", "text/plain", &[],
             &[("Allow", "GET, POST, PUT, PATCH, DELETE, HEAD, OPTIONS")],
         );
         return;
     }
     if req.path == "/hello" && (req.method == "GET" || req.method == "HEAD") {
         let body = format!("Hello from {}\n", UPSTREAM);
         let b: &[u8] = if req.method == "HEAD" { &[] } else { body.as_bytes() };
         // For HEAD, still advertise correct Content-Length of the would-be body.
         if req.method == "HEAD" {
             let head = format!(
                 "HTTP/1.1 200 OK\r\nX-Upstream: {}\r\nContent-Type: text/plain\r\nContent-Length: {}\r\nConnection: keep-alive\r\n\r\n",
                 UPSTREAM,
                 body.len()
             );
             let _ = stream.write_all(head.as_bytes());
         } else {
             send_response(stream, 200, "OK", "text/plain", b, &[]);
         }
         return;
     }
     if let Some(rest) = req.path.strip_prefix("/status/") {
         let code: u16 = rest.parse().unwrap_or(500);
         let code = if !(100..=599).contains(&code) { 500 } else { code };
         let body = format!("status {}\n", code);
         send_response(stream, code, "Status", "text/plain", body.as_bytes(), &[]);
         return;
     }
     if let Some(rest) = req.path.strip_prefix("/large/") {
         let n: usize = rest.parse().unwrap_or(0).min(10 * 1024 * 1024);
         let mut buf = Vec::with_capacity(n);
         for i in 0..n {
             buf.push(b'A' + ((i % 26) as u8));
         }
         send_response(stream, 200, "OK", "application/octet-stream", &buf, &[]);
         return;
     }
     if let Some(rest) = req.path.strip_prefix("/slow/") {
         let ms: u64 = rest.parse().unwrap_or(0).min(30000);
         thread::sleep(Duration::from_millis(ms));
         send_response(stream, 200, "OK", "text/plain", b"slept\n", &[]);
         return;
     }
     if req.path == "/echo-headers" && req.method == "GET" {
         let mut b = String::new();
         for (k, v) in &req.headers {
             b.push_str(&format!("{}: {}\n", k, v));
         }
         send_response(stream, 200, "OK", "text/plain", b.as_bytes(), &[]);
         return;
     }
     if req.path == "/echo" && req.method == "POST" {
         send_response(stream, 200, "OK", "application/octet-stream", &req.body, &[]);
         return;
     }
     if req.path == "/upload" && req.method == "POST" {
         let b = format!("Received {} bytes\n", req.body.len());
         send_response(stream, 200, "OK", "text/plain", b.as_bytes(), &[]);
         return;
     }
     if req.path == "/put" && req.method == "PUT" {
         let b = format!("PUT received {}\n", req.body.len());
         send_response(stream, 200, "OK", "text/plain", b.as_bytes(), &[]);
         return;
     }
     if req.path == "/patch" && req.method == "PATCH" {
         let b = format!("PATCH received {}\n", req.body.len());
         send_response(stream, 200, "OK", "text/plain", b.as_bytes(), &[]);
         return;
     }
     if req.path.starts_with("/item") && req.method == "DELETE" {
         send_response(stream, 204, "No Content", "text/plain", &[], &[]);
         return;
     }
     send_response(stream, 404, "Not Found", "text/plain", b"not found\n", &[]);
 }
 
 fn client_loop(mut stream: TcpStream) {
     // We can't easily loop keep-alive with our BufReader pattern without
     // ownership gymnastics; handle one request per connection.  paivana
     // opens fresh connections to us, so this is fine.
     if let Some(req) = parse_request(&mut stream) {
         handle(&req, &mut stream);
     }
 }
 
 fn main() {
     let port: u16 = env::args()
         .nth(1)
         .and_then(|s| s.parse().ok())
         .unwrap_or(8404);
     let listener = TcpListener::bind(("0.0.0.0", port)).expect("bind failed");
     eprintln!("upstream_rs listening on port {}", port);
     for stream in listener.incoming() {
         match stream {
             Ok(s) => {
                 thread::spawn(move || client_loop(s));
             }
             Err(_) => continue,
         }
     }
 }