diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..1e7bf3e --- /dev/null +++ b/.dockerignore @@ -0,0 +1,13 @@ +/target +/result +/result-* +/cache +/config +/fumi.rns +.git +.jj +.env +*.db +*.db-wal +*.db-shm +server.key diff --git a/Containerfile b/Containerfile new file mode 100644 index 0000000..23f11f7 --- /dev/null +++ b/Containerfile @@ -0,0 +1,22 @@ +# Default (non-rns) build only: microReticulum's cmake/C++ build isn't set +# up for the musl toolchain alpine gives us, mirroring packages.static in +# flake.nix. Use the Nix flake if you need the rns feature. +FROM rust:1-alpine AS builder +RUN apk add --no-cache musl-dev gcc +WORKDIR /usr/src/bunshin +COPY Cargo.toml Cargo.lock build.rs ./ +COPY src ./src +RUN cargo build --release --locked + +FROM alpine:3.20 +RUN apk add --no-cache ca-certificates \ + && adduser -D -h /data -u 10000 bunshin +COPY --from=builder /usr/src/bunshin/target/release/bunshin /usr/local/bin/bunshin +COPY docker-entrypoint.sh /usr/local/bin/docker-entrypoint.sh +RUN chmod +x /usr/local/bin/docker-entrypoint.sh +USER bunshin +WORKDIR /data +VOLUME /data +EXPOSE 1961/tcp +ENTRYPOINT ["docker-entrypoint.sh"] +CMD ["serve", "--key", "/data/server.key", "--db", "/data/mail.db", "--host", "0.0.0.0", "--port", "1961"] diff --git a/README.md b/README.md index bd6338d..9e58b34 100644 --- a/README.md +++ b/README.md @@ -6,11 +6,29 @@ A Rust implementation of the [Smol Mail](https://code.randogoth.com/randogoth/sm The server never sees plaintext, sender identities or any private key. It learns only which mailbox an envelope is for, its size, and when it arrived. -The flake's main purpose is turnkey deployment on a NixOS host: import `nixosModules.default`, point it at a key, and `nixos-rebuild switch`. +The quickest way to run it is the container image below. For a NixOS host, the flake also provides turnkey deployment as a special case: import `nixosModules.default`, point it at a key, and `nixos-rebuild switch`. + +## Deploying with a container + +Pull the published image and bring up a server in one command: + +``` +podman run -d --name bunshin -p 1961:1961 -v bunshin-data:/data code.randogoth.com/randogoth/bunshin +``` + +`docker` works the same way — the image is a standard OCI image either way. The entrypoint generates `/data/server.key` on first run if it's missing, then runs `serve` against `/data/server.key` and `/data/mail.db` on `0.0.0.0:1961`; `podman logs bunshin` prints the generated public key to publish to clients. A key already in the volume is left alone, so restarts and upgrades keep the same identity. + +Pass your own arguments to run `keygen` or a customized `serve` instead of the default — they take the same flags as a bare-metal install, e.g. `podman run --rm -v bunshin-data:/data code.randogoth.com/randogoth/bunshin keygen --key /data/server.key --force`. + +To build the image locally instead of pulling (same non-`rns` build as the release image, see the `Containerfile` header comment): + +``` +podman build -t bunshin -f Containerfile . +``` ## Deploying on NixOS -Add bunshin as a flake input and import the module: +For a NixOS host that already manages the rest of its config with Nix, import the module instead of running the container: ```nix { @@ -34,7 +52,7 @@ Add bunshin as a flake input and import the module: } ``` -`services.bunshin` also takes `host`, `port`, `dataDir`, `maxEnvelope`, `quota`, `requestsQuota`, `retentionDays`, `requestsRetentionDays`, `maxTokens`, `rateConnections`, `rateSends`, `rateTokens` and `domains`; see `flake.nix` for defaults. The module renders a `systemd` unit that runs `bunshin serve` under `DynamicUser`; it does not generate a key. +`services.bunshin` also takes `host`, `port`, `dataDir`, `maxEnvelope`, `quota`, `requestsQuota`, `retentionDays`, `requestsRetentionDays`, `maxTokens`, `rateConnections`, `rateSends`, `rateTokens`, `maxConnections` and `domains`; see `flake.nix` for defaults. The module renders a `systemd` unit that runs `bunshin serve` under `DynamicUser`; unlike the container entrypoint, it does not generate a key. For the RNS carrier, build `packages.rns`, set `services.bunshin.package` to it, and enable `services.bunshin.rns` with its `keyFile`; see [RNS.md](RNS.md) for the protocol and the remaining options. @@ -115,7 +133,7 @@ bunshin keygen --key server.key bunshin serve --key server.key --db mail.db --host 0.0.0.0 --port 1961 ``` -`serve` accepts `--max-envelope`, `--quota`, `--requests-quota`, `--retention-days`, `--requests-retention-days`, `--max-tokens`, `--invite-token`, `--rate-connections`, `--rate-sends` and `--rate-tokens` to control size limits, the mailbox's two quota tiers, their retention, the accept-token cap, registration gating and abuse control. Run `bunshin serve --help` for defaults. +`serve` accepts `--max-envelope`, `--quota`, `--requests-quota`, `--retention-days`, `--requests-retention-days`, `--max-tokens`, `--invite-token`, `--rate-connections`, `--rate-sends`, `--rate-tokens` and `--max-connections` to control size limits, the mailbox's two quota tiers, their retention, the accept-token cap, registration gating, abuse control and the concurrent-connection cap. Rate limits are sliding-window: hits age out 60 s after they happen, so a burst straddling a window boundary cannot exceed the configured rate. Connections past `--max-connections` (default 100, 0 = unlimited) are refused, mirroring the RNS carrier's link cap. Run `bunshin serve --help` for defaults. `--verbose` (or `services.bunshin.verbose` in the NixOS module) raises logging to debug level, which adds server-side metrics on both carriers: per-operation timing, request and response sizes, handshake duration, per-session summaries, and RNS link events. The default info level stays quiet on success. diff --git a/RNS.md b/RNS.md index c9a15fe..38e24c9 100644 --- a/RNS.md +++ b/RNS.md @@ -16,7 +16,7 @@ Implemented behind the `rns` cargo feature (protocol 1.2; the default build stil | Ops, status codes, envelope | §6, §12 | identical | | AUTH / REGISTER binding | Noise handshake hash, server static key | derived, see §2 | | Announce | — | at startup and every 2 h; interfaces rate-limit to ≈1/hour, which is the ceiling | -| Abuse control | per-IP rate limits | per-link request + byte limits, concurrent-link cap | +| Abuse control | per-IP sliding-window rate limits, concurrent-connection cap | per-link request + byte limits, concurrent-link cap | One wire-format detail the upstream spec leaves implicit: microReticulum splices the request payload and the response into their msgpack envelopes verbatim, so both directions carry the smolmail payload as a msgpack binary. The shim unpacks on the way in and packs on the way out; the Rust side only ever sees `op u8 || body` and `status u8 || payload`. diff --git a/docker-entrypoint.sh b/docker-entrypoint.sh new file mode 100644 index 0000000..40d2a18 --- /dev/null +++ b/docker-entrypoint.sh @@ -0,0 +1,12 @@ +#!/bin/sh +# Generates the server key on first run so a bare `docker run` against an +# empty volume works; a key already at BUNSHIN_KEY is left untouched. +set -e + +: "${BUNSHIN_KEY:=/data/server.key}" + +if [ "$1" = "serve" ] && [ ! -f "$BUNSHIN_KEY" ]; then + bunshin keygen --key "$BUNSHIN_KEY" +fi + +exec bunshin "$@" diff --git a/flake.nix b/flake.nix index ace814c..29158a9 100644 --- a/flake.nix +++ b/flake.nix @@ -109,15 +109,17 @@ name = "bunshin"; }; - # Builds packages.static and uploads it to this repo owner's - # Forgejo generic package registry, tagged by short commit hash. - # Needs FORGEJO_TOKEN (a token with write:package scope) in the - # environment; run from a checkout so `git rev-parse` and the `.` - # flake ref resolve to the right place. + # Builds packages.static and publishes it as a Forgejo release on + # this repo, tagged by short commit hash (creating the release if + # it doesn't exist yet, replacing the asset if it does — safe to + # rerun for the same commit). Needs FORGEJO_TOKEN (a token with + # write:repository scope) in the environment; run from a checkout + # so `git rev-parse` and the `.` flake ref resolve to the right + # place. apps.release-static = flake-utils.lib.mkApp { drv = pkgs.writeShellApplication { name = "bunshin-release-static"; - runtimeInputs = [ pkgs.nix pkgs.curl pkgs.git ]; + runtimeInputs = [ pkgs.nix pkgs.curl pkgs.git pkgs.jq ]; text = '' if [ -f .env ]; then set -a @@ -125,12 +127,63 @@ . ./.env set +a fi + : "''${FORGEJO_TOKEN:?set FORGEJO_TOKEN (env or .env) to a Forgejo token with write:repository scope}" + rev=$(git rev-parse --short HEAD) + sha=$(git rev-parse HEAD) out=$(nix build .#static --no-link --print-out-paths) + bin="$out/bin/bunshin" + + api="https://code.randogoth.com/api/v1/repos/randogoth/bunshin" + auth=(-H "Authorization: token ''${FORGEJO_TOKEN}") + + release_id=$(curl -sS "''${auth[@]}" "$api/releases/tags/$rev" | jq -r '.id // empty') + if [ -z "$release_id" ]; then + release_id=$(curl -sSf "''${auth[@]}" -H "Content-Type: application/json" \ + -d "$(jq -n --arg tag "$rev" --arg sha "$sha" \ + '{tag_name:$tag, target_commitish:$sha, name:$tag, body:"Static musl build.", draft:false, prerelease:false}')" \ + "$api/releases" | jq -r '.id') + fi + + asset_id=$(curl -sSf "''${auth[@]}" "$api/releases/$release_id/assets" | jq -r '.[] | select(.name=="bunshin") | .id' | head -1) + if [ -n "$asset_id" ]; then + curl -sSf -X DELETE "''${auth[@]}" "$api/releases/$release_id/assets/$asset_id" >/dev/null + fi + curl -sSf "''${auth[@]}" -F "attachment=@$bin;filename=bunshin" "$api/releases/$release_id/assets?name=bunshin" >/dev/null + + echo "released: https://code.randogoth.com/randogoth/bunshin/releases/tag/$rev" + ''; + }; + }; + + # Builds the Containerfile image (same non-rns build the release + # binary is, see its header comment) and pushes it to this repo's + # Forgejo container registry, tagged by short commit hash and + # `latest`. Needs FORGEJO_TOKEN (a token with write:package scope) + # in the environment; run from a checkout so `git rev-parse` and + # the Containerfile build context resolve to the right place. + apps.release-container = flake-utils.lib.mkApp { + drv = pkgs.writeShellApplication { + name = "bunshin-release-container"; + runtimeInputs = [ pkgs.podman pkgs.git ]; + text = '' + if [ -f .env ]; then + set -a + # shellcheck disable=SC1091 + . ./.env + set +a + fi : "''${FORGEJO_TOKEN:?set FORGEJO_TOKEN (env or .env) to a Forgejo token with write:package scope}" - url="https://code.randogoth.com/api/packages/randogoth/generic/bunshin/$rev/bunshin" - curl -sSf -H "Authorization: token ''${FORGEJO_TOKEN}" --upload-file "$out/bin/bunshin" "$url" - echo "uploaded: $url" + + rev=$(git rev-parse --short HEAD) + registry="code.randogoth.com/randogoth/bunshin" + + echo "''${FORGEJO_TOKEN}" | podman login code.randogoth.com -u randogoth --password-stdin + podman build -f Containerfile -t "$registry:$rev" -t "$registry:latest" . + podman push "$registry:$rev" + podman push "$registry:latest" + + echo "pushed: https://code.randogoth.com/randogoth/-/packages/container/bunshin" ''; }; }; @@ -151,7 +204,7 @@ package = mkOption { type = types.package; - default = self.packages.${pkgs.system}.default; + default = self.packages.${pkgs.stdenv.hostPlatform.system}.default; description = '' bunshin package to run. Use `packages.rns` when the RNS carrier is enabled: the default package is built without it. @@ -302,6 +355,12 @@ description = "Max SEND operations per minute, per accept token."; }; + maxConnections = mkOption { + type = types.ints.unsigned; + default = 100; + description = "Max concurrently served TCP connections; past the cap connections are refused. 0 means unlimited."; + }; + inviteToken = mkOption { type = types.nullOr types.str; default = null; @@ -455,6 +514,7 @@ rate_connections = cfg.rateConnections; rate_sends = cfg.rateSends; rate_tokens = cfg.rateTokens; + max_connections = cfg.maxConnections; } // lib.optionalAttrs (cfg.inviteToken != null) { invite_token = cfg.inviteToken; } // lib.optionalAttrs (cfg.inviteTokenFile != null) { @@ -502,6 +562,7 @@ --rate-connections ${toString cfg.rateConnections} --rate-sends ${toString cfg.rateSends} --rate-tokens ${toString cfg.rateTokens} + --max-connections ${toString cfg.maxConnections} ) ${lib.optionalString cfg.rns.enable '' args+=( diff --git a/src/config.rs b/src/config.rs index a4c94d9..91a2eb3 100644 --- a/src/config.rs +++ b/src/config.rs @@ -25,6 +25,7 @@ const DEFAULT_MAX_TOKENS: u16 = 1024; const DEFAULT_RATE_CONNECTIONS: u32 = 120; const DEFAULT_RATE_SENDS: u32 = 60; const DEFAULT_RATE_TOKENS: u32 = 30; +const DEFAULT_MAX_CONNECTIONS: usize = 100; /// One resolved domain, ready to serve. pub struct DomainConfig { @@ -43,6 +44,7 @@ pub struct DomainConfig { pub rate_connections: u32, pub rate_sends: u32, pub rate_tokens: u32, + pub max_connections: usize, } /// The union of every setting either table accepts; `deny_unknown_fields` @@ -67,6 +69,7 @@ struct Table { rate_connections: Option, rate_sends: Option, rate_tokens: Option, + max_connections: Option, } #[derive(Deserialize)] @@ -187,6 +190,10 @@ pub fn load(path: &str) -> anyhow::Result> { .rate_tokens .or(file.defaults.rate_tokens) .unwrap_or(DEFAULT_RATE_TOKENS), + max_connections: table + .max_connections + .or(file.defaults.max_connections) + .unwrap_or(DEFAULT_MAX_CONNECTIONS), }); } Ok(domains) @@ -218,6 +225,7 @@ mod tests { assert_eq!(d.max_envelope, 768 << 10); assert_eq!(d.quota, 64 << 20); assert_eq!(d.rate_tokens, 30); + assert_eq!(d.max_connections, 100); assert!(d.invite_token.is_none()); } diff --git a/src/main.rs b/src/main.rs index dda8f6c..ed1792b 100644 --- a/src/main.rs +++ b/src/main.rs @@ -45,7 +45,7 @@ enum Command { Serve { /// Serve one domain per [domains.] table; every domain has its /// own key, port and mailbox database - #[arg(long, conflicts_with_all = ["key", "db", "host", "port", "max_envelope", "quota", "requests_quota", "retention_days", "requests_retention_days", "max_tokens", "invite_token", "rate_connections", "rate_sends", "rate_tokens"])] + #[arg(long, conflicts_with_all = ["key", "db", "host", "port", "max_envelope", "quota", "requests_quota", "retention_days", "requests_retention_days", "max_tokens", "invite_token", "rate_connections", "rate_sends", "rate_tokens", "max_connections"])] #[cfg_attr(feature = "rns", arg(conflicts_with_all = ["enabled", "rns_key", "rns_max_envelope", "rns_fetch_budget", "rns_max_links", "rns_rate_link_requests", "rns_rate_link_bytes", "rns_link_idle", "rns_udp", "rns_udp_forward"]))] config: Option, #[arg(long, default_value = "server.key")] @@ -76,6 +76,9 @@ enum Command { rate_sends: u32, #[arg(long = "rate-tokens", default_value_t = 30)] rate_tokens: u32, + /// Refuse connections past this many concurrent ones (0 = unlimited) + #[arg(long = "max-connections", default_value_t = 100)] + max_connections: usize, #[cfg(feature = "rns")] #[command(flatten)] rns: RnsServeArgs, @@ -153,6 +156,7 @@ fn main() -> anyhow::Result<()> { rate_connections, rate_sends, rate_tokens, + max_connections, } => { if let Some(path) = config { return server::serve_domains(config::load(&path)?); @@ -172,6 +176,7 @@ fn main() -> anyhow::Result<()> { rate_connections, rate_sends, rate_tokens, + max_connections, }) } #[cfg(feature = "rns")] @@ -191,6 +196,7 @@ fn main() -> anyhow::Result<()> { rate_connections, rate_sends, rate_tokens, + max_connections, rns, } => { if let Some(path) = config { @@ -248,6 +254,7 @@ fn main() -> anyhow::Result<()> { rate_connections, rate_sends, rate_tokens, + max_connections, }) } } diff --git a/src/ratelimit.rs b/src/ratelimit.rs index 3963ae4..028a131 100644 --- a/src/ratelimit.rs +++ b/src/ratelimit.rs @@ -1,20 +1,18 @@ -//! Fixed-window per-IP counter, the whole of the server's abuse control. +//! Sliding-window per-IP counters, the whole of the server's abuse control. //! //! A server cannot see senders, so quotas, size caps and this are all it has. +//! Hits are logged per key and pruned once they age out of the window, so a +//! burst straddling a window boundary cannot exceed the configured rate the +//! way a fixed-window counter would. -use std::collections::HashMap; +use std::collections::{HashMap, VecDeque}; use std::sync::Mutex; use std::time::{Duration, Instant}; -struct Window { - start: Instant, - count: u32, -} - pub struct RateLimiter { limit: u32, window: Duration, - hits: Mutex>, + hits: Mutex>>, } impl RateLimiter { @@ -27,44 +25,44 @@ impl RateLimiter { } pub fn allow(&self, ip: &str) -> bool { + self.allow_at(ip, Instant::now()) + } + + fn allow_at(&self, key: &str, now: Instant) -> bool { if self.limit == 0 { return true; } - let now = Instant::now(); let mut hits = self.hits.lock().unwrap(); - let entry = hits.entry(ip.to_string()).or_insert(Window { - start: now, - count: 0, - }); - if now.duration_since(entry.start) >= self.window { - entry.start = now; - entry.count = 0; - } - if entry.count >= self.limit { + let entry = hits.entry(key.to_string()).or_default(); + prune(entry, now, self.window); + if entry.len() >= self.limit as usize { return false; } - entry.count += 1; + entry.push_back(now); if hits.len() > 4096 { let window = self.window; - hits.retain(|_, w| now.duration_since(w.start) < window); + hits.retain(|_, e| !e.is_empty() && now.duration_since(*e.back().unwrap()) <= window); } true } } -/// Fixed-window byte counter for per-link transfer budgets: `RateLimiter` +fn prune(entry: &mut VecDeque, now: Instant, window: Duration) { + while entry + .front() + .is_some_and(|t| now.duration_since(*t) > window) + { + entry.pop_front(); + } +} + +/// Sliding-window byte counter for per-link transfer budgets: `RateLimiter` /// counts events, this counts bytes, so it gets its own small type. #[cfg(feature = "rns")] pub struct ByteRateLimiter { limit: u64, window: Duration, - hits: Mutex>, -} - -#[cfg(feature = "rns")] -struct ByteWindow { - start: Instant, - bytes: u64, + hits: Mutex>>, } #[cfg(feature = "rns")] @@ -79,26 +77,29 @@ impl ByteRateLimiter { /// Records `bytes` against `key` if the window still has room for them. pub fn allow(&self, key: &str, bytes: usize) -> bool { + self.allow_at(key, bytes, Instant::now()) + } + + fn allow_at(&self, key: &str, bytes: usize, now: Instant) -> bool { if self.limit == 0 { return true; } - let now = Instant::now(); let mut hits = self.hits.lock().unwrap(); - let entry = hits.entry(key.to_string()).or_insert(ByteWindow { - start: now, - bytes: 0, - }); - if now.duration_since(entry.start) >= self.window { - entry.start = now; - entry.bytes = 0; + let entry = hits.entry(key.to_string()).or_default(); + while entry + .front() + .is_some_and(|(t, _)| now.duration_since(*t) > self.window) + { + entry.pop_front(); } - if entry.bytes + bytes as u64 > self.limit { + let used: u64 = entry.iter().map(|(_, n)| *n).sum(); + if used + bytes as u64 > self.limit { return false; } - entry.bytes += bytes as u64; + entry.push_back((now, bytes as u64)); if hits.len() > 4096 { let window = self.window; - hits.retain(|_, w| now.duration_since(w.start) < window); + hits.retain(|_, e| !e.is_empty() && now.duration_since(e.back().unwrap().0) <= window); } true } @@ -111,9 +112,10 @@ mod tests { #[test] fn allows_up_to_limit_then_blocks() { let rl = RateLimiter::new(2); - assert!(rl.allow("1.2.3.4")); - assert!(rl.allow("1.2.3.4")); - assert!(!rl.allow("1.2.3.4")); + let now = Instant::now(); + assert!(rl.allow_at("1.2.3.4", now)); + assert!(rl.allow_at("1.2.3.4", now)); + assert!(!rl.allow_at("1.2.3.4", now)); } #[test] @@ -131,4 +133,42 @@ mod tests { assert!(rl.allow("2.2.2.2")); assert!(!rl.allow("1.1.1.1")); } + + /// The regression the sliding window exists for: a burst straddling a + /// window boundary must not be handed a fresh window, only the slots + /// its own hits have aged out. A hit counts for the whole window, + /// including the instant it is exactly window old. + #[test] + fn boundary_burst_cannot_double_the_rate() { + let rl = RateLimiter::new(2); + let t0 = Instant::now(); + let window = Duration::from_secs(60); + assert!(rl.allow_at("1.2.3.4", t0)); + assert!(rl.allow_at("1.2.3.4", t0 + Duration::from_secs(1))); + // At the boundary a fixed-window counter would reset and admit + // two more; both hits are still window-old or fresher. + assert!(!rl.allow_at("1.2.3.4", t0 + window)); + // Half a second later the t0 hit has aged out but the +1s hit + // has not, so exactly one slot is free. + let later = t0 + window + Duration::from_millis(500); + assert!(rl.allow_at("1.2.3.4", later)); + assert!(!rl.allow_at("1.2.3.4", later)); + } + + #[cfg(feature = "rns")] + #[test] + fn byte_boundary_burst_cannot_double_the_budget() { + let bl = ByteRateLimiter::new(100); + let t0 = Instant::now(); + let window = Duration::from_secs(60); + assert!(bl.allow_at("link", 60, t0)); + assert!(bl.allow_at("link", 40, t0 + Duration::from_secs(1))); + // At the boundary both hits still count: 100 bytes used, no room. + assert!(!bl.allow_at("link", 1, t0 + window)); + // Half a second later only the 60-byte hit has aged out, freeing + // exactly its 60 bytes. + let later = t0 + window + Duration::from_millis(500); + assert!(bl.allow_at("link", 60, later)); + assert!(!bl.allow_at("link", 1, later)); + } } diff --git a/src/server.rs b/src/server.rs index b0bd510..edfdb9f 100644 --- a/src/server.rs +++ b/src/server.rs @@ -1,7 +1,7 @@ //! TCP accept loop, per-connection handling, and the background purge loop. use std::net::{TcpListener, TcpStream}; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use std::time::Duration; use crate::bind::TransportBindValues; @@ -28,6 +28,7 @@ pub struct ServeArgs { pub rate_connections: u32, pub rate_sends: u32, pub rate_tokens: u32, + pub max_connections: usize, } /// The single-domain CLI path: one anonymous domain. @@ -48,6 +49,7 @@ pub fn run(args: ServeArgs) -> anyhow::Result<()> { rate_connections: args.rate_connections, rate_sends: args.rate_sends, rate_tokens: args.rate_tokens, + max_connections: args.max_connections, }]) } @@ -94,8 +96,9 @@ pub fn serve_domains(domains: Vec) -> anyhow::Result<()> { purge_loop(purge_db_path, main_retention_secs, requests_retention_secs) }); + let gate = Arc::new(ConnGate::new(domain.max_connections)); let db_path = domain.db_path.clone(); - std::thread::spawn(move || accept_loop(listener, config, static_key, db_path)); + std::thread::spawn(move || accept_loop(listener, config, static_key, db_path, gate)); } // Each domain's accept loop runs in its own thread; nothing fails here. @@ -120,6 +123,7 @@ fn accept_loop( config: Arc, static_key: [u8; KEY_LEN], db_path: String, + gate: Arc, ) { for incoming in listener.incoming() { let stream = match incoming { @@ -139,9 +143,19 @@ fn accept_loop( continue; } + let permit = match gate.try_acquire() { + Some(p) => p, + None => { + log::warn!("connection limit reached, refusing {peer_ip}"); + continue; + } + }; + let config = Arc::clone(&config); let db_path = db_path.clone(); std::thread::spawn(move || { + // The permit is dropped with the connection, freeing its slot. + let _permit = permit; if let Err(e) = handle_connection(stream, &config, &static_key, &db_path, &peer_ip) { log::info!("connection error from {peer_ip}: {e}"); } @@ -149,6 +163,55 @@ fn accept_loop( } } +/// Cap on concurrently served TCP connections, the mirror of the RNS +/// carrier's link cap (SPEC.md sec 13.8): without one, a connection flood +/// would exhaust threads one unbounded spawn at a time. `max == 0` disables +/// the cap; past it, connections are refused, never queued. +struct ConnGate { + max: usize, + active: Mutex, +} + +impl ConnGate { + fn new(max: usize) -> Self { + ConnGate { + max, + active: Mutex::new(0), + } + } + + fn try_acquire(self: &Arc) -> Option { + if self.max == 0 { + return Some(ConnPermit { + gate: Arc::clone(self), + }); + } + let mut active = self.active.lock().unwrap(); + if *active >= self.max { + return None; + } + *active += 1; + Some(ConnPermit { + gate: Arc::clone(self), + }) + } +} + +/// Dropping releases the slot, so the count stays accurate whatever return +/// path or panic closes the connection. +struct ConnPermit { + gate: Arc, +} + +impl Drop for ConnPermit { + fn drop(&mut self) { + if self.gate.max == 0 { + return; + } + *self.gate.active.lock().unwrap() -= 1; + } +} + /// Also the hostile harness's entry point: it drives real connections /// through the same accept/handshake/session path the TCP carrier serves. pub(crate) fn handle_connection( @@ -265,3 +328,27 @@ fn purge_loop(db_path: String, main_retention_secs: i64, requests_retention_secs } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn gate_refuses_past_cap_and_frees_on_drop() { + let gate = Arc::new(ConnGate::new(2)); + let a = gate.try_acquire().unwrap(); + let b = gate.try_acquire().unwrap(); + assert!(gate.try_acquire().is_none()); + drop(b); + assert!(gate.try_acquire().is_some()); + drop(a); + } + + #[test] + fn zero_cap_means_unlimited() { + let gate = Arc::new(ConnGate::new(0)); + for _ in 0..100 { + assert!(gate.try_acquire().is_some()); + } + } +}