#50 Bound serve threads, upload-pack, and rate-limiter memory (#63 #64 #65)

src/inbox.rs+72 −4
diff --git a/src/inbox.rs b/src/inbox.rsindex a6cf081..f366558 100644--- a/src/inbox.rs+++ b/src/inbox.rs@@ -53,7 +53,22 @@ pub const QUOTA_FILES: usize = 500;5353 /// `429` when the bucket is empty.5454 pub struct RateLimiter {5555     limit_per_hour: u32,56-    buckets: std::sync::Mutex<std::collections::HashMap<String, Vec<u64>>>,56+    state: std::sync::Mutex<RateState>,57+}58+59+/// Seconds between full-map sweeps. Buckets are also pruned on access,60+/// but only the accessed key; without a sweep, source-IP rotation grows61+/// the map without bound (f0rge #65).62+pub const RATE_SWEEP_SECS: u64 = 300;63+64+/// Hard cap on distinct keys between sweeps. At the cap a new key is65+/// rejected (`429`) rather than inserted, so memory stays bounded even66+/// under a flood (fail closed).67+pub const RATE_MAX_KEYS: usize = 100_000;68+69+struct RateState {70+    buckets: std::collections::HashMap<String, Vec<u64>>,71+    last_sweep: u64,5772 }5873 5974 impl RateLimiter {@@ -61,14 +76,27 @@ impl RateLimiter {6176     pub fn new(limit_per_hour: u32) -> Self {6277         Self {6378             limit_per_hour,64-            buckets: std::sync::Mutex::new(std::collections::HashMap::new()),79+            state: std::sync::Mutex::new(RateState {80+                buckets: std::collections::HashMap::new(),81+                last_sweep: 0,82+            }),6583         }6684     }6785 6886     fn check_at(&self, ip: &str, date: &str, server_salt: &str, now: u64) -> bool {6987         let key = salted_ip_hash(ip, date, server_salt);70-        let mut map = self.buckets.lock().unwrap_or_else(|e| e.into_inner());71-        let entry = map.entry(key).or_default();88+        let mut st = self.state.lock().unwrap_or_else(|e| e.into_inner());89+        if now.saturating_sub(st.last_sweep) >= RATE_SWEEP_SECS {90+            st.buckets.retain(|_, times| {91+                times.retain(|&t| now.wrapping_sub(t) < 3600);92+                !times.is_empty()93+            });94+            st.last_sweep = now;95+        }96+        if !st.buckets.contains_key(&key) && st.buckets.len() >= RATE_MAX_KEYS {97+            return false;98+        }99+        let entry = st.buckets.entry(key).or_default();72100         entry.retain(|&t| now.wrapping_sub(t) < 3600);73101         if (entry.len() as u32) >= self.limit_per_hour {74102             return false;@@ -995,6 +1023,46 @@ mod tests {9951023         assert!(lim.check_at("1.2.3.4", "2026-09-10", "other-salt", 1_005));9961024     }9971025 1026+    #[test]1027+    fn rate_limiter_evicts_stale_keys() {1028+        let lim = RateLimiter::new(10);1029+        for i in 0..500 {1030+            assert!(lim.check_at(1031+                &format!("10.0.{}.{}", i / 250, i % 250),1032+                "2026-09-10",1033+                "s",1034+                1_0001035+            ));1036+        }1037+        assert_eq!(lim.state.lock().unwrap().buckets.len(), 500);1038+        // Sweep interval elapsed: all 500 buckets are stale and dropped.1039+        // > 1h later: the 500 old timestamps are outside the window.1040+        assert!(lim.check_at("9.9.9.9", "2026-09-10", "s", 1_000 + 3_601));1041+        assert_eq!(1042+            lim.state.lock().unwrap().buckets.len(),1043+            1,1044+            "only the fresh bucket survives the sweep"1045+        );1046+    }1047+1048+    #[test]1049+    fn rate_limiter_caps_key_count() {1050+        let lim = RateLimiter::new(10);1051+        // Bypass the sweep window so every key lands in the same map.1052+        for i in 0..RATE_MAX_KEYS {1053+            assert!(lim.check_at(&format!("ip-{i}"), "2026-09-10", "s", 1_000));1054+        }1055+        assert_eq!(lim.state.lock().unwrap().buckets.len(), RATE_MAX_KEYS);1056+        assert!(1057+            !lim.check_at("one-more", "2026-09-10", "s", 1_000),1058+            "new keys fail closed at the cap"1059+        );1060+        assert!(1061+            lim.check_at("ip-0", "2026-09-10", "s", 1_000),1062+            "existing keys still work"1063+        );1064+    }1065+9981066     #[test]9991067     fn quota_counts_files_until_drained() {10001068         let dir = std::env::temp_dir().join(format!(
src/serve.rs+173 −16
diff --git a/src/serve.rs b/src/serve.rsindex 0b06ef3..dc71347 100644--- a/src/serve.rs+++ b/src/serve.rs@@ -28,7 +28,7 @@2828 use std::io::{Read, Write};2929 use std::net::{TcpListener, TcpStream};3030 use std::path::{Path, PathBuf};31-use std::time::Duration;31+use std::time::{Duration, Instant};3232 3333 use crate::inbox;3434 @@ -71,6 +71,21 @@ impl ServeOptions {7171 }7272 7373 /// Per-connection shared state (tuned caps + rate buckets + daily salt).74+/// Max concurrent connections. Each connection owns one thread and75+/// inbox answers sleep ~500ms by design, so an unbounded accept loop is76+/// a trivial thread-exhaustion DoS (f0rge #63). Over the cap: 503 now.77+pub const MAX_CONNS: usize = 64;78+79+/// Max concurrent `git upload-pack` runs (f0rge #64).80+pub const MAX_GIT_UPLOADS: usize = 4;81+82+/// Hard timeout for one `git upload-pack` run. Larger than the 10s83+/// metadata timeout: a clone legitimately streams the whole pack.84+pub const GIT_UPLOAD_TIMEOUT: Duration = Duration::from_secs(120);85+86+/// Total time allowed to read one request head (slowloris bound).87+pub const HEAD_DEADLINE: Duration = Duration::from_secs(10);88+7489 struct Shared {7590     www: PathBuf,7691     inbox_dir: PathBuf,@@ -85,6 +100,8 @@ struct Shared {85100     /// so `git clone http://host:port/repo.bundle` works (m6, issue #32).86101     /// `None` when no bundle was present at startup (static-only mode).87102     git_dir: Option<PathBuf>,103+    /// In-flight upload-pack runs (bounded by [`MAX_GIT_UPLOADS`]).104+    git_uploads: std::sync::Arc<std::sync::atomic::AtomicUsize>,88105 }89106 90107 /// Materialize `www/repo.bundle` into a bare repo for smart HTTP.@@ -225,12 +242,24 @@ pub fn run(opts: &ServeOptions) -> i32 {225242         max_bytes: opts.max_bytes,226243         quota: opts.quota,227244         git_dir: prepare_git_dir(&opts.www),245+        git_uploads: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)),228246     });247+    let live = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));229248     for conn in listener.incoming() {230249         match conn {231-            Ok(stream) => {250+            Ok(mut stream) => {251+                use std::sync::atomic::Ordering;252+                if live.fetch_add(1, Ordering::Relaxed) >= MAX_CONNS {253+                    live.fetch_sub(1, Ordering::Relaxed);254+                    reject_overloaded(&mut stream);255+                    continue;256+                }232257                 let shared = std::sync::Arc::clone(&shared);233-                std::thread::spawn(move || handle_conn(stream, &shared));258+                let slot = ConnSlot(std::sync::Arc::clone(&live));259+                std::thread::spawn(move || {260+                    let _slot = slot;261+                    handle_conn(stream, &shared);262+                });234263             }235264             Err(e) => {236265                 eprintln!("hu13: accept: {e}");@@ -243,6 +272,41 @@ pub fn run(opts: &ServeOptions) -> i32 {243272 244273 // ---- HTTP plumbing (tiny, no deps) -------------------------------------245274 275+/// Decrements the live-connection counter on drop, whatever the thread276+/// does (panic included).277+struct ConnSlot(std::sync::Arc<std::sync::atomic::AtomicUsize>);278+279+impl Drop for ConnSlot {280+    fn drop(&mut self) {281+        self.0.fetch_sub(1, std::sync::atomic::Ordering::Relaxed);282+    }283+}284+285+/// Answer 503 without spawning a thread (the pool is already full).286+fn reject_overloaded(stream: &mut TcpStream) {287+    let _ = stream.set_write_timeout(Some(Duration::from_secs(2)));288+    let body = b"overloaded289+";290+    let head = format!(291+        "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",292+        body.len()293+    );294+    let _ = stream.write_all(head.as_bytes());295+    let _ = stream.write_all(body);296+    let _ = stream.flush();297+    // Drain the request bytes we never read: closing with unread data298+    // makes the kernel send RST, which can destroy the 503 the client299+    // has not read yet. Bounded by a short timeout, no thread.300+    let _ = stream.set_read_timeout(Some(Duration::from_millis(200)));301+    let mut buf = [0u8; 4096];302+    for _ in 0..4 {303+        match stream.read(&mut buf) {304+            Ok(0) | Err(_) => break,305+            Ok(_) => {}306+        }307+    }308+}309+246310 struct Head {247311     method: String,248312     target: String,@@ -252,10 +316,11 @@ struct Head {252316 /// Read until `\r\n\r\n` (cap 16KiB of headers). `None` = stall/overflow/253317 /// EOF → caller drops the connection.254318 fn read_head(stream: &mut TcpStream) -> Option<(Head, Vec<u8>)> {319+    let start = Instant::now();255320     let mut buf: Vec<u8> = Vec::with_capacity(4096);256321     let mut chunk = [0u8; 4096];257322     loop {258-        if buf.len() > 16 * 1024 {323+        if buf.len() > 16 * 1024 || start.elapsed() > HEAD_DEADLINE {259324             return None;260325         }261326         match stream.read(&mut chunk) {@@ -713,6 +778,56 @@ fn content_length(head: &Head) -> Option<usize> {713778 /// `POST /repo.bundle/git-upload-pack`. Answer both from the bare repo714779 /// materialized at startup (`Shared::git_dir`); anything else (wrong715780 /// method, missing bundle, git failure) is 404/405 with no echo.781+/// Bounded slot for one upload-pack run (f0rge #64).782+struct GitSlot(std::sync::Arc<std::sync::atomic::AtomicUsize>);783+784+impl GitSlot {785+    fn acquire(shared: &Shared) -> Option<Self> {786+        use std::sync::atomic::Ordering;787+        let n = shared.git_uploads.fetch_add(1, Ordering::Relaxed);788+        if n >= MAX_GIT_UPLOADS {789+            shared.git_uploads.fetch_sub(1, Ordering::Relaxed);790+            return None;791+        }792+        Some(Self(std::sync::Arc::clone(&shared.git_uploads)))793+    }794+}795+796+impl Drop for GitSlot {797+    fn drop(&mut self) {798+        self.0.fetch_sub(1, std::sync::atomic::Ordering::Relaxed);799+    }800+}801+802+/// Stream a file body with `Content-Length` (no full-pack buffering).803+fn respond_file(804+    stream: &mut TcpStream,805+    code: u16,806+    content_type: &str,807+    extra: &[(&str, &str)],808+    path: &Path,809+    len: u64,810+) {811+    let mut out = format!(812+        "HTTP/1.1 {code} {}\r\nContent-Type: {content_type}\r\nContent-Length: {len}\r\nX-Content-Type-Options: nosniff\r\nConnection: close\r\n",813+        reason(code)814+    );815+    for (k, v) in extra {816+        out.push_str(k);817+        out.push_str(": ");818+        out.push_str(v);819+        out.push_str("\r\n");820+    }821+    out.push_str("\r\n");822+    if stream.write_all(out.as_bytes()).is_err() {823+        return;824+    }825+    if let Ok(mut f) = std::fs::File::open(path) {826+        let _ = std::io::copy(&mut f, stream);827+    }828+    let _ = stream.flush();829+}830+716831 fn handle_git_smart(717832     stream: &mut TcpStream,718833     head: &Head,@@ -792,37 +907,79 @@ fn handle_git_smart(792907     if req_body.len() != len {793908         return;794909     }910+    let Some(_slot) = GitSlot::acquire(shared) else {911+        static_error(stream, 503);912+        return;913+    };914+    // Spool the pack to a temp file and stream it out: a big repo's pack915+    // used to sit in RAM in full (`wait_with_output`), and there was no916+    // deadline at all (f0rge #64).917+    let tmp = std::env::temp_dir().join(format!(918+        "hu13-pack-{}-{}",919+        std::process::id(),920+        std::time::SystemTime::now()921+            .duration_since(std::time::UNIX_EPOCH)922+            .map(|d| d.as_nanos())923+            .unwrap_or(0)924+    ));925+    let pack_file = match std::fs::File::create(&tmp) {926+        Ok(f) => f,927+        Err(_) => {928+            static_error(stream, 503);929+            return;930+        }931+    };795932     let mut child = match std::process::Command::new("git")796933         .arg("upload-pack")797934         .arg("--stateless-rpc")798935         .arg(git_dir)799936         .stdin(std::process::Stdio::piped())800-        .stdout(std::process::Stdio::piped())937+        .stdout(std::process::Stdio::from(pack_file))801938         .stderr(std::process::Stdio::null())802939         .spawn()803940     {804941         Ok(c) => c,805942         Err(_) => {943+            let _ = std::fs::remove_file(&tmp);806944             static_error(stream, 503);807945             return;808946         }809947     };810948     if let Some(mut stdin) = child.stdin.take() {811949         let _ = stdin.write_all(&req_body);950+        // Drop closes the pipe so upload-pack sees EOF.812951     }813-    let out = child.wait_with_output();814-    match out {815-        Ok(o) if o.status.success() => {816-            respond(817-                stream,818-                200,819-                "application/x-git-upload-pack-result",820-                &[("Cache-Control", "no-cache")],821-                &o.stdout,822-            );952+    let start = Instant::now();953+    let status = loop {954+        match child.try_wait() {955+            Ok(Some(st)) => break Some(st),956+            Ok(None) => {957+                if start.elapsed() > GIT_UPLOAD_TIMEOUT {958+                    let _ = child.kill();959+                    let _ = child.wait();960+                    break None;961+                }962+                std::thread::sleep(Duration::from_millis(10));963+            }964+            Err(_) => break None,823965         }824-        _ => static_error(stream, 503),966+    };967+    let ok = status.is_some_and(|st| st.success());968+    if !ok {969+        let _ = std::fs::remove_file(&tmp);970+        static_error(stream, 503);971+        return;825972     }973+    let len = std::fs::metadata(&tmp).map(|m| m.len()).unwrap_or(0);974+    respond_file(975+        stream,976+        200,977+        "application/x-git-upload-pack-result",978+        &[("Cache-Control", "no-cache")],979+        &tmp,980+        len,981+    );982+    let _ = std::fs::remove_file(&tmp);826983 }827984 828985 #[allow(clippy::too_many_lines)]
tests/abuse.rs+48 −1
diff --git a/tests/abuse.rs b/tests/abuse.rsindex 9cca7d0..4e43707 100644--- a/tests/abuse.rs+++ b/tests/abuse.rs@@ -505,3 +505,51 @@ fn caps_tunable_via_env() {505505     );506506     assert!(srv.inbox_files().is_empty());507507 }508+509+#[test]510+fn connection_cap_answers_503_instead_of_spawning_more_threads() {511+    // f0rge #63: MAX_CONNS=64 threads, each parked in read_head. The512+    // 65th connection must get an immediate 503, not another thread.513+    let srv = start_server(&[], &no_env());514+    let mut idle = Vec::new();515+    for _ in 0..64 {516+        match TcpStream::connect(&srv.addr) {517+            Ok(s) => idle.push(s),518+            Err(_) => break,519+        }520+    }521+    assert_eq!(idle.len(), 64, "opened the full pool of idle sockets");522+    // Let the accept loop park all 64 worker threads on read_head.523+    std::thread::sleep(Duration::from_millis(300));524+525+    let start = Instant::now();526+    let resp = transact(527+        &srv.addr,528+        b"GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n",529+    );530+    let took = start.elapsed();531+    assert!(532+        resp.starts_with("HTTP/1.1 503"),533+        "overloaded server answers 503, got: {resp}"534+    );535+    assert!(536+        took < Duration::from_secs(2),537+        "503 is immediate (no ack sleep), took {took:?}"538+    );539+    assert!(540+        resp.contains("Retry-After: 1"),541+        "retry hint present: {resp}"542+    );543+544+    // Dropping the idle sockets frees slots: the next request succeeds.545+    drop(idle);546+    std::thread::sleep(Duration::from_millis(300));547+    let resp = transact(548+        &srv.addr,549+        b"GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n",550+    );551+    assert!(552+        resp.starts_with("HTTP/1.1 200"),553+        "capacity is released when connections close, got: {resp}"554+    );555+}508-- 2.56.0