1 From 38a8ae9845c8dcce58110be97eb6ab38c1f38205 Mon Sep 17 00:00:00 20012 From: hu13-test <[email protected]>3 Date: Tue, 27 Oct 2026 15:47:14 +00004 Subject: [PATCH] bound serve threads, upload-pack, and rate-limiter memory5  (#63 #64 #65)6 7 #63: cap live connections at 64. Beyond that the accept loop answers8 503 + Retry-After immediately instead of spawning more threads9 (inbox answers sleep ~500ms by design, so idle sockets were a10 trivial thread-exhaustion DoS). Also a 10s total deadline for11 reading a request head. A ConnSlot guard decrements on drop.12 13 #64: upload-pack is bounded to 4 concurrent runs, has a 120s hard14 timeout with SIGKILL, spools the pack to a temp file (no15 wait_with_output in RAM), streams it with Content-Length via16 respond_file, and always removes the temp file. Overload answers 503.17 18 #65: RateLimiter sweeps the whole map every 5min (previously only19 the accessed key was pruned, so IP rotation grew it forever) and20 fails closed at 100k distinct keys between sweeps.21 ---22  src/inbox.rs   |  76 ++++++++++++++++++--23  src/serve.rs   | 189 ++++++++++++++++++++++++++++++++++++++++++++-----24  tests/abuse.rs |  48 +++++++++++++25  3 files changed, 293 insertions(+), 20 deletions(-)26 27 diff --git a/src/inbox.rs b/src/inbox.rs28 index a6cf081..f366558 10064429 --- a/src/inbox.rs30 +++ b/src/inbox.rs31 @@ -53,7 +53,22 @@ pub const QUOTA_FILES: usize = 500;32  /// `429` when the bucket is empty.33  pub struct RateLimiter {34      limit_per_hour: u32,35 -    buckets: std::sync::Mutex<std::collections::HashMap<String, Vec<u64>>>,36 +    state: std::sync::Mutex<RateState>,37 +}38 +39 +/// Seconds between full-map sweeps. Buckets are also pruned on access,40 +/// but only the accessed key; without a sweep, source-IP rotation grows41 +/// the map without bound (f0rge #65).42 +pub const RATE_SWEEP_SECS: u64 = 300;43 +44 +/// Hard cap on distinct keys between sweeps. At the cap a new key is45 +/// rejected (`429`) rather than inserted, so memory stays bounded even46 +/// under a flood (fail closed).47 +pub const RATE_MAX_KEYS: usize = 100_000;48 +49 +struct RateState {50 +    buckets: std::collections::HashMap<String, Vec<u64>>,51 +    last_sweep: u64,52  }53  54  impl RateLimiter {55 @@ -61,14 +76,27 @@ impl RateLimiter {56      pub fn new(limit_per_hour: u32) -> Self {57          Self {58              limit_per_hour,59 -            buckets: std::sync::Mutex::new(std::collections::HashMap::new()),60 +            state: std::sync::Mutex::new(RateState {61 +                buckets: std::collections::HashMap::new(),62 +                last_sweep: 0,63 +            }),64          }65      }66  67      fn check_at(&self, ip: &str, date: &str, server_salt: &str, now: u64) -> bool {68          let key = salted_ip_hash(ip, date, server_salt);69 -        let mut map = self.buckets.lock().unwrap_or_else(|e| e.into_inner());70 -        let entry = map.entry(key).or_default();71 +        let mut st = self.state.lock().unwrap_or_else(|e| e.into_inner());72 +        if now.saturating_sub(st.last_sweep) >= RATE_SWEEP_SECS {73 +            st.buckets.retain(|_, times| {74 +                times.retain(|&t| now.wrapping_sub(t) < 3600);75 +                !times.is_empty()76 +            });77 +            st.last_sweep = now;78 +        }79 +        if !st.buckets.contains_key(&key) && st.buckets.len() >= RATE_MAX_KEYS {80 +            return false;81 +        }82 +        let entry = st.buckets.entry(key).or_default();83          entry.retain(|&t| now.wrapping_sub(t) < 3600);84          if (entry.len() as u32) >= self.limit_per_hour {85              return false;86 @@ -995,6 +1023,46 @@ mod tests {87          assert!(lim.check_at("1.2.3.4", "2026-09-10", "other-salt", 1_005));88      }89  90 +    #[test]91 +    fn rate_limiter_evicts_stale_keys() {92 +        let lim = RateLimiter::new(10);93 +        for i in 0..500 {94 +            assert!(lim.check_at(95 +                &format!("10.0.{}.{}", i / 250, i % 250),96 +                "2026-09-10",97 +                "s",98 +                1_00099 +            ));100 +        }101 +        assert_eq!(lim.state.lock().unwrap().buckets.len(), 500);102 +        // Sweep interval elapsed: all 500 buckets are stale and dropped.103 +        // > 1h later: the 500 old timestamps are outside the window.104 +        assert!(lim.check_at("9.9.9.9", "2026-09-10", "s", 1_000 + 3_601));105 +        assert_eq!(106 +            lim.state.lock().unwrap().buckets.len(),107 +            1,108 +            "only the fresh bucket survives the sweep"109 +        );110 +    }111 +112 +    #[test]113 +    fn rate_limiter_caps_key_count() {114 +        let lim = RateLimiter::new(10);115 +        // Bypass the sweep window so every key lands in the same map.116 +        for i in 0..RATE_MAX_KEYS {117 +            assert!(lim.check_at(&format!("ip-{i}"), "2026-09-10", "s", 1_000));118 +        }119 +        assert_eq!(lim.state.lock().unwrap().buckets.len(), RATE_MAX_KEYS);120 +        assert!(121 +            !lim.check_at("one-more", "2026-09-10", "s", 1_000),122 +            "new keys fail closed at the cap"123 +        );124 +        assert!(125 +            lim.check_at("ip-0", "2026-09-10", "s", 1_000),126 +            "existing keys still work"127 +        );128 +    }129 +130      #[test]131      fn quota_counts_files_until_drained() {132          let dir = std::env::temp_dir().join(format!(133 diff --git a/src/serve.rs b/src/serve.rs134 index 0b06ef3..dc71347 100644135 --- a/src/serve.rs136 +++ b/src/serve.rs137 @@ -28,7 +28,7 @@138  use std::io::{Read, Write};139  use std::net::{TcpListener, TcpStream};140  use std::path::{Path, PathBuf};141 -use std::time::Duration;142 +use std::time::{Duration, Instant};143  144  use crate::inbox;145  146 @@ -71,6 +71,21 @@ impl ServeOptions {147  }148  149  /// Per-connection shared state (tuned caps + rate buckets + daily salt).150 +/// Max concurrent connections. Each connection owns one thread and151 +/// inbox answers sleep ~500ms by design, so an unbounded accept loop is152 +/// a trivial thread-exhaustion DoS (f0rge #63). Over the cap: 503 now.153 +pub const MAX_CONNS: usize = 64;154 +155 +/// Max concurrent `git upload-pack` runs (f0rge #64).156 +pub const MAX_GIT_UPLOADS: usize = 4;157 +158 +/// Hard timeout for one `git upload-pack` run. Larger than the 10s159 +/// metadata timeout: a clone legitimately streams the whole pack.160 +pub const GIT_UPLOAD_TIMEOUT: Duration = Duration::from_secs(120);161 +162 +/// Total time allowed to read one request head (slowloris bound).163 +pub const HEAD_DEADLINE: Duration = Duration::from_secs(10);164 +165  struct Shared {166      www: PathBuf,167      inbox_dir: PathBuf,168 @@ -85,6 +100,8 @@ struct Shared {169      /// so `git clone http://host:port/repo.bundle` works (m6, issue #32).170      /// `None` when no bundle was present at startup (static-only mode).171      git_dir: Option<PathBuf>,172 +    /// In-flight upload-pack runs (bounded by [`MAX_GIT_UPLOADS`]).173 +    git_uploads: std::sync::Arc<std::sync::atomic::AtomicUsize>,174  }175  176  /// Materialize `www/repo.bundle` into a bare repo for smart HTTP.177 @@ -225,12 +242,24 @@ pub fn run(opts: &ServeOptions) -> i32 {178          max_bytes: opts.max_bytes,179          quota: opts.quota,180          git_dir: prepare_git_dir(&opts.www),181 +        git_uploads: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)),182      });183 +    let live = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));184      for conn in listener.incoming() {185          match conn {186 -            Ok(stream) => {187 +            Ok(mut stream) => {188 +                use std::sync::atomic::Ordering;189 +                if live.fetch_add(1, Ordering::Relaxed) >= MAX_CONNS {190 +                    live.fetch_sub(1, Ordering::Relaxed);191 +                    reject_overloaded(&mut stream);192 +                    continue;193 +                }194                  let shared = std::sync::Arc::clone(&shared);195 -                std::thread::spawn(move || handle_conn(stream, &shared));196 +                let slot = ConnSlot(std::sync::Arc::clone(&live));197 +                std::thread::spawn(move || {198 +                    let _slot = slot;199 +                    handle_conn(stream, &shared);200 +                });201              }202              Err(e) => {203                  eprintln!("hu13: accept: {e}");204 @@ -243,6 +272,41 @@ pub fn run(opts: &ServeOptions) -> i32 {205  206  // ---- HTTP plumbing (tiny, no deps) -------------------------------------207  208 +/// Decrements the live-connection counter on drop, whatever the thread209 +/// does (panic included).210 +struct ConnSlot(std::sync::Arc<std::sync::atomic::AtomicUsize>);211 +212 +impl Drop for ConnSlot {213 +    fn drop(&mut self) {214 +        self.0.fetch_sub(1, std::sync::atomic::Ordering::Relaxed);215 +    }216 +}217 +218 +/// Answer 503 without spawning a thread (the pool is already full).219 +fn reject_overloaded(stream: &mut TcpStream) {220 +    let _ = stream.set_write_timeout(Some(Duration::from_secs(2)));221 +    let body = b"overloaded222 +";223 +    let head = format!(224 +        "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",225 +        body.len()226 +    );227 +    let _ = stream.write_all(head.as_bytes());228 +    let _ = stream.write_all(body);229 +    let _ = stream.flush();230 +    // Drain the request bytes we never read: closing with unread data231 +    // makes the kernel send RST, which can destroy the 503 the client232 +    // has not read yet. Bounded by a short timeout, no thread.233 +    let _ = stream.set_read_timeout(Some(Duration::from_millis(200)));234 +    let mut buf = [0u8; 4096];235 +    for _ in 0..4 {236 +        match stream.read(&mut buf) {237 +            Ok(0) | Err(_) => break,238 +            Ok(_) => {}239 +        }240 +    }241 +}242 +243  struct Head {244      method: String,245      target: String,246 @@ -252,10 +316,11 @@ struct Head {247  /// Read until `\r\n\r\n` (cap 16KiB of headers). `None` = stall/overflow/248  /// EOF → caller drops the connection.249  fn read_head(stream: &mut TcpStream) -> Option<(Head, Vec<u8>)> {250 +    let start = Instant::now();251      let mut buf: Vec<u8> = Vec::with_capacity(4096);252      let mut chunk = [0u8; 4096];253      loop {254 -        if buf.len() > 16 * 1024 {255 +        if buf.len() > 16 * 1024 || start.elapsed() > HEAD_DEADLINE {256              return None;257          }258          match stream.read(&mut chunk) {259 @@ -713,6 +778,56 @@ fn content_length(head: &Head) -> Option<usize> {260  /// `POST /repo.bundle/git-upload-pack`. Answer both from the bare repo261  /// materialized at startup (`Shared::git_dir`); anything else (wrong262  /// method, missing bundle, git failure) is 404/405 with no echo.263 +/// Bounded slot for one upload-pack run (f0rge #64).264 +struct GitSlot(std::sync::Arc<std::sync::atomic::AtomicUsize>);265 +266 +impl GitSlot {267 +    fn acquire(shared: &Shared) -> Option<Self> {268 +        use std::sync::atomic::Ordering;269 +        let n = shared.git_uploads.fetch_add(1, Ordering::Relaxed);270 +        if n >= MAX_GIT_UPLOADS {271 +            shared.git_uploads.fetch_sub(1, Ordering::Relaxed);272 +            return None;273 +        }274 +        Some(Self(std::sync::Arc::clone(&shared.git_uploads)))275 +    }276 +}277 +278 +impl Drop for GitSlot {279 +    fn drop(&mut self) {280 +        self.0.fetch_sub(1, std::sync::atomic::Ordering::Relaxed);281 +    }282 +}283 +284 +/// Stream a file body with `Content-Length` (no full-pack buffering).285 +fn respond_file(286 +    stream: &mut TcpStream,287 +    code: u16,288 +    content_type: &str,289 +    extra: &[(&str, &str)],290 +    path: &Path,291 +    len: u64,292 +) {293 +    let mut out = format!(294 +        "HTTP/1.1 {code} {}\r\nContent-Type: {content_type}\r\nContent-Length: {len}\r\nX-Content-Type-Options: nosniff\r\nConnection: close\r\n",295 +        reason(code)296 +    );297 +    for (k, v) in extra {298 +        out.push_str(k);299 +        out.push_str(": ");300 +        out.push_str(v);301 +        out.push_str("\r\n");302 +    }303 +    out.push_str("\r\n");304 +    if stream.write_all(out.as_bytes()).is_err() {305 +        return;306 +    }307 +    if let Ok(mut f) = std::fs::File::open(path) {308 +        let _ = std::io::copy(&mut f, stream);309 +    }310 +    let _ = stream.flush();311 +}312 +313  fn handle_git_smart(314      stream: &mut TcpStream,315      head: &Head,316 @@ -792,37 +907,79 @@ fn handle_git_smart(317      if req_body.len() != len {318          return;319      }320 +    let Some(_slot) = GitSlot::acquire(shared) else {321 +        static_error(stream, 503);322 +        return;323 +    };324 +    // Spool the pack to a temp file and stream it out: a big repo's pack325 +    // used to sit in RAM in full (`wait_with_output`), and there was no326 +    // deadline at all (f0rge #64).327 +    let tmp = std::env::temp_dir().join(format!(328 +        "hu13-pack-{}-{}",329 +        std::process::id(),330 +        std::time::SystemTime::now()331 +            .duration_since(std::time::UNIX_EPOCH)332 +            .map(|d| d.as_nanos())333 +            .unwrap_or(0)334 +    ));335 +    let pack_file = match std::fs::File::create(&tmp) {336 +        Ok(f) => f,337 +        Err(_) => {338 +            static_error(stream, 503);339 +            return;340 +        }341 +    };342      let mut child = match std::process::Command::new("git")343          .arg("upload-pack")344          .arg("--stateless-rpc")345          .arg(git_dir)346          .stdin(std::process::Stdio::piped())347 -        .stdout(std::process::Stdio::piped())348 +        .stdout(std::process::Stdio::from(pack_file))349          .stderr(std::process::Stdio::null())350          .spawn()351      {352          Ok(c) => c,353          Err(_) => {354 +            let _ = std::fs::remove_file(&tmp);355              static_error(stream, 503);356              return;357          }358      };359      if let Some(mut stdin) = child.stdin.take() {360          let _ = stdin.write_all(&req_body);361 +        // Drop closes the pipe so upload-pack sees EOF.362      }363 -    let out = child.wait_with_output();364 -    match out {365 -        Ok(o) if o.status.success() => {366 -            respond(367 -                stream,368 -                200,369 -                "application/x-git-upload-pack-result",370 -                &[("Cache-Control", "no-cache")],371 -                &o.stdout,372 -            );373 +    let start = Instant::now();374 +    let status = loop {375 +        match child.try_wait() {376 +            Ok(Some(st)) => break Some(st),377 +            Ok(None) => {378 +                if start.elapsed() > GIT_UPLOAD_TIMEOUT {379 +                    let _ = child.kill();380 +                    let _ = child.wait();381 +                    break None;382 +                }383 +                std::thread::sleep(Duration::from_millis(10));384 +            }385 +            Err(_) => break None,386          }387 -        _ => static_error(stream, 503),388 +    };389 +    let ok = status.is_some_and(|st| st.success());390 +    if !ok {391 +        let _ = std::fs::remove_file(&tmp);392 +        static_error(stream, 503);393 +        return;394      }395 +    let len = std::fs::metadata(&tmp).map(|m| m.len()).unwrap_or(0);396 +    respond_file(397 +        stream,398 +        200,399 +        "application/x-git-upload-pack-result",400 +        &[("Cache-Control", "no-cache")],401 +        &tmp,402 +        len,403 +    );404 +    let _ = std::fs::remove_file(&tmp);405  }406  407  #[allow(clippy::too_many_lines)]408 diff --git a/tests/abuse.rs b/tests/abuse.rs409 index 9cca7d0..4e43707 100644410 --- a/tests/abuse.rs411 +++ b/tests/abuse.rs412 @@ -505,3 +505,51 @@ fn caps_tunable_via_env() {413      );414      assert!(srv.inbox_files().is_empty());415  }416 +417 +#[test]418 +fn connection_cap_answers_503_instead_of_spawning_more_threads() {419 +    // f0rge #63: MAX_CONNS=64 threads, each parked in read_head. The420 +    // 65th connection must get an immediate 503, not another thread.421 +    let srv = start_server(&[], &no_env());422 +    let mut idle = Vec::new();423 +    for _ in 0..64 {424 +        match TcpStream::connect(&srv.addr) {425 +            Ok(s) => idle.push(s),426 +            Err(_) => break,427 +        }428 +    }429 +    assert_eq!(idle.len(), 64, "opened the full pool of idle sockets");430 +    // Let the accept loop park all 64 worker threads on read_head.431 +    std::thread::sleep(Duration::from_millis(300));432 +433 +    let start = Instant::now();434 +    let resp = transact(435 +        &srv.addr,436 +        b"GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n",437 +    );438 +    let took = start.elapsed();439 +    assert!(440 +        resp.starts_with("HTTP/1.1 503"),441 +        "overloaded server answers 503, got: {resp}"442 +    );443 +    assert!(444 +        took < Duration::from_secs(2),445 +        "503 is immediate (no ack sleep), took {took:?}"446 +    );447 +    assert!(448 +        resp.contains("Retry-After: 1"),449 +        "retry hint present: {resp}"450 +    );451 +452 +    // Dropping the idle sockets frees slots: the next request succeeds.453 +    drop(idle);454 +    std::thread::sleep(Duration::from_millis(300));455 +    let resp = transact(456 +        &srv.addr,457 +        b"GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n",458 +    );459 +    assert!(460 +        resp.starts_with("HTTP/1.1 200"),461 +        "capacity is released when connections close, got: {resp}"462 +    );463 +}464 -- 465 2.56.0466