fix: cap listener connections, serve HEAD, fix feed parser CR round-trip

This commit is contained in:
randogoth 2026-10-05 11:12:23 +03:00
parent 1e782310b6
commit 267625b411
3 changed files with 244 additions and 13 deletions

View file

@ -4,6 +4,7 @@
use std::io::{Read, Write};
use std::net::{TcpListener, TcpStream};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::UNIX_EPOCH;
use fumi_core::crypto::sha256;
@ -12,13 +13,50 @@ use smolboid_core::time;
use crate::feedstore::FeedStore;
const MAX_REQUEST: usize = 8 * 1024;
/// Concurrent connection cap: each connection holds a thread until its socket
/// timeouts, so a flood must be shed rather than absorbed (audit.md).
const MAX_CONNECTIONS: usize = 64;
static ACTIVE: AtomicUsize = AtomicUsize::new(0);
/// Decrements the connection count when the handler finishes, including on
/// early returns, so the cap cannot leak slots.
struct ConnGuard;
impl ConnGuard {
fn acquire() -> Self {
ACTIVE.fetch_add(1, Ordering::Relaxed);
ConnGuard
}
}
impl Drop for ConnGuard {
fn drop(&mut self) {
ACTIVE.fetch_sub(1, Ordering::Relaxed);
}
}
pub fn serve(listener: TcpListener, feeds: FeedStore) -> ! {
loop {
match listener.accept() {
Ok((stream, _)) => {
Ok((mut stream, _)) => {
if ACTIVE.load(Ordering::Relaxed) >= MAX_CONNECTIONS {
let mut shed = stream.try_clone().ok();
if let Some(shed) = shed.as_mut() {
let _ = write_simple(shed, 503, "Service Unavailable", "busy\n", true);
}
continue;
}
let feeds = feeds.clone();
std::thread::spawn(move || handle(stream, feeds));
let spawned = std::thread::Builder::new().spawn(move || {
let _guard = ConnGuard::acquire();
handle(stream, feeds);
});
// A failed spawn must not strand a slot: handle acquires its
// own guard, so nothing was counted until the thread ran.
if spawned.is_err() {
eprintln!("http: connection thread failed to spawn");
}
}
Err(e) => eprintln!("http: accept failed: {e}"),
}
@ -37,26 +75,27 @@ fn handle(mut stream: TcpStream, feeds: FeedStore) {
let (method, path, headers) = match parse_request(&request) {
Some(parts) => parts,
None => {
let _ = write_simple(&mut stream, 400, "Bad Request", "bad request\n");
let _ = write_simple(&mut stream, 400, "Bad Request", "bad request\n", true);
return;
}
};
if method != "GET" {
let _ = write_simple(&mut stream, 405, "Method Not Allowed", "GET only\n");
let head_only = method == "HEAD";
if method != "GET" && !head_only {
let _ = write_simple(&mut stream, 405, "Method Not Allowed", "GET, HEAD only\n", true);
return;
}
let Some(username) = username_of(path) else {
let _ = write_simple(&mut stream, 404, "Not Found", "not found\n");
let _ = write_simple(&mut stream, 404, "Not Found", "not found\n", !head_only);
return;
};
// An admitted author's feed exists from `admit` on; anything else,
// including a valid username with no feed, is not there to be served.
if !feeds.exists(&username).unwrap_or(false) {
let _ = write_simple(&mut stream, 404, "Not Found", "not found\n");
let _ = write_simple(&mut stream, 404, "Not Found", "not found\n", !head_only);
return;
}
let Ok(feed) = feeds.load(&username) else {
let _ = write_simple(&mut stream, 404, "Not Found", "not found\n");
let _ = write_simple(&mut stream, 404, "Not Found", "not found\n", !head_only);
return;
};
let body = feed.render().into_bytes();
@ -80,7 +119,10 @@ fn handle(mut stream: TcpStream, feeds: FeedStore) {
body.len()
));
let _ = stream.write_all(head.as_bytes());
let _ = stream.write_all(&body);
// HEAD answers with the headers GET would, minus the body (audit.md).
if !head_only {
let _ = stream.write_all(&body);
}
}
/// `If-None-Match` takes precedence; `If-Modified-Since` applies only without
@ -136,12 +178,22 @@ fn parse_request(request: &[u8]) -> Option<(&str, &str, Vec<(String, String)>)>
Some((method, path, headers))
}
fn write_simple(stream: &mut TcpStream, code: u16, label: &str, body: &str) -> std::io::Result<()> {
fn write_simple(
stream: &mut TcpStream,
code: u16,
label: &str,
body: &str,
emit_body: bool,
) -> std::io::Result<()> {
write!(
stream,
"HTTP/1.1 {code} {label}\r\nContent-Type: text/plain; charset=utf-8\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
"HTTP/1.1 {code} {label}\r\nContent-Type: text/plain; charset=utf-8\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
body.len()
)
)?;
if emit_body {
stream.write_all(body.as_bytes())?;
}
Ok(())
}
fn mtime(feeds: &FeedStore, username: &str) -> i64 {
@ -312,4 +364,98 @@ mod tests {
}
assert_eq!(http_date(0), "Thu, 01 Jan 1970 00:00:00 GMT");
}
/// xorshift: deterministic pseudo-randomness, dependency-free.
fn next(seed: &mut u64) -> u64 {
*seed ^= *seed << 13;
*seed ^= *seed >> 7;
*seed ^= *seed << 17;
*seed
}
#[test]
fn head_serves_headers_without_a_body() {
let (_, addr) = with_feed("head");
let addr: std::net::SocketAddr = addr.parse().unwrap();
let response = get(&addr, "HEAD /twtxt/tobias.txt HTTP/1.1\r\n\r\n");
assert!(response.starts_with("HTTP/1.1 200 OK\r\n"), "{response}");
assert!(response.contains("Content-Length: "), "{response}");
assert_eq!(response.split("\r\n\r\n").nth(1), Some(""), "HEAD must not emit a body");
// The same feed via GET still carries one.
let full = get(&addr, "GET /twtxt/tobias.txt HTTP/1.1\r\n\r\n");
assert!(full.contains("# smol:host_key = "));
}
#[test]
fn the_connection_cap_sheds_floods_with_503() {
let (_, addr) = with_feed("cap");
let addr: std::net::SocketAddr = addr.parse().unwrap();
// Fill every slot with a partial request, as slowloris would.
let mut held: Vec<TcpStream> = Vec::new();
for _ in 0..MAX_CONNECTIONS {
let mut s = TcpStream::connect(addr).unwrap();
s.write_all(b"G").unwrap();
held.push(s);
}
std::thread::sleep(std::time::Duration::from_millis(500));
let mut extra = TcpStream::connect(addr).unwrap();
extra
.set_read_timeout(Some(std::time::Duration::from_secs(15)))
.unwrap();
extra.write_all(b"GET /twtxt/tobias.txt HTTP/1.1\r\n\r\n").unwrap();
// A single recv may split the response; read to the status line's end.
let mut buf = Vec::new();
while !buf.windows(2).any(|w| w == b"\r\n") {
let mut chunk = [0u8; 512];
let n = extra.read(&mut chunk).expect("503 response");
if n == 0 {
break;
}
buf.extend_from_slice(&chunk[..n]);
}
assert!(
buf.starts_with(b"HTTP/1.1 503"),
"the flood was absorbed instead of shed: {}",
String::from_utf8_lossy(&buf)
);
// Closing the flood frees the slots and the listener recovers.
drop(held);
let mut served = false;
for _ in 0..40 {
let response = get(&addr, "GET /twtxt/tobias.txt HTTP/1.1\r\n\r\n");
if response.starts_with("HTTP/1.1 200") {
served = true;
break;
}
std::thread::sleep(std::time::Duration::from_millis(250));
}
assert!(served, "the listener never recovered after the flood closed");
}
#[test]
fn request_parsing_never_panics_and_gates_usernames() {
let alphabet = [
"GET", "HEAD", " ", "/twtxt/", ".txt", "..", "%2F", "\u{0}", "\r\n", ":", "a", "-",
"\u{e9}", "\r\n\r\n", "smol:mail ",
];
let mut seed = 0x5eed_c0de_u64;
for _ in 0..2000 {
let n = (next(&mut seed) % 24) as usize;
let raw: String = (0..n)
.map(|_| alphabet[(next(&mut seed) % alphabet.len() as u64) as usize])
.collect();
let bytes = raw.as_bytes();
if let Some((_, path, _)) = parse_request(bytes) {
if let Some(name) = username_of(path) {
// Whatever the parser accepts must be path-safe.
assert!(crate::feedstore::valid_username(&name), "{name:?}");
assert!(!name.contains('/'));
assert!(!name.contains('\u{0}'));
}
}
}
}
}