From 38a8ae9845c8dcce58110be97eb6ab38c1f38205 Mon Sep 17 00:00:00 2001 From: hu13-test Date: Tue, 27 Oct 2026 15:47:14 +0000 Subject: [PATCH] bound serve threads, upload-pack, and rate-limiter memory (#63 #64 #65) #63: cap live connections at 64. Beyond that the accept loop answers 503 + Retry-After immediately instead of spawning more threads (inbox answers sleep ~500ms by design, so idle sockets were a trivial thread-exhaustion DoS). Also a 10s total deadline for reading a request head. A ConnSlot guard decrements on drop. #64: upload-pack is bounded to 4 concurrent runs, has a 120s hard timeout with SIGKILL, spools the pack to a temp file (no wait_with_output in RAM), streams it with Content-Length via respond_file, and always removes the temp file. Overload answers 503. #65: RateLimiter sweeps the whole map every 5min (previously only the accessed key was pruned, so IP rotation grew it forever) and fails closed at 100k distinct keys between sweeps. --- src/inbox.rs | 76 ++++++++++++++++++-- src/serve.rs | 189 ++++++++++++++++++++++++++++++++++++++++++++----- tests/abuse.rs | 48 +++++++++++++ 3 files changed, 293 insertions(+), 20 deletions(-) diff --git a/src/inbox.rs b/src/inbox.rs index a6cf081..f366558 100644 --- a/src/inbox.rs +++ b/src/inbox.rs @@ -53,7 +53,22 @@ pub const QUOTA_FILES: usize = 500; /// `429` when the bucket is empty. pub struct RateLimiter { limit_per_hour: u32, - buckets: std::sync::Mutex>>, + state: std::sync::Mutex, +} + +/// Seconds between full-map sweeps. Buckets are also pruned on access, +/// but only the accessed key; without a sweep, source-IP rotation grows +/// the map without bound (f0rge #65). +pub const RATE_SWEEP_SECS: u64 = 300; + +/// Hard cap on distinct keys between sweeps. At the cap a new key is +/// rejected (`429`) rather than inserted, so memory stays bounded even +/// under a flood (fail closed). +pub const RATE_MAX_KEYS: usize = 100_000; + +struct RateState { + buckets: std::collections::HashMap>, + last_sweep: u64, } impl RateLimiter { @@ -61,14 +76,27 @@ impl RateLimiter { pub fn new(limit_per_hour: u32) -> Self { Self { limit_per_hour, - buckets: std::sync::Mutex::new(std::collections::HashMap::new()), + state: std::sync::Mutex::new(RateState { + buckets: std::collections::HashMap::new(), + last_sweep: 0, + }), } } fn check_at(&self, ip: &str, date: &str, server_salt: &str, now: u64) -> bool { let key = salted_ip_hash(ip, date, server_salt); - let mut map = self.buckets.lock().unwrap_or_else(|e| e.into_inner()); - let entry = map.entry(key).or_default(); + let mut st = self.state.lock().unwrap_or_else(|e| e.into_inner()); + if now.saturating_sub(st.last_sweep) >= RATE_SWEEP_SECS { + st.buckets.retain(|_, times| { + times.retain(|&t| now.wrapping_sub(t) < 3600); + !times.is_empty() + }); + st.last_sweep = now; + } + if !st.buckets.contains_key(&key) && st.buckets.len() >= RATE_MAX_KEYS { + return false; + } + let entry = st.buckets.entry(key).or_default(); entry.retain(|&t| now.wrapping_sub(t) < 3600); if (entry.len() as u32) >= self.limit_per_hour { return false; @@ -995,6 +1023,46 @@ mod tests { assert!(lim.check_at("1.2.3.4", "2026-09-10", "other-salt", 1_005)); } + #[test] + fn rate_limiter_evicts_stale_keys() { + let lim = RateLimiter::new(10); + for i in 0..500 { + assert!(lim.check_at( + &format!("10.0.{}.{}", i / 250, i % 250), + "2026-09-10", + "s", + 1_000 + )); + } + assert_eq!(lim.state.lock().unwrap().buckets.len(), 500); + // Sweep interval elapsed: all 500 buckets are stale and dropped. + // > 1h later: the 500 old timestamps are outside the window. + assert!(lim.check_at("9.9.9.9", "2026-09-10", "s", 1_000 + 3_601)); + assert_eq!( + lim.state.lock().unwrap().buckets.len(), + 1, + "only the fresh bucket survives the sweep" + ); + } + + #[test] + fn rate_limiter_caps_key_count() { + let lim = RateLimiter::new(10); + // Bypass the sweep window so every key lands in the same map. + for i in 0..RATE_MAX_KEYS { + assert!(lim.check_at(&format!("ip-{i}"), "2026-09-10", "s", 1_000)); + } + assert_eq!(lim.state.lock().unwrap().buckets.len(), RATE_MAX_KEYS); + assert!( + !lim.check_at("one-more", "2026-09-10", "s", 1_000), + "new keys fail closed at the cap" + ); + assert!( + lim.check_at("ip-0", "2026-09-10", "s", 1_000), + "existing keys still work" + ); + } + #[test] fn quota_counts_files_until_drained() { let dir = std::env::temp_dir().join(format!( diff --git a/src/serve.rs b/src/serve.rs index 0b06ef3..dc71347 100644 --- a/src/serve.rs +++ b/src/serve.rs @@ -28,7 +28,7 @@ use std::io::{Read, Write}; use std::net::{TcpListener, TcpStream}; use std::path::{Path, PathBuf}; -use std::time::Duration; +use std::time::{Duration, Instant}; use crate::inbox; @@ -71,6 +71,21 @@ impl ServeOptions { } /// Per-connection shared state (tuned caps + rate buckets + daily salt). +/// Max concurrent connections. Each connection owns one thread and +/// inbox answers sleep ~500ms by design, so an unbounded accept loop is +/// a trivial thread-exhaustion DoS (f0rge #63). Over the cap: 503 now. +pub const MAX_CONNS: usize = 64; + +/// Max concurrent `git upload-pack` runs (f0rge #64). +pub const MAX_GIT_UPLOADS: usize = 4; + +/// Hard timeout for one `git upload-pack` run. Larger than the 10s +/// metadata timeout: a clone legitimately streams the whole pack. +pub const GIT_UPLOAD_TIMEOUT: Duration = Duration::from_secs(120); + +/// Total time allowed to read one request head (slowloris bound). +pub const HEAD_DEADLINE: Duration = Duration::from_secs(10); + struct Shared { www: PathBuf, inbox_dir: PathBuf, @@ -85,6 +100,8 @@ struct Shared { /// so `git clone http://host:port/repo.bundle` works (m6, issue #32). /// `None` when no bundle was present at startup (static-only mode). git_dir: Option, + /// In-flight upload-pack runs (bounded by [`MAX_GIT_UPLOADS`]). + git_uploads: std::sync::Arc, } /// Materialize `www/repo.bundle` into a bare repo for smart HTTP. @@ -225,12 +242,24 @@ pub fn run(opts: &ServeOptions) -> i32 { max_bytes: opts.max_bytes, quota: opts.quota, git_dir: prepare_git_dir(&opts.www), + git_uploads: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)), }); + let live = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); for conn in listener.incoming() { match conn { - Ok(stream) => { + Ok(mut stream) => { + use std::sync::atomic::Ordering; + if live.fetch_add(1, Ordering::Relaxed) >= MAX_CONNS { + live.fetch_sub(1, Ordering::Relaxed); + reject_overloaded(&mut stream); + continue; + } let shared = std::sync::Arc::clone(&shared); - std::thread::spawn(move || handle_conn(stream, &shared)); + let slot = ConnSlot(std::sync::Arc::clone(&live)); + std::thread::spawn(move || { + let _slot = slot; + handle_conn(stream, &shared); + }); } Err(e) => { eprintln!("hu13: accept: {e}"); @@ -243,6 +272,41 @@ pub fn run(opts: &ServeOptions) -> i32 { // ---- HTTP plumbing (tiny, no deps) ------------------------------------- +/// Decrements the live-connection counter on drop, whatever the thread +/// does (panic included). +struct ConnSlot(std::sync::Arc); + +impl Drop for ConnSlot { + fn drop(&mut self) { + self.0.fetch_sub(1, std::sync::atomic::Ordering::Relaxed); + } +} + +/// Answer 503 without spawning a thread (the pool is already full). +fn reject_overloaded(stream: &mut TcpStream) { + let _ = stream.set_write_timeout(Some(Duration::from_secs(2))); + let body = b"overloaded +"; + let head = format!( + "HTTP/1.1 503 Service Unavailable\r\nContent-Type: text/plain; charset=utf-8\r\nContent-Length: {}\r\nRetry-After: 1\r\nX-Content-Type-Options: nosniff\r\nConnection: close\r\n\r\n", + body.len() + ); + let _ = stream.write_all(head.as_bytes()); + let _ = stream.write_all(body); + let _ = stream.flush(); + // Drain the request bytes we never read: closing with unread data + // makes the kernel send RST, which can destroy the 503 the client + // has not read yet. Bounded by a short timeout, no thread. + let _ = stream.set_read_timeout(Some(Duration::from_millis(200))); + let mut buf = [0u8; 4096]; + for _ in 0..4 { + match stream.read(&mut buf) { + Ok(0) | Err(_) => break, + Ok(_) => {} + } + } +} + struct Head { method: String, target: String, @@ -252,10 +316,11 @@ struct Head { /// Read until `\r\n\r\n` (cap 16KiB of headers). `None` = stall/overflow/ /// EOF → caller drops the connection. fn read_head(stream: &mut TcpStream) -> Option<(Head, Vec)> { + let start = Instant::now(); let mut buf: Vec = Vec::with_capacity(4096); let mut chunk = [0u8; 4096]; loop { - if buf.len() > 16 * 1024 { + if buf.len() > 16 * 1024 || start.elapsed() > HEAD_DEADLINE { return None; } match stream.read(&mut chunk) { @@ -713,6 +778,56 @@ fn content_length(head: &Head) -> Option { /// `POST /repo.bundle/git-upload-pack`. Answer both from the bare repo /// materialized at startup (`Shared::git_dir`); anything else (wrong /// method, missing bundle, git failure) is 404/405 with no echo. +/// Bounded slot for one upload-pack run (f0rge #64). +struct GitSlot(std::sync::Arc); + +impl GitSlot { + fn acquire(shared: &Shared) -> Option { + use std::sync::atomic::Ordering; + let n = shared.git_uploads.fetch_add(1, Ordering::Relaxed); + if n >= MAX_GIT_UPLOADS { + shared.git_uploads.fetch_sub(1, Ordering::Relaxed); + return None; + } + Some(Self(std::sync::Arc::clone(&shared.git_uploads))) + } +} + +impl Drop for GitSlot { + fn drop(&mut self) { + self.0.fetch_sub(1, std::sync::atomic::Ordering::Relaxed); + } +} + +/// Stream a file body with `Content-Length` (no full-pack buffering). +fn respond_file( + stream: &mut TcpStream, + code: u16, + content_type: &str, + extra: &[(&str, &str)], + path: &Path, + len: u64, +) { + let mut out = format!( + "HTTP/1.1 {code} {}\r\nContent-Type: {content_type}\r\nContent-Length: {len}\r\nX-Content-Type-Options: nosniff\r\nConnection: close\r\n", + reason(code) + ); + for (k, v) in extra { + out.push_str(k); + out.push_str(": "); + out.push_str(v); + out.push_str("\r\n"); + } + out.push_str("\r\n"); + if stream.write_all(out.as_bytes()).is_err() { + return; + } + if let Ok(mut f) = std::fs::File::open(path) { + let _ = std::io::copy(&mut f, stream); + } + let _ = stream.flush(); +} + fn handle_git_smart( stream: &mut TcpStream, head: &Head, @@ -792,37 +907,79 @@ fn handle_git_smart( if req_body.len() != len { return; } + let Some(_slot) = GitSlot::acquire(shared) else { + static_error(stream, 503); + return; + }; + // Spool the pack to a temp file and stream it out: a big repo's pack + // used to sit in RAM in full (`wait_with_output`), and there was no + // deadline at all (f0rge #64). + let tmp = std::env::temp_dir().join(format!( + "hu13-pack-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0) + )); + let pack_file = match std::fs::File::create(&tmp) { + Ok(f) => f, + Err(_) => { + static_error(stream, 503); + return; + } + }; let mut child = match std::process::Command::new("git") .arg("upload-pack") .arg("--stateless-rpc") .arg(git_dir) .stdin(std::process::Stdio::piped()) - .stdout(std::process::Stdio::piped()) + .stdout(std::process::Stdio::from(pack_file)) .stderr(std::process::Stdio::null()) .spawn() { Ok(c) => c, Err(_) => { + let _ = std::fs::remove_file(&tmp); static_error(stream, 503); return; } }; if let Some(mut stdin) = child.stdin.take() { let _ = stdin.write_all(&req_body); + // Drop closes the pipe so upload-pack sees EOF. } - let out = child.wait_with_output(); - match out { - Ok(o) if o.status.success() => { - respond( - stream, - 200, - "application/x-git-upload-pack-result", - &[("Cache-Control", "no-cache")], - &o.stdout, - ); + let start = Instant::now(); + let status = loop { + match child.try_wait() { + Ok(Some(st)) => break Some(st), + Ok(None) => { + if start.elapsed() > GIT_UPLOAD_TIMEOUT { + let _ = child.kill(); + let _ = child.wait(); + break None; + } + std::thread::sleep(Duration::from_millis(10)); + } + Err(_) => break None, } - _ => static_error(stream, 503), + }; + let ok = status.is_some_and(|st| st.success()); + if !ok { + let _ = std::fs::remove_file(&tmp); + static_error(stream, 503); + return; } + let len = std::fs::metadata(&tmp).map(|m| m.len()).unwrap_or(0); + respond_file( + stream, + 200, + "application/x-git-upload-pack-result", + &[("Cache-Control", "no-cache")], + &tmp, + len, + ); + let _ = std::fs::remove_file(&tmp); } #[allow(clippy::too_many_lines)] diff --git a/tests/abuse.rs b/tests/abuse.rs index 9cca7d0..4e43707 100644 --- a/tests/abuse.rs +++ b/tests/abuse.rs @@ -505,3 +505,51 @@ fn caps_tunable_via_env() { ); assert!(srv.inbox_files().is_empty()); } + +#[test] +fn connection_cap_answers_503_instead_of_spawning_more_threads() { + // f0rge #63: MAX_CONNS=64 threads, each parked in read_head. The + // 65th connection must get an immediate 503, not another thread. + let srv = start_server(&[], &no_env()); + let mut idle = Vec::new(); + for _ in 0..64 { + match TcpStream::connect(&srv.addr) { + Ok(s) => idle.push(s), + Err(_) => break, + } + } + assert_eq!(idle.len(), 64, "opened the full pool of idle sockets"); + // Let the accept loop park all 64 worker threads on read_head. + std::thread::sleep(Duration::from_millis(300)); + + let start = Instant::now(); + let resp = transact( + &srv.addr, + b"GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n", + ); + let took = start.elapsed(); + assert!( + resp.starts_with("HTTP/1.1 503"), + "overloaded server answers 503, got: {resp}" + ); + assert!( + took < Duration::from_secs(2), + "503 is immediate (no ack sleep), took {took:?}" + ); + assert!( + resp.contains("Retry-After: 1"), + "retry hint present: {resp}" + ); + + // Dropping the idle sockets frees slots: the next request succeeds. + drop(idle); + std::thread::sleep(Duration::from_millis(300)); + let resp = transact( + &srv.addr, + b"GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n", + ); + assert!( + resp.starts_with("HTTP/1.1 200"), + "capacity is released when connections close, got: {resp}" + ); +} -- 2.56.0