diff --git a/README.md b/README.md index 7c6610f..fcfed78 100644 --- a/README.md +++ b/README.md @@ -55,7 +55,7 @@ bunshin rns-keygen --key server.rns.key bunshin serve --rns --key server.key --rns-key server.rns.key --db mail.db --rns-udp 0.0.0.0:4242 ``` -`rns-keygen` writes the 64-byte Reticulum identity and prints the destination hash; publish `smol+rns://@` as the server's address. RNS-specific limits (`--rns-max-envelope`, `--rns-fetch-budget`, `--rns-max-links`, `--rns-rate-link-requests`, `--rns-rate-link-bytes`) default to mesh-friendly 32 KiB caps; the UDP interface is the one supported transport interface so far. See [RNS.md](RNS.md). +`rns-keygen` writes the 64-byte Reticulum identity and prints the destination hash; publish `smol+rns://@` as the server's address. RNS-specific limits (`--rns-max-envelope`, `--rns-fetch-budget`, `--rns-max-links`, `--rns-rate-link-requests`, `--rns-rate-link-bytes`) default to mesh-friendly 32 KiB caps; `--rns-link-idle` (default 300 s) reaps links whose client went silent, since nothing else times a link out; the UDP interface is the one supported transport interface so far. See [RNS.md](RNS.md). ## Building @@ -77,6 +77,8 @@ 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. +`--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. + ## Status Implements SPEC.md version 1.1 in full: `AUTH`, `RESOLVE`, `SEND`, `FETCH`, `DELETE` and `REGISTER`, including accept tokens and the main/requests tier split, fetch cursors, 32-byte message ids, `REGISTER` proof of possession, and dual-signed key rotation chains. diff --git a/RNS.md b/RNS.md index 29eb159..c9a15fe 100644 --- a/RNS.md +++ b/RNS.md @@ -126,6 +126,7 @@ One process serves both carriers: shared `Store`, shared config, one systemd uni | `--rns-max-links` | 100 | concurrent links — refused past the cap | | `--rns-rate-link-requests` | 60 | per link per minute | | `--rns-rate-link-bytes` | 1048576 | per link per minute, via `ByteRateLimiter` | +| `--rns-link-idle` | 300 | seconds of silence before a link is reaped; 0 disables the reaper | | `--rns-udp` | `127.0.0.1:4242` | UDP interface to listen on, host[:port] | | `--rns-udp-forward` | unset | optional forward target; without it the interface replies to the last datagram's source | @@ -144,8 +145,9 @@ The per-link request limiter reuses `RateLimiter` keyed by the link hex. The per ## 9. Remaining open items 1. Cross-transport envelope size: per-transport caps are enforced; a 768 KiB TCP envelope is still delivered whole to an RNS FETCH. Operators should apply the smaller cap mailbox-wide when in doubt (upstream spec 13.7). -2. Interfaces beyond UDP (RNS-compatible TCP hub/client, I2P): microReticulum ships none; the UDP interface is the first supported one. -3. The reference `test_interop/run_request_response.sh` harness was not rerun against the shim; its PlatformIO build step is unavailable here. The equivalent interop was exercised directly against `smolmail_rns.py` (§10.6). +2. Upstream `microReticulum` deadlock: a client dying mid-resource-transfer wedges the single transport loop forever — `Resource::request_next` sends its `RESOURCE_REQ` through `Transport::outbound` from inside the `jobs()` resource tick, and `outbound` busy-waits on `_jobs_running`, a flag `jobs()` itself holds. No announces, no new links, mesh dead until restart. The vendored source is a local fork carrying the fix (same-thread reentrant calls skip the cycle wait); the equivalent report is filed upstream. +3. Interfaces beyond UDP (RNS-compatible TCP hub/client, I2P): microReticulum ships none; the UDP interface is the first supported one. +4. The reference `test_interop/run_request_response.sh` harness was not rerun against the shim; its PlatformIO build step is unavailable here. The equivalent interop was exercised directly against `smolmail_rns.py` (§10.6). Formerly open items resolved during implementation: raw identity bytes exist (`load_private_key`); the shim ABI and run-loop drive pattern are §7.1/§7.2; `ALLOW_ALL` matches the reference; announcing at startup + every 2 h, unconditional even when invite-only (the invite gates REGISTER, not the link); the interface question is §7.4. diff --git a/THIRD_PARTY_NOTICES.md b/THIRD_PARTY_NOTICES.md index c4bf123..bdfa0d7 100644 --- a/THIRD_PARTY_NOTICES.md +++ b/THIRD_PARTY_NOTICES.md @@ -1,6 +1,6 @@ # Third-Party Notices -bunshin incorporates the third-party components listed below. Components are grouped by license, each with the copyright notice that applies to it, followed by the full license text. A component listed under several licenses is multi-licensed under those options; any one of them may be chosen. All components are redistributed unmodified from their published sources. +bunshin incorporates the third-party components listed below. Components are grouped by license, each with the copyright notice that applies to it, followed by the full license text. A component listed under several licenses is multi-licensed under those options; any one of them may be chosen. Components are redistributed from their published sources, except where a bullet notes a local patch; all other components are unmodified. The Rust crates listed are all crates resolved by `Cargo.lock`, including transitive dependencies, which are statically linked into the bunshin binary. The SQLite C library is bundled in source form by the `libsqlite3-sys` crate and is public domain (see the final section). @@ -102,7 +102,7 @@ The libraries marked "vendored source" are pinned nix flake inputs; each is iden - zerocopy-derive 0.8.59 - Copyright 2023 The Fuchsia Authors - zeroize 1.9.0 - Copyright (c) 2018-2026 The RustCrypto Project Developers - zeroize_derive 1.5.0 - Copyright (c) 2019-2026 The RustCrypto Project Developers -- microReticulum (vendored source, https://github.com/attermann/microReticulum, rev 40fa628809d5) +- microReticulum (vendored source, https://github.com/attermann/microReticulum, rev 40fa628809d5; patched locally — same-thread reentrant calls skip the Transport::outbound/inbound jobs-cycle wait, fixing a transport-loop deadlock, see RNS.md §9) - microStore (vendored source, https://github.com/attermann/microStore, rev 0f28567fe00a) ``` diff --git a/flake.lock b/flake.lock index 4880ea0..f167260 100644 --- a/flake.lock +++ b/flake.lock @@ -21,18 +21,18 @@ "microReticulum": { "flake": false, "locked": { - "lastModified": 1784575896, - "narHash": "sha256-cqo+BJ3tsmw4fD+SL0D3X9FKCgf+n0VQng2pSIuyK/8=", - "owner": "attermann", - "repo": "microReticulum", - "rev": "40fa628809d57140180c1c833559ab96fec992c1", - "type": "github" + "lastModified": 1790668417, + "narHash": "sha256-BrOkpepwcCfB7o9TOYOQ3SnZfwHVx3WyUTJsIGwAdLk=", + "ref": "fix/outbound-reentrancy-deadlock", + "rev": "03be60051d40e556933101289c0203d9f3308b62", + "revCount": 226, + "type": "git", + "url": "file:///mnt/data/Projects/code/microReticulum" }, "original": { - "owner": "attermann", - "repo": "microReticulum", - "rev": "40fa628809d57140180c1c833559ab96fec992c1", - "type": "github" + "ref": "fix/outbound-reentrancy-deadlock", + "type": "git", + "url": "file:///mnt/data/Projects/code/microReticulum" } }, "nixpkgs": { diff --git a/flake.nix b/flake.nix index 2ff3679..71e4afd 100644 --- a/flake.nix +++ b/flake.nix @@ -9,8 +9,13 @@ # so every microReticulum FetchContent dependency is vendored as its own # flake input and handed over via RNS__SOURCE_DIR. microReticulum # is pinned at the commit RNS.md sec 5 verified against. + # Local fork of attermann/microReticulum at rev 40fa628809d5 plus a + # local patch: Transport::outbound/inbound skip their jobs-cycle wait + # for same-thread reentrant calls, which otherwise deadlock the single + # transport loop when a stalled resource tick sends a RESOURCE_REQ + # from inside jobs() (see RNS.md sec 9). microReticulum = { - url = "github:attermann/microReticulum/40fa628809d57140180c1c833559ab96fec992c1"; + url = "git+file:///mnt/data/Projects/code/microReticulum?ref=fix/outbound-reentrancy-deadlock"; flake = false; }; rns-arduinojson = { @@ -220,6 +225,15 @@ description = "Open the configured TCP port in the firewall."; }; + verbose = mkOption { + type = types.bool; + default = false; + description = '' + Enable debug-level logging (--verbose): per-op timing, + session summaries and RNS link events on both carriers. + ''; + }; + rns = { enable = mkEnableOption "the RNS carrier alongside TCP (needs an rns-built package)"; @@ -263,6 +277,15 @@ description = "Max request bytes per minute, per link."; }; + linkIdleSecs = mkOption { + type = types.ints.unsigned; + default = 300; + description = '' + Seconds after which an idle RNS link is reaped, freeing + its session and link slot; 0 disables the reaper. + ''; + }; + udpListenPort = mkOption { type = types.port; default = 4242; @@ -328,11 +351,14 @@ --rns-max-links ${toString cfg.rns.maxLinks} --rns-rate-link-requests ${toString cfg.rns.rateLinkRequests} --rns-rate-link-bytes ${toString cfg.rns.rateLinkBytes} + --rns-link-idle ${toString cfg.rns.linkIdleSecs} --rns-udp 0.0.0.0:${toString cfg.rns.udpListenPort} ) ''} ${lib.optionalString (cfg.rns.enable && cfg.rns.udpForward != null) ''args+=(--rns-udp-forward ${lib.escapeShellArg cfg.rns.udpForward})''} + ${lib.optionalString cfg.verbose + ''args+=(--verbose)''} ${lib.optionalString (cfg.inviteToken != null) ''args+=(--invite-token ${lib.escapeShellArg cfg.inviteToken})''} ${lib.optionalString (cfg.inviteTokenFile != null) diff --git a/src/channel.rs b/src/channel.rs index fd67c93..bee9074 100644 --- a/src/channel.rs +++ b/src/channel.rs @@ -9,7 +9,10 @@ use std::net::TcpStream; use snow::{Builder, TransportState}; -use crate::proto::{ProtocolError, MAX_FRAME, NOISE_PARAMS, NOISE_PAYLOAD, PROLOGUE}; +use crate::proto::{ + ProtocolError, HEADER_TIMEOUT_SECS, IDLE_TIMEOUT_SECS, MAX_FRAME, NOISE_PARAMS, NOISE_PAYLOAD, + PROLOGUE, +}; /// Runs the Noise_NX responder handshake. The initiator stays anonymous; /// only we hold a static key. Returns the transport state and the @@ -72,6 +75,15 @@ impl Channel { } fn read_noise(&mut self) -> anyhow::Result> { + // A frame already started must finish quickly; only a client idle + // between frames earns the full session idle timeout. + let deadline = if self.buf.is_empty() { + IDLE_TIMEOUT_SECS + } else { + HEADER_TIMEOUT_SECS + }; + self.stream + .set_read_timeout(Some(std::time::Duration::from_secs(deadline)))?; let len = read_u16_len(&mut self.stream)?; let mut ciphertext = vec![0u8; len]; read_exact_into(&mut self.stream, &mut ciphertext)?; diff --git a/src/harness.rs b/src/harness.rs new file mode 100644 index 0000000..a8959a3 --- /dev/null +++ b/src/harness.rs @@ -0,0 +1,617 @@ +//! Hostile transport harness (RNS.md sec 9 roster): a client that completes +//! the real Noise_NX handshake and then deliberately misbehaves — garbage and +//! truncated handshakes, malformed frames, oversized payloads, absurd +//! cursors, mid-frame drops — against the same accept/handshake/session path +//! the TCP carrier serves. Every scenario ends by asserting the server's +//! response class and that a fresh connection still completes a handshake: +//! the carrier must die never, the session must die cleanly. +//! +//! Transport-level kills (mid-link, mid-spin, mid-resource-transfer) belong +//! to the RNS carrier and were live-fired against the deployed build during +//! the stress campaign; they are recorded in MAIL.md, not replayed here. + +#![cfg(test)] + +use std::io::{Read, Write}; +use std::net::TcpStream; + +use ed25519_dalek::{Signer, SigningKey}; +use snow::{Builder, TransportState}; + +use crate::crypto::generate_static_key; +use crate::proto::{ + AUTH_REQUIRED, ENVELOPE_MAGIC, ENVELOPE_VERSION, ID_LEN, LABEL_AUTH, LABEL_REGISTER, + MALFORMED, MAX_FRAME, NOISE_PARAMS, NOISE_PAYLOAD, OP_AUTH, OP_FETCH, OP_REGISTER, OP_RESOLVE, + OP_SEND, PROLOGUE, TOO_LARGE, +}; +use crate::ratelimit::RateLimiter; +use crate::server::handle_connection; +use crate::session::ServerConfig; +use crate::store::Store; + +/// One harness target: a real accept loop serving `handle_connection` on an +/// ephemeral port, one thread per connection, exactly like `server::run`. +struct Target { + port: u16, + static_public: [u8; 32], +} + +impl Target { + fn start(max_envelope: usize) -> Self { + let (static_private, static_public) = generate_static_key(); + let config: &'static ServerConfig = Box::leak(Box::new(ServerConfig { + max_envelope, + fetch_budget: 512 << 10, + main_quota: 8 << 20, + requests_quota: 2 << 20, + max_tokens: 16, + invite_token: None, + // Unlimited: the harness is hostile by content, not by volume. + conn_limiter: RateLimiter::new(0), + send_limiter: RateLimiter::new(0), + token_limiter: RateLimiter::new(0), + })); + static TARGETS: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0); + let seq = TARGETS.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + let db_path = std::env::temp_dir() + .join(format!("bunshin-harness-{}-{seq}.db", std::process::id())) + .to_string_lossy() + .into_owned(); + let _ = std::fs::remove_file(&db_path); + Store::open(&db_path).unwrap(); + + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + std::thread::spawn(move || { + for stream in listener.incoming().flatten() { + let config = config; + let db_path = db_path.clone(); + std::thread::spawn(move || { + // The harness deliberately produces hostile connections; + // a panic in one must not take the accept loop with it. + let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + handle_connection( + stream, + config, + &static_private, + &db_path, + "127.0.0.1", + ) + })); + }); + } + }); + Target { + port, + static_public, + } + } + + /// Proof of life: a fresh connection completes the handshake and gets a + /// protocol answer (any status — UNKNOWN_USER counts; dead is no answer). + fn assert_alive(&self) { + let mut client = Client::connect(self.port); + client.handshake().expect("server survived"); + let (status, _) = client.request(OP_RESOLVE, &[7, b'h', b'a', b'r', b'n', b'e', b's', b's']); + assert_ne!(status, 0xFF, "server answered with a status byte"); + } +} + +/// The hostile client: a real Noise_NX initiator with raw-stream access so +/// scenarios can write garbage, truncated, or oversized wire data at will. +struct Client { + stream: TcpStream, + transport: Option, + handshake_hash: Vec, +} + +impl Client { + fn connect(port: u16) -> Self { + Client { + stream: TcpStream::connect(("127.0.0.1", port)).unwrap(), + transport: None, + handshake_hash: Vec::new(), + } + } + + /// The genuine client side of the server's `channel::handshake`. + fn handshake(&mut self) -> anyhow::Result<()> { + let params: snow::params::NoiseParams = NOISE_PARAMS.parse()?; + let mut noise = Builder::new(params).prologue(PROLOGUE).build_initiator()?; + let mut buf = [0u8; 65535]; + + let n = noise.write_message(&[], &mut buf)?; + self.stream.write_all(&(n as u16).to_be_bytes())?; + self.stream.write_all(&buf[..n])?; + + let mut len_buf = [0u8; 2]; + self.stream.read_exact(&mut len_buf)?; + let len = u16::from_be_bytes(len_buf) as usize; + let mut msg = vec![0u8; len]; + self.stream.read_exact(&mut msg)?; + let mut plaintext = [0u8; 65535]; + noise.read_message(&msg, &mut plaintext)?; + + anyhow::ensure!(noise.is_handshake_finished(), "handshake did not complete"); + self.handshake_hash = noise.get_handshake_hash().to_vec(); + self.transport = Some(noise.into_transport_mode()?); + Ok(()) + } + + fn write_noise(&mut self, payload: &[u8]) { + let transport = self.transport.as_mut().expect("handshake first"); + let mut packet = vec![0u8; payload.len() + 16]; + let n = transport.write_message(payload, &mut packet).unwrap(); + packet.truncate(n); + self.stream.write_all(&(packet.len() as u16).to_be_bytes()).unwrap(); + self.stream.write_all(&packet).unwrap(); + } + + fn read_noise(&mut self) -> Vec { + let transport = self.transport.as_mut().expect("handshake first"); + let mut len_buf = [0u8; 2]; + self.stream.read_exact(&mut len_buf).unwrap(); + let len = u16::from_be_bytes(len_buf) as usize; + let mut ciphertext = vec![0u8; len]; + self.stream.read_exact(&mut ciphertext).unwrap(); + let mut plaintext = vec![0u8; len]; + let n = transport.read_message(&ciphertext, &mut plaintext).unwrap(); + plaintext.truncate(n); + plaintext + } + + /// A well-formed frame — the baseline the hostile scenarios deviate from. + fn request(&mut self, op: u8, body: &[u8]) -> (u8, Vec) { + let mut frame = Vec::with_capacity(5 + body.len()); + frame.extend_from_slice(&((1 + body.len()) as u32).to_be_bytes()); + frame.push(op); + frame.extend_from_slice(body); + for chunk in frame.chunks(NOISE_PAYLOAD) { + self.write_noise(chunk); + } + self.read_frame() + } + + fn read_frame(&mut self) -> (u8, Vec) { + let mut buf = Vec::new(); + while buf.len() < 5 { + buf.extend_from_slice(&self.read_noise()); + } + let length = u32::from_be_bytes(buf[..4].try_into().unwrap()) as usize; + while buf.len() < 4 + length { + buf.extend_from_slice(&self.read_noise()); + } + let frame: Vec = buf[4..4 + length].to_vec(); + (frame[0], frame[1..].to_vec()) + } + + /// Raw bytes before the handshake — garbage on the wire. + fn write_raw(&mut self, bytes: &[u8]) { + self.stream.write_all(bytes).unwrap(); + } + + /// Claims a Noise message of `claimed` bytes but sends only `sent`, then + /// keeps the connection open briefly so the server sees a short read. + fn truncated_noise_len(&mut self, claimed: usize, sent: &[u8]) { + self.stream + .write_all(&(claimed as u16).to_be_bytes()) + .unwrap(); + self.stream.write_all(sent).unwrap(); + } + + /// One raw post-handshake Noise message carrying arbitrary plaintext. + fn raw_frame_bytes(&mut self, bytes: &[u8]) { + self.write_noise(bytes); + } +} + +/// SERVER-QUIET (the inverse): the server must never answer a connection +/// that sends nothing, and dropping it must cost nothing. +#[test] +fn silent_connection_costs_nothing_and_server_survives() { + let target = Target::start(768 << 10); + let mut client = Client::connect(target.port); + client.handshake().unwrap(); + std::thread::sleep(std::time::Duration::from_millis(50)); + drop(client); + target.assert_alive(); +} + +#[test] +fn short_garbage_handshake_is_refused_and_server_survives() { + let target = Target::start(768 << 10); + let mut client = Client::connect(target.port); + // Below the 32-byte ephemeral an NX first message needs, so snow must + // reject it outright and the server closes without writing anything. + client.write_raw(&(10u16).to_be_bytes()); + client.write_raw(&[0xA5; 10]); + let mut sink = [0u8; 16]; + assert!(client.stream.read(&mut sink).unwrap_or(0) == 0, "server closed"); + target.assert_alive(); +} + +/// Noise_NX authenticates the server in message two; the initiator's first +/// message is unauthenticated by design, so 32+ bytes of random data can +/// parse as an ephemeral key and draw a handshake reply. That reply must be +/// the most a garbage sender ever gets: no session forms from it. +#[test] +fn long_garbage_first_message_never_forms_a_session() { + let target = Target::start(768 << 10); + let mut client = Client::connect(target.port); + client.write_raw(&(100u16).to_be_bytes()); + client.write_raw(&[0xA5; 100]); + // Reply or close, either is protocol-legal; drop without completing. + drop(client); + target.assert_alive(); +} + +#[test] +fn truncated_handshake_is_survived() { + let target = Target::start(768 << 10); + let mut client = Client::connect(target.port); + client.truncated_noise_len(1000, &[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]); + drop(client); + target.assert_alive(); +} + +#[test] +fn unknown_op_gets_malformed_and_server_survives() { + let target = Target::start(768 << 10); + let mut client = Client::connect(target.port); + client.handshake().unwrap(); + let (op, body) = client.request(0x99, b""); + assert_eq!(op, 0x99); + assert_eq!(body[0], MALFORMED); + target.assert_alive(); +} + +#[test] +fn fetch_without_auth_gets_auth_required() { + let target = Target::start(768 << 10); + let mut client = Client::connect(target.port); + client.handshake().unwrap(); + let (op, body) = client.request(OP_FETCH, &[0u8; 40]); + assert_eq!(op, OP_FETCH); + assert_eq!(body[0], AUTH_REQUIRED); + target.assert_alive(); +} + +#[test] +fn garbage_resolve_body_gets_malformed() { + let target = Target::start(768 << 10); + let mut client = Client::connect(target.port); + client.handshake().unwrap(); + let (op, body) = client.request(OP_RESOLVE, &[0xEE; 50]); + assert_eq!(op, OP_RESOLVE); + assert_eq!(body[0], MALFORMED); + target.assert_alive(); +} + +#[test] +fn zero_length_frame_gets_malformed_then_close() { + let target = Target::start(768 << 10); + let mut client = Client::connect(target.port); + client.handshake().unwrap(); + // Four length bytes alone never reach the length check — read_frame + // buffers five bytes first — so the hostile frame is length + op byte. + client.raw_frame_bytes(&[0, 0, 0, 0, 0x00]); + let (op, body) = client.read_frame(); + assert_eq!(op, 0); + assert_eq!(body[0], MALFORMED); + let mut sink = [0u8; 16]; + assert!(client.stream.read(&mut sink).unwrap_or(0) == 0, "closed after bad frame"); + target.assert_alive(); +} + +#[test] +fn over_max_frame_length_gets_malformed_then_close() { + let target = Target::start(768 << 10); + let mut client = Client::connect(target.port); + client.handshake().unwrap(); + let claimed = (MAX_FRAME + 1) as u32; + let mut frame = claimed.to_be_bytes().to_vec(); + frame.push(0x00); + client.raw_frame_bytes(&frame); + let (op, body) = client.read_frame(); + assert_eq!(op, 0); + assert_eq!(body[0], MALFORMED); + target.assert_alive(); +} + +#[test] +fn oversized_envelope_gets_too_large() { + let target = Target::start(768 << 10); + let mut client = Client::connect(target.port); + client.handshake().unwrap(); + // Well-formed frame, well-formed envelope header, size past the cap. + let mut envelope = Vec::with_capacity(800 << 10); + envelope.extend_from_slice(ENVELOPE_MAGIC); + envelope.push(ENVELOPE_VERSION); + envelope.extend_from_slice(&[0u8; ID_LEN]); // recipient never looked up + envelope.resize(800 << 10, 0x41); + let mut body = vec![0u8]; // no accept-token MAC + body.extend_from_slice(&envelope); + let (op, resp) = client.request(OP_SEND, &body); + let body = &resp[..]; + assert_eq!(op, OP_SEND); + assert_eq!(body[0], TOO_LARGE); + target.assert_alive(); +} + +#[test] +fn bad_envelope_magic_gets_malformed() { + let target = Target::start(768 << 10); + let mut client = Client::connect(target.port); + client.handshake().unwrap(); + let mut envelope = vec![0x00; 128]; + envelope[4] = ENVELOPE_VERSION; + let (op, body) = client.request(OP_SEND, &envelope); + assert_eq!(op, OP_SEND); + assert_eq!(body[0], MALFORMED); + target.assert_alive(); +} + +#[test] +fn mid_frame_drop_is_survived() { + let target = Target::start(768 << 10); + let mut client = Client::connect(target.port); + client.handshake().unwrap(); + // Claim a frame, deliver a third of its header bytes, vanish. + client.raw_frame_bytes(&[0x00, 0x10, 0x00]); + drop(client); + target.assert_alive(); +} + +/// The well-behaved baseline plus the out-of-order cursor: a FETCH whose +/// cursor points past the end of time must answer an empty page, not an +/// error or a wrap-around into someone else's mail. +#[test] +fn registered_flow_survives_absurd_future_cursor() { + let target = Target::start(768 << 10); + let key = SigningKey::from_bytes(&[7u8; 32]); + let verifying = key.verifying_key(); + let identity: &[u8] = verifying.as_bytes(); + + let mut client = Client::connect(target.port); + client.handshake().unwrap(); + + // REGISTER with a real proof of possession over the server's static key. + let mut pop = LABEL_REGISTER.to_vec(); + pop.extend_from_slice(&target.static_public); + pop.extend_from_slice(b"harness"); + pop.extend_from_slice(identity); + let mut body = vec![7]; + body.extend_from_slice(b"harness"); + body.extend_from_slice(identity); + body.extend_from_slice(&key.sign(&pop).to_bytes()); + body.push(0); // no invite token + body.push(0); // plain registration + let (op, resp) = client.request(OP_REGISTER, &body); + assert_eq!((op, resp.as_slice()), (OP_REGISTER, &[0u8][..])); + + // AUTH over the real handshake hash this session actually negotiated. + let mut auth_msg = LABEL_AUTH.to_vec(); + auth_msg.extend_from_slice(&client.handshake_hash); + let mut body = vec![7]; + body.extend_from_slice(b"harness"); + body.extend_from_slice(identity); + body.extend_from_slice(&key.sign(&auth_msg).to_bytes()); + body.push(0); // sync = 0 + body.extend_from_slice(&0u16.to_be_bytes()); + let (op, resp) = client.request(OP_AUTH, &body); + assert_eq!(op, OP_AUTH); + assert_eq!(resp[0], 0, "status OK"); + assert_eq!( + u16::from_be_bytes(resp[1..3].try_into().unwrap()), + 0, + "zero accepted tokens" + ); + + // FETCH with a cursor at the end of time: OK, zero messages. + let mut body = Vec::new(); + body.extend_from_slice(&i64::MAX.to_be_bytes()); + body.extend_from_slice(&[0xFF; ID_LEN]); + let (op, resp) = client.request(OP_FETCH, &body); + assert_eq!(op, OP_FETCH); + assert_eq!(resp[0], 0, "status OK"); + assert_eq!(u16::from_be_bytes(resp[1..3].try_into().unwrap()), 0, "empty page"); + + // And the registered name resolves — the store came through intact. + let (op, resp) = client.request(OP_RESOLVE, &[7, b'h', b'a', b'r', b'n', b'e', b's', b's']); + assert_eq!(op, OP_RESOLVE); + assert_eq!(resp[0], 0); + target.assert_alive(); +} + + +// --------------------------------------------------------------------------- +// Long-lived multi-session driver: the last agreed stress-campaign phase. +// A dozen authenticated sessions live on one server simultaneously, run +// concurrent FETCH cursors against one mailbox, and — in the slow scenario — +// are held past the session idle timeout to verify the server reaps idle +// sessions cleanly instead of leaking their threads. + +impl Client { + /// AUTH as an already-registered identity (the driver registers once). + fn auth(&mut self, username: &str, key: &SigningKey) { + let verifying = key.verifying_key(); + let identity: &[u8] = verifying.as_bytes(); + let mut auth_msg = LABEL_AUTH.to_vec(); + auth_msg.extend_from_slice(&self.handshake_hash); + let mut body = vec![username.len() as u8]; + body.extend_from_slice(username.as_bytes()); + body.extend_from_slice(identity); + body.extend_from_slice(&key.sign(&auth_msg).to_bytes()); + body.push(0); + body.extend_from_slice(&0u16.to_be_bytes()); + let (op, resp) = self.request(OP_AUTH, &body); + assert_eq!(op, OP_AUTH); + assert_eq!(resp[0], 0, "AUTH ok"); + } + + /// Pages FETCH from `cursor` to the end, returning every message id. + fn fetch_all(&mut self, cursor: (i64, [u8; 32])) -> Vec<[u8; 32]> { + let (mut after_at, mut after_id) = cursor; + let mut ids = Vec::new(); + loop { + let mut body = Vec::new(); + body.extend_from_slice(&after_at.to_be_bytes()); + body.extend_from_slice(&after_id); + let (op, resp) = self.request(OP_FETCH, &body); + assert_eq!(op, OP_FETCH); + assert_eq!(resp[0], 0, "FETCH ok"); + let count = u16::from_be_bytes(resp[1..3].try_into().unwrap()); + if count == 0 { + return ids; + } + let mut off = 3; + for _ in 0..count { + let id: [u8; 32] = resp[off..off + 32].try_into().unwrap(); + off += 32; + let received_at = i64::from_be_bytes(resp[off..off + 8].try_into().unwrap()); + off += 8; + off += 1; // unsolicited flag + let len = u32::from_be_bytes(resp[off..off + 4].try_into().unwrap()) as usize; + off += 4 + len; + ids.push(id); + // The cursor is (received_at, id) strictly after the last + // delivered record; a future-dated probe terminates cleanly. + after_at = received_at; + after_id = id; + } + } + } +} + +/// The driver's mailbox: an envelope addressed to a registered identity. +fn envelope_to(recipient: &[u8], payload_len: usize) -> Vec { + let mut envelope = Vec::with_capacity(69 + payload_len + 16); + envelope.extend_from_slice(ENVELOPE_MAGIC); + envelope.push(ENVELOPE_VERSION); + envelope.extend_from_slice(recipient); + envelope.resize(69 + payload_len, 0x42); // epk + ciphertext + tag + envelope +} + +#[test] +fn dozen_sessions_run_concurrent_fetch_cursors() { + let target = Target::start(768 << 10); + let key = SigningKey::from_bytes(&[8u8; 32]); + let verifying = key.verifying_key(); + let identity: &[u8] = verifying.as_bytes(); + + // Register the driver identity from a first session. + let mut reg = Client::connect(target.port); + reg.handshake().unwrap(); + let mut pop = LABEL_REGISTER.to_vec(); + pop.extend_from_slice(&target.static_public); + pop.extend_from_slice(b"driver"); + pop.extend_from_slice(identity); + let mut body = vec![6]; + body.extend_from_slice(b"driver"); + body.extend_from_slice(identity); + body.extend_from_slice(&key.sign(&pop).to_bytes()); + body.push(0); + body.push(0); + let (op, resp) = reg.request(OP_REGISTER, &body); + assert_eq!((op, resp.as_slice()), (OP_REGISTER, &[0u8][..])); + + // Seed 20 envelopes for the driver mailbox; ids are deterministic. + let mut expected = std::collections::HashSet::new(); + for i in 0..20u8 { + let envelope = envelope_to(identity, 40 + i as usize); + let mut body = vec![0u8]; // no accept-token MAC + body.extend_from_slice(&envelope); + let (op, resp) = reg.request(OP_SEND, &body); + assert_eq!((op, resp[0]), (OP_SEND, 0), "seed send {i}"); + let id: [u8; 32] = resp[1..33].try_into().unwrap(); + expected.insert(id); + } + + // A dozen sessions, each authenticated and fetching the whole mailbox + // concurrently; all must see the same 20 ids (nothing is deleted, so + // every session's cursor walk is independent and complete). + let mut handles = Vec::new(); + for _ in 0..12 { + let port = target.port; + let session_key = key.clone(); + handles.push(std::thread::spawn(move || { + let mut client = Client::connect(port); + client.handshake().unwrap(); + client.auth("driver", &session_key); + client.fetch_all((0, [0u8; 32])) + })); + } + for handle in handles { + let ids = handle.join().unwrap(); + let got: std::collections::HashSet<[u8; 32]> = ids.into_iter().collect(); + assert_eq!(got, expected, "every session saw the whole mailbox"); + } + target.assert_alive(); +} + +/// Held-open sessions must be closed by the server's idle timeout, not +/// leaked. Slow by nature: it waits out the full IDLE_TIMEOUT_SECS. +#[test] +#[ignore = "holds a dozen sessions past the 120 s idle timeout"] +fn idle_sessions_are_reaped_by_the_server() { + let target = Target::start(768 << 10); + let key = SigningKey::from_bytes(&[9u8; 32]); + let verifying = key.verifying_key(); + let identity: &[u8] = verifying.as_bytes(); + + let mut reg = Client::connect(target.port); + reg.handshake().unwrap(); + let mut pop = LABEL_REGISTER.to_vec(); + pop.extend_from_slice(&target.static_public); + pop.extend_from_slice(b"idler"); + pop.extend_from_slice(identity); + let mut body = vec![5]; + body.extend_from_slice(b"idler"); + body.extend_from_slice(identity); + body.extend_from_slice(&key.sign(&pop).to_bytes()); + body.push(0); + body.push(0); + let (op, resp) = reg.request(OP_REGISTER, &body); + assert_eq!((op, resp.as_slice()), (OP_REGISTER, &[0u8][..])); + + // A dozen sessions authenticate, then say nothing at all. + let mut clients = Vec::new(); + for _ in 0..12 { + let mut client = Client::connect(target.port); + client.handshake().unwrap(); + client.auth("idler", &key); + clients.push(client); + } + + // Past the idle timeout the server closes them; the client sees EOF. + std::thread::sleep(std::time::Duration::from_secs(125)); + for client in &mut clients { + let mut sink = [0u8; 16]; + assert!( + client.stream.read(&mut sink).unwrap_or(0) == 0, + "server closed the idle session" + ); + } + target.assert_alive(); +} + +/// A frame started but not finished must not hold a thread for the full +/// session idle timeout: the header deadline (30 s) closes it first. +#[test] +#[ignore = "waits out the 30 s header deadline"] +fn partial_frame_is_dropped_after_the_header_deadline() { + let target = Target::start(768 << 10); + let mut client = Client::connect(target.port); + client.handshake().unwrap(); + // Three bytes of a frame header, then silence. Before this hardening + // the connection idled the full 120 s; the header deadline must close + // it well before that. + client.raw_frame_bytes(&[0x00, 0x10, 0x00]); + let mut sink = [0u8; 16]; + assert!( + client.stream.read(&mut sink).unwrap_or(0) == 0, + "server dropped the stalled frame" + ); + target.assert_alive(); +} diff --git a/src/main.rs b/src/main.rs index 34ab05c..5ec7d66 100644 --- a/src/main.rs +++ b/src/main.rs @@ -3,6 +3,8 @@ mod bind; mod channel; mod crypto; +#[cfg(test)] +mod harness; mod proto; mod ratelimit; #[cfg(feature = "rns")] @@ -101,6 +103,9 @@ struct RnsServeArgs { rns_rate_link_requests: u32, #[arg(long = "rns-rate-link-bytes", default_value_t = 1 << 20)] rns_rate_link_bytes: u64, + /// Reap RNS links idle for this many seconds (0 disables) + #[arg(long = "rns-link-idle", default_value_t = 300)] + rns_link_idle: u64, /// UDP interface to listen on, host[:port] #[arg(long = "rns-udp", default_value = "127.0.0.1:4242")] rns_udp: String, @@ -200,6 +205,7 @@ fn main() -> anyhow::Result<()> { max_links: rns.rns_max_links, rate_link_requests: rns.rns_rate_link_requests, rate_link_bytes: rns.rns_rate_link_bytes, + link_idle_secs: rns.rns_link_idle, udp_listen_host: listen_host, udp_listen_port: listen_port, udp_forward_host: forward_host, diff --git a/src/proto.rs b/src/proto.rs index f080386..d5c841b 100644 --- a/src/proto.rs +++ b/src/proto.rs @@ -51,6 +51,11 @@ pub const FETCH_BUDGET: usize = 512 * 1024; // must stay under MAX_FRAME #[cfg(feature = "rns")] pub const AUTH_FULL_TOKENS: usize = 164; pub const IDLE_TIMEOUT_SECS: u64 = 120; +/// Budget for completing a handshake or a frame already started. Without it +/// a connection that sends a partial frame header and goes silent holds a +/// thread for the full idle timeout — bounded by the connection limiter, +/// but a free hold-and-wait for the hostile kind of client. +pub const HEADER_TIMEOUT_SECS: u64 = 30; pub const PURGE_INTERVAL_SECS: u64 = 60; /// A peer sent something unparseable. Always answered with MALFORMED. diff --git a/src/rns.rs b/src/rns.rs index ac4eb6a..279329a 100644 --- a/src/rns.rs +++ b/src/rns.rs @@ -12,6 +12,7 @@ use std::collections::{HashMap, HashSet}; use std::ffi::CString; use std::sync::{Mutex, OnceLock}; +use std::time::{Duration, Instant}; use crate::bind::TransportBindValues; use crate::proto::{AUTH_FULL_TOKENS, INTERNAL_ERROR, MALFORMED, RATE_LIMITED, TOO_LARGE}; @@ -59,6 +60,7 @@ pub struct RnsArgs { pub udp_listen_port: u16, pub udp_forward_host: Option, pub udp_forward_port: u16, + pub link_idle_secs: u64, pub max_tokens: u16, pub main_quota: i64, pub requests_quota: i64, @@ -71,6 +73,10 @@ struct RnsState { destination: [u8; 16], sessions: Mutex>>, links: Mutex>, + /// Last-seen instant per link, the reaper's clock. Updated before the + /// limiters so a client still sending — even one being refused — counts + /// as alive; only a silent link is reapable. + last_activity: Mutex>, request_limiter: RateLimiter, byte_limiter: ByteRateLimiter, max_links: usize, @@ -117,6 +123,7 @@ pub fn start(args: RnsArgs) -> anyhow::Result { destination, sessions: Mutex::new(HashMap::new()), links: Mutex::new(HashSet::new()), + last_activity: Mutex::new(HashMap::new()), request_limiter: RateLimiter::new(args.rate_link_requests), byte_limiter: ByteRateLimiter::new(args.rate_link_bytes), max_links: args.max_links, @@ -124,6 +131,13 @@ pub fn start(args: RnsArgs) -> anyhow::Result { }; STATE.get_or_init(|| state); + // microReticulum never times a link out and the shim has no server-side + // close, so a client that dies mid-link would hold its Store handle and + // maxLinks slot forever; 0 keeps the old never-expire behavior. + if args.link_idle_secs > 0 { + std::thread::spawn(move || reap_loop(Duration::from_secs(args.link_idle_secs))); + } + let storage_dir = CString::new(args.storage_dir).unwrap(); let listen_host = CString::new(args.udp_listen_host).unwrap(); let forward_host = args.udp_forward_host.map(|h| CString::new(h).unwrap()); @@ -185,6 +199,9 @@ fn handle_request(request: &[u8], link_id: &[u8; 16]) -> Vec { // an AUTH carrying a full token set, whichever is larger. let max_request = (state.config.max_envelope + 34) .max(AUTH_FULL_TOKENS + 32 * state.config.max_tokens as usize); + // Any request, even one about to be refused, proves the link is alive. + state.last_activity.lock().unwrap().insert(*link_id, Instant::now()); + if request.len() > max_request { log::warn!( "request of {} bytes over {} byte cap", @@ -194,6 +211,9 @@ fn handle_request(request: &[u8], link_id: &[u8; 16]) -> Vec { return response(TOO_LARGE); } if !state.request_limiter.allow(&link) || !state.byte_limiter.allow(&link, request.len()) { + // Pre-dispatch refusals never reach the session metrics, so they get + // their own line or the journal undercounts a spinning client. + log::debug!("link {link} rate limited: request refused"); return response(RATE_LIMITED); } @@ -264,6 +284,45 @@ extern "C" fn smolmail_rns_take_response(out: *mut u8, cap: usize) -> usize { } } +/// Links whose last activity is at least `idle` old, with their age. +fn expired_links( + last_activity: &HashMap<[u8; 16], Instant>, + now: Instant, + idle: Duration, +) -> Vec<([u8; 16], Duration)> { + last_activity + .iter() + .filter(|(_, seen)| now.duration_since(**seen) >= idle) + .map(|(id, seen)| (*id, now.duration_since(*seen))) + .collect() +} + +/// Evicts links that have gone silent: microReticulum never times a link +/// out and the shim has no server-side close, so a client that dies +/// mid-link would otherwise hold its Store handle and maxLinks slot +/// forever. A reaped link that somehow speaks again is treated as a fresh +/// session (unauthenticated until the next AUTH), which is safe. +fn reap_loop(idle: Duration) { + let interval = (idle / 4).clamp(Duration::from_secs(1), Duration::from_secs(60)); + loop { + std::thread::sleep(interval); + let Some(state) = STATE.get() else { + return; + }; + let expired = expired_links(&state.last_activity.lock().unwrap(), Instant::now(), idle); + for (link, age) in expired { + state.links.lock().unwrap().remove(&link); + state.sessions.lock().unwrap().remove(&link); + state.last_activity.lock().unwrap().remove(&link); + log::debug!( + "link {} reaped after {:?} idle", + data_encoding::HEXLOWER.encode(&link), + age + ); + } + } +} + #[no_mangle] extern "C" fn smolmail_rns_on_link_opened(link_id: *const u8) -> i32 { let Some(state) = STATE.get() else { @@ -280,6 +339,10 @@ extern "C" fn smolmail_rns_on_link_opened(link_id: *const u8) -> i32 { return 1; } links.insert(link); + if let Some(state) = STATE.get() { + state.last_activity.lock().unwrap().insert(link, Instant::now()); + } + log::debug!("link {} opened ({} open)", data_encoding::HEXLOWER.encode(&link), links.len()); 0 } @@ -289,6 +352,8 @@ extern "C" fn smolmail_rns_on_link_closed(link_id: *const u8) { let link: [u8; 16] = unsafe { std::slice::from_raw_parts(link_id, 16).try_into().unwrap() }; state.links.lock().unwrap().remove(&link); state.sessions.lock().unwrap().remove(&link); + state.last_activity.lock().unwrap().remove(&link); + log::debug!("link {} closed", data_encoding::HEXLOWER.encode(&link)); } } @@ -309,4 +374,18 @@ mod tests { "799855f4955f1b09fd20a13cd84f4e71" ); } + + #[test] + fn expired_links_selects_only_silent_ones() { + let mut last_activity = HashMap::new(); + let now = Instant::now(); + let idle = Duration::from_secs(300); + last_activity.insert([1u8; 16], now - Duration::from_secs(301)); // silent + last_activity.insert([2u8; 16], now - Duration::from_secs(299)); // active + last_activity.insert([3u8; 16], now); // just seen + let expired = expired_links(&last_activity, now, idle); + assert_eq!(expired.len(), 1); + assert_eq!(expired[0].0, [1u8; 16]); + assert!(expired[0].1 >= idle); + } } diff --git a/src/server.rs b/src/server.rs index 49a33de..50389ca 100644 --- a/src/server.rs +++ b/src/server.rs @@ -7,7 +7,7 @@ use std::time::Duration; use crate::bind::TransportBindValues; use crate::channel::{handshake, Channel}; use crate::crypto::{b32, derive_public}; -use crate::proto::{FETCH_BUDGET, IDLE_TIMEOUT_SECS, KEY_LEN, MALFORMED, PURGE_INTERVAL_SECS}; +use crate::proto::{FETCH_BUDGET, HEADER_TIMEOUT_SECS, KEY_LEN, MALFORMED, PURGE_INTERVAL_SECS}; use crate::ratelimit::RateLimiter; use crate::session::{ServerConfig, Session}; use crate::store::Store; @@ -94,15 +94,21 @@ pub fn run(args: ServeArgs) -> anyhow::Result<()> { Ok(()) } -fn handle_connection( +/// 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( mut stream: TcpStream, config: &ServerConfig, static_key: &[u8; KEY_LEN], db_path: &str, peer_ip: &str, ) -> anyhow::Result<()> { - stream.set_read_timeout(Some(Duration::from_secs(IDLE_TIMEOUT_SECS)))?; + // The handshake runs under the header budget; once it completes, + // Channel::read_noise picks between the header and idle deadlines per + // frame state. + stream.set_read_timeout(Some(Duration::from_secs(HEADER_TIMEOUT_SECS)))?; + let handshake_started = std::time::Instant::now(); let (transport, handshake_hash) = match handshake(&mut stream, static_key) { Ok(v) => v, Err(e) => { @@ -110,6 +116,17 @@ fn handle_connection( return Ok(()); } }; + log::debug!( + "handshake from {peer_ip} in {:?}", + handshake_started.elapsed() + ); + + // Drop logs the summary once, whatever return path closes the session. + let summary = SessionSummary { + peer_ip, + started: std::time::Instant::now(), + ops: std::cell::Cell::new(0), + }; let store = Store::open(db_path)?; let server_static = derive_public(static_key); @@ -129,6 +146,7 @@ fn handle_connection( return Ok(()); } }; + summary.ops.set(summary.ops.get() + 1); let (status, payload) = session.dispatch(op, &body); let mut response = Vec::with_capacity(1 + payload.len()); @@ -140,6 +158,23 @@ fn handle_connection( } } +struct SessionSummary<'a> { + peer_ip: &'a str, + started: std::time::Instant, + ops: std::cell::Cell, +} + +impl Drop for SessionSummary<'_> { + fn drop(&mut self) { + log::debug!( + "session {} closed: {} op(s) in {:?}", + self.peer_ip, + self.ops.get(), + self.started.elapsed() + ); + } +} + fn is_eof_like(e: &anyhow::Error) -> bool { if let Some(io_err) = e.downcast_ref::() { return matches!( diff --git a/src/session.rs b/src/session.rs index 680a293..c434435 100644 --- a/src/session.rs +++ b/src/session.rs @@ -44,6 +44,19 @@ impl From for HandlerError { type OpResult = Result<(u8, Vec), HandlerError>; +/// Wire op byte to log name; unknown bytes are logged as dispatched too. +fn op_name(op: u8) -> &'static str { + match op { + OP_AUTH => "AUTH", + OP_RESOLVE => "RESOLVE", + OP_SEND => "SEND", + OP_FETCH => "FETCH", + OP_DELETE => "DELETE", + OP_REGISTER => "REGISTER", + _ => "UNKNOWN", + } +} + pub struct Session<'a> { config: &'a ServerConfig, store: Store, @@ -72,7 +85,25 @@ impl<'a> Session<'a> { /// panicking or propagating errors to the caller: a bad frame or a /// storage error both become a response, and the caller decides /// separately whether to keep the connection open. + /// + /// Timing lives in this wrapper because both carriers dispatch through + /// it, so the metrics flag covers TCP and RNS from one place. pub fn dispatch(&mut self, op: u8, body: &[u8]) -> (u8, Vec) { + let started = std::time::Instant::now(); + let (status, payload) = self.dispatch_op(op, body); + log::debug!( + "op {} from {} in {:?}: status {}, {}B request, {}B response", + op_name(op), + self.peer_ip, + started.elapsed(), + status, + body.len(), + payload.len() + ); + (status, payload) + } + + fn dispatch_op(&mut self, op: u8, body: &[u8]) -> (u8, Vec) { if matches!(op, OP_FETCH | OP_DELETE) && self.username.is_none() { return (AUTH_REQUIRED, Vec::new()); }