From c614d8345fe0226f2354d3add9e208f5ad2b2e25 Mon Sep 17 00:00:00 2001 From: Michael Date: Wed, 30 Sep 2026 06:56:52 -0400 Subject: [PATCH] Protocol v2: X25519 handshake, sequence-number nonces, hardened transport MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Red-team review of the (byte-exact) v1 port found replay, cross-session frame portability, pre-auth allocation amplification, and a symmetric key that let either peer forge as itself — flaws inherent to the v1 wire format. v2 closes them: - fixed 69-byte OVX2 hello; server validates before replying; v1 and wrong-version peers see an immediate close - HKDF(ephemeral X25519 DH, PSK, transcript) -> direction-separated session keys; forward secrecy per connection - strictly increasing u64 sequence nonces: any replay or reorder drops the connection (regression tests in tests/protocol_v2.rs) - Config: message-size cap (64 KiB default), connection semaphore, idle/body-stall/handshake/send timeouts on both roles (client-side timers retire the old no-client-timeouts quirk) - message processing inline per connection; user callbacks panic- isolated; poison-tolerant mutexes; demo key refused in release builds - PROTOCOL-V2.md supersedes PARITY-CONTRACT.md (retired, kept as port history); Cereal payload encoding is unchanged Payload serialization remains byte-exact with the original C++ (the golden scrypt/cereal vectors still govern). Zero unsafe. Verified live: server<->client handshake, encrypted echo, clean teardown. Cuts over from the C++ implementation, which is deleted (phase 5 of the port contract); it remains in the original gitea repository. --- Cargo.lock | 118 +++++++- Cargo.toml | 8 +- PARITY-CONTRACT.md | 9 + PROTOCOL-V2.md | 114 ++++++++ README.md | 71 ++--- examples/omnivox-client.rs | 10 +- examples/omnivox-server.rs | 12 +- src/crypto.rs | 23 ++ src/handshake.rs | 171 +++++++++++ src/lib.rs | 581 ++++++++++++++++++++++++++----------- tests/loopback.rs | 259 ++++++++--------- tests/protocol_v2.rs | 259 +++++++++++++++++ tests/support/mod.rs | 42 +++ 13 files changed, 1325 insertions(+), 352 deletions(-) create mode 100644 PROTOCOL-V2.md create mode 100644 src/handshake.rs create mode 100644 tests/protocol_v2.rs create mode 100644 tests/support/mod.rs diff --git a/Cargo.lock b/Cargo.lock index 94bdc0f..32d278f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -100,6 +100,32 @@ dependencies = [ "typenum", ] +[[package]] +name = "curve25519-dalek" +version = "4.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "97fb8b7c4503de7d6ae7b42ab72a5a59857b4c937ec27a3d4539dba95b5ab2be" +dependencies = [ + "cfg-if", + "cpufeatures", + "curve25519-dalek-derive", + "fiat-crypto", + "rustc_version", + "subtle", + "zeroize", +] + +[[package]] +name = "curve25519-dalek-derive" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f46882e17999c6cc590af592290432be3bce0428cb0d5f8b6715e4dc7b383eb3" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "digest" version = "0.10.7" @@ -121,6 +147,12 @@ dependencies = [ "windows-sys", ] +[[package]] +name = "fiat-crypto" +version = "0.2.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "28dea519a9695b9977216879a3ebfddf92f1c08c05d984f8996aecd6ecdc811d" + [[package]] name = "generic-array" version = "0.14.7" @@ -154,6 +186,15 @@ dependencies = [ "wasip2", ] +[[package]] +name = "hkdf" +version = "0.12.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b5f8eb2ad728638ea2c7d47a21db23b7b58a72ed6a38256b8a1849f15fbbdf7" +dependencies = [ + "hmac", +] + [[package]] name = "hmac" version = "0.12.1" @@ -221,15 +262,19 @@ dependencies = [ [[package]] name = "omnivox" -version = "0.13.0" +version = "0.14.0" dependencies = [ "chacha20poly1305", "getrandom 0.3.4", + "hkdf", + "rand_core", "scrypt", + "sha2", "thiserror", "tokio", "tracing", "tracing-subscriber", + "x25519-dalek", ] [[package]] @@ -347,6 +392,15 @@ dependencies = [ "bitflags", ] +[[package]] +name = "rustc_version" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cfcb3a22ef46e85b45de6ee7e79d063319ebb6594faafcf1c225ea92ab6e9b92" +dependencies = [ + "semver", +] + [[package]] name = "salsa20" version = "0.10.2" @@ -374,6 +428,42 @@ dependencies = [ "sha2", ] +[[package]] +name = "semver" +version = "1.0.28" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8a7852d02fc848982e0c167ef163aaff9cd91dc640ba85e263cb1ce46fae51cd" + +[[package]] +name = "serde" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba" +dependencies = [ + "serde_core", + "serde_derive", +] + +[[package]] +name = "serde_core" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.6", +] + [[package]] name = "sha2" version = "0.10.9" @@ -632,8 +722,34 @@ version = "0.57.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e" +[[package]] +name = "x25519-dalek" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c7e468321c81fb07fa7f4c636c3972b9100f0346e5b6a9f2bd0603a52f7ed277" +dependencies = [ + "curve25519-dalek", + "rand_core", + "serde", + "zeroize", +] + [[package]] name = "zeroize" version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e" +dependencies = [ + "zeroize_derive", +] + +[[package]] +name = "zeroize_derive" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3c50655cbb0fe3fc43170059e702f1ce5e19b84cec58dc87b037a09935c2f328" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] diff --git a/Cargo.toml b/Cargo.toml index a7f6320..45c2d9c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,13 +1,17 @@ [package] name = "omnivox" -version = "0.13.0" +version = "0.14.0" edition = "2024" license = "GPL-3.0-or-later" -description = "Encrypted async TCP client-server messaging (ChaCha20-Poly1305), Rust port of the OmniVox C++ library" +description = "Encrypted async TCP client-server messaging: X25519 handshake, ChaCha20-Poly1305, replay-proof sequencing (OmniVox protocol v2)" [dependencies] chacha20poly1305 = "0.10" getrandom = "0.3" +hkdf = "0.12" +rand_core = { version = "0.6", features = ["getrandom"] } +sha2 = "0.10" +x25519-dalek = { version = "2", features = ["zeroize"] } scrypt = "0.11" thiserror = "2" tokio = { version = "1", features = ["rt-multi-thread", "net", "io-util", "time", "sync", "macros"] } diff --git a/PARITY-CONTRACT.md b/PARITY-CONTRACT.md index c225760..2525aab 100644 --- a/PARITY-CONTRACT.md +++ b/PARITY-CONTRACT.md @@ -1,3 +1,12 @@ +> **RETIRED 2026-09.** The port it governed was completed and proven +> (all golden vectors + both cross-directions passed), then superseded by +> protocol v2 (`PROTOCOL-V2.md`) in 0.14.0: a red-team review found replay, +> frame portability, and pre-auth allocation flaws baked into the v1 wire +> format, unfixable behind parity. The v2 handshake rejects v1 traffic. +> This file remains as the historical record of the port — what was +> preserved, what was dropped, and the C++ byte formats the `Cereal` codec +> still speaks for payloads. + # OmniVox Port — Parity Contract Source: `omnivox/` (C++17/20, asio + cereal + CryptoPP + fmt) diff --git a/PROTOCOL-V2.md b/PROTOCOL-V2.md new file mode 100644 index 0000000..3a32f99 --- /dev/null +++ b/PROTOCOL-V2.md @@ -0,0 +1,114 @@ +# OmniVox Protocol v2 — Normative Specification + +Version 2, 2026-09. Supersedes the v1 wire format (byte-compatible with the +original C++ implementation); v1 traffic is rejected at the handshake and +`PARITY-CONTRACT.md` is retired alongside it. + +## Threat model + +The PSK (a scrypt hash of a shared password) authenticates peers. The network +is hostile: it can observe, inject, replay, reorder, and truncate. Goals: +session confidentiality and integrity, replay protection, forward secrecy +for recorded traffic, and no attacker-controlled work before authentication. + +## Connection lifecycle + +``` +TCP connect + │ + ├── HELLO exchange (plaintext, fixed size, timeout: handshake_timeout_ms) + ├── key schedule (HKDF over ephemeral DH + PSK) + ├── framed messages (AEAD, per-direction sequence numbers) + └── close (any timeout / violation closes immediately) +``` + +## HELLO (both directions, 69 bytes, no attacker-controlled lengths) + +``` +bytes 0..4 magic "OVX2" +byte 4 version 2 +bytes 5..37 x25519 ephemeral public key +bytes 37..69 session nonce (32 random bytes) +``` + +The client sends first. The server validates magic, version, and key +encoding **before replying**; any violation closes the connection. The +server rejects an all-zero DH output (small-order contribution). + +## Key schedule + +``` +dh = X25519(eph_local, eph_peer) +prk = HKDF-SHA256-Extract(salt = nonce_client ‖ nonce_server, ikm = PSK) +okm = HKDF-SHA256-Expand(prk, info = "omnivox-v2" ‖ hello_client ‖ hello_server, 64) +key_c2s = okm[0..32] key_s2c = okm[32..64] +``` + +The client's send key is `key_c2s`, receive key `key_s2c`; reversed for the +server. Direction separation means neither peer can present itself forged +frames (v1's single symmetric key could). The ephemeral DH gives forward +secrecy against later PSK compromise; the PSK authenticates against +off-path attackers. A weak password remains guessable by an active attacker +with the PSK budget to grind scrypt — use random 32-byte PSKs; a PAKE is the +future upgrade if passwords must stay human. + +## Framing (both directions, after key confirmation) + +``` +u32 LE length (0 < len ≤ max_message_size, default cap 64 KiB) +AeadPacket: + nonce [u8; 12] four zero bytes ‖ u64 BE sequence number + tag [u8; 16] Poly1305 tag + len u64 LE (cereal vector length, retained for payload lineage) + ct [len] ChaCha20-Poly1305 ciphertext of the serialized payload +``` + +AEAD is ChaCha20-Poly1305 (RFC 8439) with the packet's own nonce as the IETF +nonce and empty AAD — identical to v1's construction, now keyed per session +per direction. + +### Sequence discipline (the replay defense) + +Each sender's counter starts at 1 and increments by exactly 1 per frame; a +failed write must not consume a number. The receiver's first expected value +is 1. A frame whose decoded nonce is not the expected value is a protocol +violation: **log and close the connection.** There is no reorder window. + +### Payload encoding + +The plaintext is the `Cereal` binary encoding of the message type — raw +little-endian fields, `u64` length prefixes for byte vectors, no archive +header — unchanged from the C++ original. + +## Limits and timeouts (all server- and client-side) + +| Knob | Default | Violation | +|---|---|---| +| handshake_timeout_ms | 10000 | close | +| idle_timeout_ms (silence between messages) | 15000 | close | +| body_stall_ms (mid-body silence) | 5000 | close | +| send_timeout_ms (peer not reading) | 5000 | close that peer | +| max_message_size | 65536 | close on header | +| max_connections (server) | 1024 | new connections wait | + +Corrupt frames with the *correct* sequence number (failed AEAD or payload +decode) are logged and skipped; the connection continues. Anything else — +sequence violations, unreasonable headers, stalls — is a disconnect. + +## Security properties + +| Property | Mechanism | Tested by | +|---|---|---| +| Confidentiality | ChaCha20-Poly1305 per-session keys | `protocol_v2::frames_from_another_session_are_undecryptable` | +| Replay defense | strict sequence numbers | `protocol_v2::replayed_frame_is_dropped_and_the_connection_dies` | +| Session binding | nonces enter the salt/info | same cross-session test | +| Forward secrecy | ephemeral X25519 per connection | design (single-connection compromise leaks only it) | +| No pre-auth work | fixed 69-byte hello, caps only after key confirmation | `wrong_version_hello_is_closed`, `v1_traffic_is_rejected_at_the_handshake` | +| Liveness both roles | idle + stall timeouts client **and** server | `malicious_server_silence_times_out_the_client`, `header_stall_times_out` | +| Slow-client containment | per-connection locks + send timeout | design (no shared writer mutex) | +| DoS containment | connection semaphore, message caps, callback panic isolation | `panicking_callback_does_not_kill_the_connection` | + +## Refusals + +Release builds refuse to start with the original demo key (scrypt of +`SecurePassword123` / salt `01…08`) unless `allow_insecure_defaults` is set. diff --git a/README.md b/README.md index 8e6e6a8..63025ac 100644 --- a/README.md +++ b/README.md @@ -1,57 +1,60 @@ # omnivox-rs -Rust port of [OmniVox](https://git.0xfeedc0de.com/michael/OmniVox), a -templated encrypted TCP client-server library — **wire-compatible -byte-for-byte** with the original C++ implementation. A Rust client talks -to a C++ server and vice versa. +Encrypted async TCP client-server messaging — protocol **v2**. A Rust +library plus example binaries, descended from the OmniVox C++ library and +hardened past it after a full red-team review of v1. -- Async TCP built on tokio -- ChaCha20-Poly1305 authenticated encryption (RFC 8439) -- scrypt key derivation (RFC 7914, `N = 2^17, r = 8, p = 1`) -- Byte-exact cereal `BinaryOutputArchive` field layout -- Zero `unsafe` +- X25519 handshake on every connection → forward secrecy, session-bound keys +- ChaCha20-Poly1305 with strictly increasing sequence numbers → replay-proof +- Direction-separated traffic keys, fixed-size pre-auth surface +- scrypt (RFC 7914) password-based key derivation; `Cereal` payload encoding + (byte-exact with the original's cereal binary format) for data portability +- tokio transport: connection caps, idle/stall/send timeouts, panic-isolated + callbacks; zero `unsafe` -## Wire format - -Framing is a `u32` little-endian length prefix. Each message is a -`AeadPacket`: `nonce[12] || tag[16] || u64 len || ciphertext`. Payloads -implement the `Cereal` trait, which reproduces cereal's raw-LE field -encoding (see `PARITY-CONTRACT.md` for the normative spec). +**Not wire-compatible with the C++ original or v1** — by design; v1 peers +are refused at the handshake. See `PROTOCOL-V2.md` for the full normative +spec and `PARITY-CONTRACT.md` (retired) for the port's history. ## Usage ```rust -use omnivox::{OmniVox, Cereal, CodecError, Decision, derive_encrypt_key}; +use omnivox::{Config, Cereal, CodecError, Decision, OmniVox, derive_encrypt_key}; +use std::sync::Arc; + +let endpoint = OmniVox::::new(); +endpoint.set_encrypt_key(derive_encrypt_key(b"password", b"salt", 17, 8, 1)?); +endpoint.start_server(8080, Config::default(), on_data, on_disconnect, on_connect).await?; ``` -Create an `OmniVox` where `T: Cereal`, set the key (both peers must -agree), then use `start_server` / `connect_to_server` / `send_to_client` -/ `send_to_server`. Runnable ports of the original demo apps live in -`examples/`. +Payload types implement `Cereal` (raw little-endian fields, `u64` vector +lengths — the original's serialization). Runnable demos: `examples/`. -## Parity +## Security posture -`PARITY-CONTRACT.md` is the definition of done: it names the unit of -parity, the accepted differences, and the original's quirks this port -deliberately preserves (B1–B3, B5) rather than silently fixing. Parity -evidence: +| v1 weakness | v2 answer | +|---|---| +| captured frames replay forever | strict sequence numbers; a replay kills the connection | +| frames portable across connections | keys bound to per-session nonces | +| 5 MiB pre-auth allocation from 8 bytes | fixed 69-byte handshake before any attacker-sized read | +| one symmetric key, either peer can forge as itself | direction-separated HKDF session keys | +| static key, no forward secrecy | ephemeral X25519 per connection | +| no client-side timeouts | full idle/stall discipline both roles | +| panicking user callback killed the read loop | callbacks panic-isolated | +| demo password in every repo checkout | release builds refuse the demo key | -- `tests/vectors.rs` — golden vectors generated by the C++ CryptoPP + - cereal stack (`reference/cppgen.cpp`); keys, serialization bytes, and - AEAD ciphertext/tags must match exactly. -- `tests/loopback.rs` — runtime behavior clauses of the contract. -- Differential cross-tests: this crate's binaries were verified talking - to the unmodified C++ binaries in both directions over loopback. +Use random 32-byte PSKs where you can; an active attacker can grind scrypt +against human passwords (PAKE planned). ## Build & test ```bash -cargo test # note: the scrypt vectors use N=2^17 (~1 s, 128 MiB per derivation) -cargo clippy --all-targets +cargo test # includes attack regressions: replay, version, panic, timeouts +cargo clippy --all-targets -- -D warnings cargo run --example omnivox-server # terminal 1 cargo run --example omnivox-client # terminal 2 ``` ## License -GPL-3.0-or-later, matching the original OmniVox. \ No newline at end of file +GPL-3.0-or-later. diff --git a/examples/omnivox-client.rs b/examples/omnivox-client.rs index d7ddc1f..fa33a4f 100644 --- a/examples/omnivox-client.rs +++ b/examples/omnivox-client.rs @@ -5,7 +5,7 @@ mod common; use common::{DatagramPayload, PASSWORD, SALT, SERVER_ADDR, SERVER_PORT}; -use omnivox::{ConnectionContext, ConnectionStatus, OmniVox, derive_encrypt_key}; +use omnivox::{Config, ConnectionContext, ConnectionStatus, OmniVox, derive_encrypt_key}; use std::sync::Arc; #[tokio::main] @@ -62,7 +62,13 @@ async fn main() -> Result<(), Box> { }); client - .connect_to_server(SERVER_ADDR, SERVER_PORT, Some(data_cb), connection_cb) + .connect_to_server( + SERVER_ADDR, + SERVER_PORT, + Config::default(), + Some(data_cb), + connection_cb, + ) .await?; tokio::time::sleep(std::time::Duration::from_secs(15)).await; diff --git a/examples/omnivox-server.rs b/examples/omnivox-server.rs index 9ace2f4..313ee0f 100644 --- a/examples/omnivox-server.rs +++ b/examples/omnivox-server.rs @@ -5,7 +5,7 @@ mod common; use common::{DatagramPayload, PASSWORD, SALT, SERVER_PORT}; -use omnivox::{ConnectionContext, Decision, OmniVox, derive_encrypt_key}; +use omnivox::{Config, ConnectionContext, Decision, OmniVox, derive_encrypt_key}; use std::{ collections::HashMap, sync::{Arc, Mutex}, @@ -18,7 +18,7 @@ async fn main() -> Result<(), Box> { .without_time() .init(); - tracing::info!("Starting server on port {SERVER_PORT} with 2 threads"); + tracing::info!("Starting server on port {SERVER_PORT}"); let server = Arc::new(OmniVox::::new()); let key = derive_encrypt_key(PASSWORD.as_bytes(), SALT, 17, 8, 1)?; @@ -78,7 +78,13 @@ async fn main() -> Result<(), Box> { }); server - .start_server(SERVER_PORT, 2, 15000, received, disconnect, connect) + .start_server( + SERVER_PORT, + Config::default(), + received, + disconnect, + connect, + ) .await?; tracing::info!("Server started successfully"); diff --git a/src/crypto.rs b/src/crypto.rs index b5ade1e..71a507f 100644 --- a/src/crypto.rs +++ b/src/crypto.rs @@ -136,3 +136,26 @@ pub fn decrypt(key: &[u8; KEY_LEN], packet: &AeadPacket) -> Result, Cryp .map_err(|_| CryptoError::AuthenticationFailed)?; Ok(payload) } + +/// v2 frame nonce: four zero bytes ‖ u64 big-endian sequence number. +/// Deterministic nonces make reuse impossible and turn the AEAD into a +/// replay detector (see `PROTOCOL-V2.md`). +#[must_use] +pub fn frame_nonce(counter: u64) -> [u8; NONCE_LEN] { + let mut n = [0u8; NONCE_LEN]; + n[4..].copy_from_slice(&counter.to_be_bytes()); + n +} + +/// Recovers the sequence number from a v2 frame nonce, or `None` if the +/// nonce is not in v2 format. +#[must_use] +pub fn nonce_counter(nonce: &[u8; NONCE_LEN]) -> Option { + if nonce[..4] == [0; 4] { + let mut c = [0u8; 8]; + c.copy_from_slice(&nonce[4..]); + Some(u64::from_be_bytes(c)) + } else { + None + } +} diff --git a/src/handshake.rs b/src/handshake.rs new file mode 100644 index 0000000..8608c25 --- /dev/null +++ b/src/handshake.rs @@ -0,0 +1,171 @@ +//! OmniVox v2 handshake and session key schedule. +//! +//! Both peers exchange a fixed-size `HELLO` (no attacker-controlled +//! lengths before key confirmation): +//! +//! ```text +//! "OVX2" ‖ u8 version ‖ x25519 pubkey[32] ‖ nonce[32] (69 bytes) +//! ``` +//! +//! The session keys are derived as +//! +//! ```text +//! dh = X25519(eph_local, eph_peer) +//! prk = HKDF-Extract(salt = nonce_client ‖ nonce_server, ikm = psk) +//! keys = HKDF-Expand(prk, "omnivox-v2" ‖ hello_client ‖ hello_server) +//! ``` +//! +//! `keys` is split into a client-to-server and a server-to-client half: +//! direction-separated keys, so a peer cannot present itself forged frames, +//! unlike v1's single symmetric key. The ephemeral pair gives forward +//! secrecy for recorded sessions; the PSK (scrypt of the shared password) +//! authenticates the exchange. + +use hkdf::Hkdf; +use sha2::Sha256; +use thiserror::Error; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; +use x25519_dalek::{EphemeralSecret, PublicKey}; + +/// Protocol magic at the start of every hello. +pub const MAGIC: &[u8; 4] = b"OVX2"; +/// Protocol version carried in the hello. +pub const VERSION: u8 = 2; +/// Exact hello length; nothing about a hello is attacker-length. +pub const HELLO_LEN: usize = 4 + 1 + 32 + 32; + +/// Errors surfaced from the handshake; any error closes the connection. +#[derive(Debug, Error)] +pub enum HandshakeError { + #[error("peer sent bad magic or unsupported version")] + BadVersion, + #[error("degenerate x25519 shared secret (rejected small-order peer key)")] + DegenerateKey, + #[error(transparent)] + Io(#[from] std::io::Error), +} + +/// A session's direction-separated traffic keys. +#[derive(Clone)] +pub struct Session { + /// Key for frames this peer sends. + pub send_key: [u8; 32], + /// Key for frames this peer expects to receive. + pub recv_key: [u8; 32], +} + +/// The v2 key schedule, given both hellos on the wire and the raw X25519 +/// DH output. Returns 64 bytes: c2s key ‖ s2c key. +/// +/// # Errors +/// +/// [`HandshakeError::DegenerateKey`] if the DH output is all-zero +/// (small-order peer contribution). +pub fn key_schedule( + client_hello: &[u8; HELLO_LEN], + server_hello: &[u8; HELLO_LEN], + shared_secret: &[u8; 32], + psk: &[u8; 32], +) -> Result<[u8; 64], HandshakeError> { + if shared_secret.iter().all(|&b| b == 0) { + return Err(HandshakeError::DegenerateKey); + } + let mut salt = [0u8; 64]; + salt[..32].copy_from_slice(&client_hello[37..]); + salt[32..].copy_from_slice(&server_hello[37..]); + + let hk = Hkdf::::new(Some(&salt), psk); + let mut info = Vec::with_capacity(10 + 2 * HELLO_LEN); + info.extend_from_slice(b"omnivox-v2"); + info.extend_from_slice(client_hello); + info.extend_from_slice(server_hello); + + let mut keys = [0u8; 64]; + hk.expand(&info, &mut keys) + .expect("64-byte output is within HKDF-SHA256 limits"); + Ok(keys) +} + +/// Splits key-schedule output into this peer's direction keys. +pub fn split_keys(keys: &[u8; 64], server: bool) -> Session { + let (c2s, s2c) = keys.split_at(32); + let (mut a, mut b) = ([0u8; 32], [0u8; 32]); + a.copy_from_slice(c2s); + b.copy_from_slice(s2c); + if server { + Session { + send_key: b, + recv_key: a, + } + } else { + Session { + send_key: a, + recv_key: b, + } + } +} + +/// Builds the exact 69-byte hello for a public key and session nonce. +pub fn build_hello(pubkey: &PublicKey, nonce: &[u8; 32]) -> [u8; HELLO_LEN] { + let mut hello = [0u8; HELLO_LEN]; + hello[..4].copy_from_slice(MAGIC); + hello[4] = VERSION; + hello[5..37].copy_from_slice(pubkey.as_bytes()); + hello[37..].copy_from_slice(nonce); + hello +} + +/// Parses and validates a hello's magic and version. +pub fn check_hello(hello: &[u8; HELLO_LEN]) -> Result { + if &hello[..4] != MAGIC || hello[4] != VERSION { + return Err(HandshakeError::BadVersion); + } + let mut bytes = [0u8; 32]; + bytes.copy_from_slice(&hello[5..37]); + Ok(PublicKey::from(bytes)) +} + +/// Performs the client or server side of the v2 handshake over an +/// already-connected stream pair, returning the session keys. +pub async fn perform( + reader: &mut R, + writer: &mut W, + psk: &[u8; 32], + server: bool, +) -> Result +where + R: AsyncRead + Unpin, + W: AsyncWrite + Unpin, +{ + let secret = EphemeralSecret::random_from_rng(rand_core::OsRng); + let pubkey = PublicKey::from(&secret); + let mut nonce = [0u8; 32]; + getrandom::fill(&mut nonce).expect("OS RNG failure"); + let our_hello = build_hello(&pubkey, &nonce); + + // The client speaks first; the server reads before replying, so the + // transcript order (client ‖ server) is unambiguous for both sides. + let mut peer_hello = [0u8; HELLO_LEN]; + if server { + // Validate the peer's hello BEFORE replying: a wrong-version or + // non-OmniVox peer must see a closed connection, not an echo. + reader.read_exact(&mut peer_hello).await?; + check_hello(&peer_hello)?; + writer.write_all(&our_hello).await?; + writer.flush().await?; + } else { + writer.write_all(&our_hello).await?; + writer.flush().await?; + reader.read_exact(&mut peer_hello).await?; + } + let peer_pub = check_hello(&peer_hello)?; + + let shared = secret.diffie_hellman(&peer_pub); + let (client_hello, server_hello) = if server { + (peer_hello, our_hello) + } else { + (our_hello, peer_hello) + }; + let keys = key_schedule(&client_hello, &server_hello, shared.as_bytes(), psk)?; + Ok(split_keys(&keys, server)) +} diff --git a/src/lib.rs b/src/lib.rs index b6fc16c..a4f4c5e 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,14 +1,21 @@ -//! OmniVox — encrypted async TCP client/server messaging. +//! OmniVox — encrypted async TCP client/server messaging, protocol v2. //! -//! Rust port of the OmniVox C++ library, wire-compatible byte-for-byte with -//! the original (framing, cereal field layout, scrypt + ChaCha20-Poly1305). -//! See `PARITY-CONTRACT.md` at the workspace root for the normative spec, -//! including the quirks this port deliberately preserves (B1, B2, B5). +//! Wire format and key schedule: see `PROTOCOL-V2.md`. In one paragraph: +//! a fixed-size X25519 hello handshake on every connection derives +//! direction-separated session keys via HKDF from the PSK (scrypt of the +//! shared password), then every frame is ChaCha20-Poly1305 sealed under the +//! sender's direction key with a strictly increasing sequence-number nonce. +//! Replay, frame portability across connections, and pre-auth attacker +//! allocations are structurally impossible; recorded sessions lose their +//! secrecy only together with the session's ephemeral keys. +//! +//! Payloads implement [`Cereal`], the byte-exact cereal binary field +//! encoding, unchanged from v1. //! //! # Example //! //! ```no_run -//! use omnivox::{Cereal, CodecError, ConnectionContext, Decision, OmniVox, OmniVoxError}; +//! use omnivox::{Cereal, CodecError, Config, ConnectionContext, Decision, OmniVox, OmniVoxError}; //! use std::sync::Arc; //! //! #[derive(Default)] @@ -25,8 +32,7 @@ //! server //! .start_server( //! 8080, -//! 2, -//! 15000, +//! Config::default(), //! Arc::new(|_data: &Ping, _ctx: &ConnectionContext| {}), //! Arc::new(|_ctx| {}), //! Arc::new(|_ctx| Decision::Allow), @@ -35,18 +41,21 @@ //! # Ok(()) } //! ``` -mod codec; -mod crypto; +pub mod codec; +pub mod crypto; +pub mod handshake; pub use codec::{Cereal, CodecError, Reader, write_byte_vec}; pub use crypto::{ AeadPacket, CryptoError, KEY_LEN, NONCE_LEN, SCRYPT_LOG_N, SCRYPT_P, SCRYPT_R, TAG_LEN, - decrypt, derive_encrypt_key, encrypt, encrypt_with_nonce, + decrypt, derive_encrypt_key, encrypt, encrypt_with_nonce, frame_nonce, nonce_counter, }; +pub use handshake::{HandshakeError, Session}; use std::{ collections::HashMap, net::SocketAddr, + panic::{AssertUnwindSafe, catch_unwind}, sync::{ Arc, Mutex, OnceLock, atomic::{AtomicBool, AtomicU32, Ordering}, @@ -60,16 +69,18 @@ use tokio::{ TcpListener, TcpStream, tcp::{OwnedReadHalf, OwnedWriteHalf}, }, - sync::Mutex as AsyncMutex, + sync::{Mutex as AsyncMutex, OwnedSemaphorePermit, Semaphore}, task::JoinHandle, time::{Duration, timeout}, }; -use tracing::{error, info}; +use tracing::{error, info, warn}; -/// 5 MiB per-message allocation cap, matching `MAX_MESSAGE_SIZE`. -pub const MAX_MESSAGE_SIZE: u32 = 1024 * 1024 * 5; -/// Fixed body-read stall cap, matching `MAX_TIMEOUT_STALL_MS`. -pub const MAX_TIMEOUT_STALL_MS: u64 = 5000; +/// scrypt of the original demo password + salt; servers and clients refuse +/// to run with it in release builds unless explicitly allowed. +const DEMO_KEY: [u8; KEY_LEN] = [ + 0xf6, 0x4f, 0x4c, 0x05, 0xeb, 0x28, 0x67, 0x6b, 0xe5, 0xe4, 0xdd, 0x76, 0x0c, 0x7f, 0x94, 0x09, + 0x32, 0x7a, 0x47, 0xe1, 0x1a, 0xbb, 0x66, 0xad, 0xc5, 0x93, 0xb2, 0xbd, 0x03, 0x9a, 0x06, 0x50, +]; /// State of a connection, matching the C++ `ConnectionStatus`. #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -83,7 +94,7 @@ pub enum ConnectionStatus { /// Per-connection metadata handed to every callback. /// /// `total_sent` mirrors the C++ `iTotalSent`, which the original never -/// increments; it is preserved as always-zero for parity. +/// incremented; it is preserved as always-zero. #[derive(Debug, Clone)] pub struct ConnectionContext { pub connected: ConnectionStatus, @@ -118,16 +129,53 @@ pub type DataHandler = Arc; pub type DecisionHandler = Arc Decision + Send + Sync + 'static>; -/// Library errors surfaced from setup calls. Runtime message failures are -/// logged and swallowed, matching the C++ exception swallowing (contract B2). +/// Runtime knobs. Everything has a conservative default; tune deliberately. +#[derive(Debug, Clone)] +pub struct Config { + /// Max silence between messages (or between handshake halves) before + /// the connection is dropped. Applies to server and client alike. + pub idle_timeout_ms: u64, + /// Max gap between bytes of one message body once its header arrived. + pub body_stall_ms: u64, + /// Max duration of a full frame write before the peer is dropped. + pub send_timeout_ms: u64, + /// Max duration of the v2 handshake. + pub handshake_timeout_ms: u64, + /// Hard cap on one message body; oversized headers disconnect. + pub max_message_size: u32, + /// Concurrent connections (server); excess connections wait. + pub max_connections: usize, + /// Allow the retired demo key in release builds. Not for production. + pub allow_insecure_defaults: bool, +} + +impl Default for Config { + fn default() -> Self { + Self { + idle_timeout_ms: 15_000, + body_stall_ms: 5_000, + send_timeout_ms: 5_000, + handshake_timeout_ms: 10_000, + max_message_size: 64 * 1024, + max_connections: 1024, + allow_insecure_defaults: false, + } + } +} + +/// Library errors surfaced from setup and security-violation paths. #[derive(Debug, thiserror::Error)] pub enum OmniVoxError { #[error(transparent)] Io(#[from] std::io::Error), #[error(transparent)] Crypto(#[from] CryptoError), - #[error("already started")] - AlreadyStarted, + #[error(transparent)] + Handshake(#[from] HandshakeError), + #[error( + "refusing to run with the retired demo key; set allow_insecure_defaults or change the password/salt" + )] + InsecureDefaults, } fn now_micros() -> u64 { @@ -137,24 +185,29 @@ fn now_micros() -> u64 { .unwrap_or(0) } +/// One peer's write side: half of the socket plus its traffic key and the +/// sequence number the next frame will carry. +struct WriterState { + half: OwnedWriteHalf, + key: [u8; KEY_LEN], + counter: u64, +} + struct ClientHandle { - /// Held across the full frame write, mirroring the C++ clients-mutex - /// hold across the blocking `asio::write` (contract B3). `Option` so a - /// disconnect can close the socket even while the read task still holds - /// the handle, matching the C++ `handleDisconnect` socket close. - writer: AsyncMutex>, + /// Held across the full frame write, so a slow reader stalls at most its + /// own connection, and at most `send_timeout_ms`. + writer: AsyncMutex>, context: ConnectionContext, } struct Inner { - key: Mutex>, + psk: Mutex>, + config: OnceLock, clients: Mutex>>, next_id: AtomicU32, stopping: AtomicBool, - /// Client-mode connection state; id 0 means "never connected", so the - /// C++ `ctx.id != serverContext.id` timer-gating reproduces exactly. server_ctx: Mutex>, - server_writer: AsyncMutex>, + server_writer: AsyncMutex>, received: OnceLock>, disconnected: OnceLock, connect_decision: OnceLock, @@ -163,9 +216,6 @@ struct Inner { } /// Encrypted TCP endpoint usable in either client or server role. -/// -/// Cheap to hand out: state is shared behind an `Arc`; clone or borrow the -/// same instance from callbacks. pub struct OmniVox { inner: Arc>, } @@ -176,13 +226,20 @@ impl Default for OmniVox { } } +fn check_defaults(config: &Config, psk: [u8; KEY_LEN]) -> Result<(), OmniVoxError> { + if !config.allow_insecure_defaults && !cfg!(debug_assertions) && psk == DEMO_KEY { + return Err(OmniVoxError::InsecureDefaults); + } + Ok(()) +} + impl OmniVox { - /// Creates an endpoint with no key set (traffic then uses the all-zero - /// key, reproducing the C++ default-constructed `SecByteBlock`). + /// Creates an endpoint with no key set. pub fn new() -> Self { Self { inner: Arc::new(Inner { - key: Mutex::new(None), + psk: Mutex::new(None), + config: OnceLock::new(), clients: Mutex::new(HashMap::new()), next_id: AtomicU32::new(1), stopping: AtomicBool::new(false), @@ -197,41 +254,50 @@ impl OmniVox { } } - /// Sets the ChaCha20-Poly1305 key. Until set, traffic uses the all-zero - /// key, which is what the C++ default-constructed `SecByteBlock` did. + /// Sets the pre-shared key for the v2 handshake. Both peers must set + /// the identical key (typically [`derive_encrypt_key`]). pub fn set_encrypt_key(&self, key: [u8; KEY_LEN]) { - *self.inner.key.lock().unwrap() = Some(key); + *self.inner.psk.lock().unwrap() = Some(key); } - fn key_bytes(&self) -> [u8; KEY_LEN] { - self.inner.key.lock().unwrap().unwrap_or([0u8; KEY_LEN]) + fn psk(&self) -> Option<[u8; KEY_LEN]> { + *self.inner.psk.lock().unwrap() } - /// Starts the server on `0.0.0.0:port`. - /// - /// `num_threads` is accepted for API parity with the C++ signature; the - /// tokio runtime's worker pool already provides the parallelism the C++ - /// `ThreadPool_` existed to add, so it configures nothing (contract: - /// thread structure is an accepted difference). - /// - /// `timeout_ms` is the per-client idle cap for waiting on a message - /// header. Errors only if already started; runtime failures are logged. + fn install_config(&self, config: Config) -> Config { + let _ = self.inner.config.set(config.clone()); + self.inner.config.get().expect("just set").clone() + } + + fn config(&self) -> Config { + self.inner.config.get().cloned().unwrap_or_default() + } + + /// Starts the server on `0.0.0.0:port`. Connections beyond + /// `config.max_connections` wait for a slot instead of being queued + /// without bound. pub async fn start_server( &self, port: u16, - num_threads: usize, - timeout_ms: u64, + config: Config, data: DataHandler, disconnect: ConnHandler, connect: DecisionHandler, ) -> Result<(), OmniVoxError> { - let _ = num_threads; + let config = self.install_config(config); + let psk = self + .psk() + .ok_or(OmniVoxError::Crypto(CryptoError::EmptyInput))?; + check_defaults(&config, psk)?; + let _ = self.inner.received.set(data); let _ = self.inner.disconnected.set(disconnect); let _ = self.inner.connect_decision.set(connect); let listener = TcpListener::bind(("0.0.0.0", port)).await?; + let permits = Arc::new(Semaphore::new(config.max_connections)); let accept_inner = Arc::clone(&self.inner); + let accept_psk = psk; let accept_task = tokio::spawn(async move { loop { let (stream, peer) = match listener.accept().await { @@ -242,9 +308,14 @@ impl OmniVox { continue; } }; + let permit = Arc::clone(&permits) + .acquire_owned() + .await + .expect("semaphore closed"); let conn_inner = Arc::clone(&accept_inner); + let cfg = accept_inner.config.get().cloned().unwrap_or_default(); let task = tokio::spawn(async move { - Self::serve_connection(conn_inner, stream, peer, timeout_ms).await; + Self::serve_connection(conn_inner, stream, peer, accept_psk, cfg, permit).await; }); accept_inner.tasks.lock().unwrap().push(task); } @@ -252,7 +323,8 @@ impl OmniVox { self.inner.tasks.lock().unwrap().push(accept_task); info!( - "[OMNIVOX] Server started on port {port} with {num_threads} threads and {timeout_ms} ms timeout" + "[OMNIVOX] Server v2 started on port {port} (idle {} ms, cap {} conns, {} ms stall)", + config.idle_timeout_ms, config.max_connections, config.body_stall_ms ); Ok(()) } @@ -261,9 +333,32 @@ impl OmniVox { inner: Arc>, mut stream: TcpStream, peer: SocketAddr, - timeout_ms: u64, + psk: [u8; KEY_LEN], + config: Config, + permit: OwnedSemaphorePermit, ) { - let id = inner.next_id.fetch_add(1, Ordering::Relaxed); + let session = { + let (mut r, mut w) = stream.split(); + match timeout( + Duration::from_millis(config.handshake_timeout_ms), + handshake::perform(&mut r, &mut w, &psk, true), + ) + .await + { + Err(_) => { + warn!("[OMNIVOX] handshake timed out from {peer}"); + return; + } + Ok(Err(e)) => { + warn!("[OMNIVOX] handshake failed from {peer}: {e}"); + return; + } + Ok(Ok(s)) => s, + } + }; + let (mut reader, mut writer) = stream.into_split(); + + let id = Self::claim_id(&inner); let mut ctx = ConnectionContext { connected: ConnectionStatus::Connected, id, @@ -276,25 +371,30 @@ impl OmniVox { let decision = inner .connect_decision .get() - .map(|f| f(&ctx)) + .map(|f| fire_decision(f, &ctx)) .unwrap_or(Decision::Allow); if decision == Decision::Deny { - let _ = stream.shutdown().await; + let _ = writer.shutdown().await; + drop(reader); ctx.connected = ConnectionStatus::Rejected; if let Some(f) = inner.disconnected.get() { - f(&ctx); + fire_ctx(f, &ctx); } info!( "[OMNIVOX] Client connection from {}:{} rejected", ctx.ip, ctx.port ); + drop(permit); return; } - let (mut reader, writer) = stream.into_split(); let handle = Arc::new(ClientHandle { - writer: AsyncMutex::new(Some(writer)), + writer: AsyncMutex::new(Some(WriterState { + half: writer, + key: session.send_key, + counter: 1, + })), context: ctx.clone(), }); inner @@ -307,135 +407,175 @@ impl OmniVox { ctx.ip, ctx.port ); - Self::read_loop(inner, handle, &mut reader, timeout_ms).await; + Self::read_loop(inner, handle, &mut reader, session.recv_key, &config).await; + drop(permit); } + fn claim_id(inner: &Inner) -> u32 { + let clients = inner.clients.lock().unwrap(); + loop { + let id = inner.next_id.fetch_add(1, Ordering::Relaxed); + if id != 0 && !clients.contains_key(&id) { + return id; + } + } + } + + #[allow(clippy::too_many_arguments)] async fn read_loop( inner: Arc>, handle: Arc, reader: &mut OwnedReadHalf, - timeout_ms: u64, + recv_key: [u8; KEY_LEN], + config: &Config, ) { - let ctx = &handle.context; + let ctx = handle.context.clone(); + let mut expected: u64 = 1; loop { - // Header: u32 LE message length, bounded by the client timeout. let mut header = [0u8; 4]; match timeout( - Duration::from_millis(timeout_ms), + Duration::from_millis(config.idle_timeout_ms), reader.read_exact(&mut header), ) .await { - Err(_elapsed) => { + Err(_) => { info!( "[OMNIVOX] Client {} {}:{} Timed out", ctx.id, ctx.ip, ctx.port ); - Self::handle_client_disconnect(&inner, ctx).await; + Self::handle_client_disconnect(&inner, &ctx).await; return; } Ok(Err(_)) => { - Self::handle_client_disconnect(&inner, ctx).await; + Self::handle_client_disconnect(&inner, &ctx).await; return; } Ok(Ok(_)) => {} } let size = u32::from_le_bytes(header); - // Contract B1, preserved: on an unreasonable size the C++ code - // disconnects but does NOT return — it still attempts the body - // read (a 0-byte read succeeds and its empty packet then fails - // to decrypt, which is logged, not fatal). - if size == 0 || size > MAX_MESSAGE_SIZE { + if size == 0 || size > config.max_message_size { + // v2: disconnect and RETURN (the v1 no-return quirk is retired). info!( "[OMNIVOX] Client {} {}:{} has an unreasonable header size!", ctx.id, ctx.ip, ctx.port ); - Self::handle_client_disconnect(&inner, ctx).await; + Self::handle_client_disconnect(&inner, &ctx).await; + return; } - // Body: fixed 5 s stall cap against slow-body attacks. let mut body = vec![0u8; size as usize]; match timeout( - Duration::from_millis(MAX_TIMEOUT_STALL_MS), + Duration::from_millis(config.body_stall_ms), reader.read_exact(&mut body), ) .await { - Err(_elapsed) => { + Err(_) => { info!( "[OMNIVOX] Client {} {}:{} had a stall during body read!", ctx.id, ctx.ip, ctx.port ); - Self::handle_client_disconnect(&inner, ctx).await; + Self::handle_client_disconnect(&inner, &ctx).await; return; } Ok(Err(_)) => { - Self::handle_client_disconnect(&inner, ctx).await; + Self::handle_client_disconnect(&inner, &ctx).await; return; } Ok(Ok(_)) => {} } - // Contract B2, preserved: bad packets are logged and skipped; - // the connection stays open. - if let Err(e) = Self::process_message(&inner, &body, ctx) { - error!("[OMNIVOX] Error processing message: {e}"); + let opened = open_frame(&recv_key, expected, &body); + let (plaintext, next) = match opened { + Ok(v) => v, + Err(e) => { + // Security violation or junk: the v2 stance is to drop + // replays/reorderings outright; junk bodies are logged + // and skipped (v1's B2 survives only for corrupt frames + // with a valid sequence number). + match e { + FrameError::Replay { seen, expected } => { + warn!( + "[OMNIVOX] Client {} {}:{} replayed/out-of-order frame (seen {seen}, expected {expected})", + ctx.id, ctx.ip, ctx.port + ); + Self::handle_client_disconnect(&inner, &ctx).await; + return; + } + other => { + error!("[OMNIVOX] Error processing message: {other}"); + continue; + } + } + } + }; + expected = next; + + match T::decode(&plaintext) { + Ok(payload) => { + if let Some(f) = inner.received.get() { + fire_data(f, &payload, &ctx); + } + } + Err(e) => error!("[OMNIVOX] Error processing message: {e}"), } } } - fn process_message( - inner: &Inner, - body: &[u8], - ctx: &ConnectionContext, - ) -> Result<(), ProcessError> { - let packet = AeadPacket::decode(body)?; - let plaintext = decrypt(&inner_key(inner), &packet)?; - let payload = T::decode(&plaintext)?; - if let Some(f) = inner.received.get() { - f(&payload, ctx); - } - Ok(()) - } - async fn handle_client_disconnect(inner: &Inner, ctx: &ConnectionContext) { - let removed = { - let mut clients = inner.clients.lock().unwrap(); - clients.remove(&ctx.id) - }; + let removed = inner.clients.lock().unwrap().remove(&ctx.id); if let Some(handle) = removed { - if let Some(mut writer) = handle.writer.lock().await.take() { - let _ = writer.shutdown().await; + if let Some(mut state) = handle.writer.lock().await.take() { + let _ = state.half.shutdown().await; } if let Some(f) = inner.disconnected.get() { - f(ctx); + fire_ctx(f, ctx); } info!("[OMNIVOX] Client {} disconnected", ctx.id); } } - /// Connects to `host:port`, registering callbacks. The read loop runs - /// only when `data` is provided — the C++ gate on `receivedCallback`, - /// including contract B5 (client mode never arms stall timers). - /// - /// The `connection` callback fires exactly once with the outcome; data - /// arriving later reaches `data` on a background task. + /// Connects to `host:port`, completing the v2 handshake before the + /// connection callback fires with `Connected`. pub async fn connect_to_server( &self, host: &str, port: u16, + config: Config, data: Option>, connection: ConnHandler, ) -> Result<(), OmniVoxError> { + let config = self.install_config(config); + let psk = self + .psk() + .ok_or(OmniVoxError::Crypto(CryptoError::EmptyInput))?; + check_defaults(&config, psk)?; + if let Some(d) = data { let _ = self.inner.received.set(d); } let _ = self.inner.server_connected.set(connection); - let ctx = match Self::dial(host, port).await { + let outcome = async { + let stream = Self::dial(host, port).await?; + let (mut reader, mut writer) = stream.into_split(); + let session = timeout( + Duration::from_millis(config.handshake_timeout_ms), + handshake::perform(&mut reader, &mut writer, &psk, false), + ) + .await + .map_err(|_| { + std::io::Error::new(std::io::ErrorKind::TimedOut, "handshake timed out") + })??; + Ok::<_, OmniVoxError>((reader, writer, session)) + } + .await; + + let ctx = match outcome { Err(_) => ConnectionContext::unknown(host, port), - Ok(stream) => { + Ok((reader, writer, session)) => { let ctx = ConnectionContext { connected: ConnectionStatus::Connected, id: self.inner.next_id.fetch_add(1, Ordering::Relaxed), @@ -445,17 +585,22 @@ impl OmniVox { port, }; *self.inner.server_ctx.lock().unwrap() = Some(ctx.clone()); - let (reader, writer) = stream.into_split(); - *self.inner.server_writer.lock().await = Some(writer); + *self.inner.server_writer.lock().await = Some(WriterState { + half: writer, + key: session.send_key, + counter: 1, + }); if self.inner.received.get().is_some() { let inner = Arc::clone(&self.inner); let read_ctx = ctx.clone(); + let cfg = config.clone(); self.inner .tasks .lock() .unwrap() .push(tokio::spawn(async move { - Self::client_read_loop(inner, reader, read_ctx).await; + Self::client_read_loop(inner, reader, session.recv_key, read_ctx, cfg) + .await; })); } ctx @@ -463,14 +608,13 @@ impl OmniVox { }; if let Some(f) = self.inner.server_connected.get() { - f(&ctx); + fire_ctx(f, &ctx); } Ok(()) } async fn dial(host: &str, port: u16) -> Result { - // asio::async_connect tries the resolved list in order and stops at - // the first success. + // asio::async_connect tried the resolved list in order; keep that. let addrs = tokio::net::lookup_host((host, port)).await?; let mut last = Err(std::io::Error::other("no addresses resolved")); for addr in addrs { @@ -485,39 +629,65 @@ impl OmniVox { async fn client_read_loop( inner: Arc>, mut reader: OwnedReadHalf, + recv_key: [u8; KEY_LEN], ctx: ConnectionContext, + config: Config, ) { + let mut expected: u64 = 1; loop { let mut header = [0u8; 4]; - if reader.read_exact(&mut header).await.is_err() { - break; + match timeout( + Duration::from_millis(config.idle_timeout_ms), + reader.read_exact(&mut header), + ) + .await + { + Err(_) | Ok(Err(_)) => break, + Ok(Ok(_)) => {} } let size = u32::from_le_bytes(header); - if size == 0 || size > MAX_MESSAGE_SIZE { - info!( - "[OMNIVOX] Client {} {}:{} has an unreasonable header size!", - ctx.id, ctx.ip, ctx.port - ); - } - let mut body = vec![0u8; size as usize]; - if reader.read_exact(&mut body).await.is_err() { + if size == 0 || size > config.max_message_size { break; } - if let Err(e) = Self::process_message(&inner, &body, &ctx) { - error!("[OMNIVOX] Error processing message: {e}"); + let mut body = vec![0u8; size as usize]; + match timeout( + Duration::from_millis(config.body_stall_ms), + reader.read_exact(&mut body), + ) + .await + { + Err(_) | Ok(Err(_)) => break, + Ok(Ok(_)) => {} + } + match open_frame(&recv_key, expected, &body) { + Ok((plaintext, next)) => { + expected = next; + match T::decode(&plaintext) { + Ok(payload) => { + if let Some(f) = inner.received.get() { + fire_data(f, &payload, &ctx); + } + } + Err(e) => error!("[OMNIVOX] Error processing message: {e}"), + } + } + Err(e) => { + warn!("[OMNIVOX] Bad frame from server: {e}"); + break; + } } } *inner.server_writer.lock().await = None; if let Some(mut sc) = inner.server_ctx.lock().unwrap().take() { sc.connected = ConnectionStatus::NotConnected; if let Some(f) = inner.server_connected.get() { - f(&sc); + fire_ctx(f, &sc); } } } - /// Sends to the connected server. `false` if not connected — the C++ - /// checks `is_open() && status == Connected` and returns, not throws. + /// Sends to the connected server. `false` if not connected or the + /// write failed/timed out. pub async fn send_to_server(&self, data: &T) -> bool { let connected = self .inner @@ -529,46 +699,66 @@ impl OmniVox { if !connected { return false; } - let frame = match Self::sealed_frame(&self.key_bytes(), data) { - Ok(f) => f, - Err(()) => return false, - }; + let config = self.config(); let mut guard = self.inner.server_writer.lock().await; - match guard.as_mut() { - Some(writer) => writer.write_all(&frame).await.is_ok(), - None => false, + let Some(state) = guard.as_mut() else { + return false; + }; + let frame = match seal_frame(&state.key, state.counter, data) { + Some(f) => f, + None => return false, + }; + match timeout( + Duration::from_millis(config.send_timeout_ms), + state.half.write_all(&frame), + ) + .await + { + Ok(Ok(())) => { + state.counter += 1; + true + } + _ => { + *guard = None; + false + } } } - /// Sends to client `id`. `false` if unknown — contract B3 preserved: the - /// write lock is held across the full frame write, as the C++ held the - /// clients mutex across the blocking `asio::write`. + /// Sends to client `id`. `false` if unknown, closed, or the write + /// timed out (slow-client drop). The per-connection lock is held for + /// the full write; no other client is affected. pub async fn send_to_client(&self, data: &T, id: u32) -> bool { let client = self.inner.clients.lock().unwrap().get(&id).cloned(); let Some(client) = client else { return false }; - let Ok(frame) = Self::sealed_frame(&self.key_bytes(), data) else { + let config = self.config(); + let mut guard = client.writer.lock().await; + let Some(state) = guard.as_mut() else { return false; }; - let mut guard = client.writer.lock().await; - match guard.as_mut() { - Some(writer) => writer.write_all(&frame).await.is_ok(), - None => false, + let frame = match seal_frame(&state.key, state.counter, data) { + Some(f) => f, + None => return false, + }; + match timeout( + Duration::from_millis(config.send_timeout_ms), + state.half.write_all(&frame), + ) + .await + { + Ok(Ok(())) => { + state.counter += 1; + true + } + _ => { + *guard = None; + drop(guard); + Self::handle_client_disconnect(&self.inner, &client.context).await; + false + } } } - fn sealed_frame(key: &[u8; KEY_LEN], data: &T) -> Result, ()> { - let mut serialized = Vec::new(); - data.encode(&mut serialized); - let packet = encrypt(key, &serialized); - let mut packet_bytes = Vec::new(); - packet.encode(&mut packet_bytes); - let mut frame = Vec::with_capacity(4 + packet_bytes.len()); - // C++ memcpy'd a size_t into a u32 slot — truncation kept. - frame.extend_from_slice(&(packet_bytes.len() as u32).to_le_bytes()); - frame.extend_from_slice(&packet_bytes); - Ok(frame) - } - /// Stops everything: accept loop, read loops, all sockets. Idempotent. pub async fn stop(&self) { self.inner.stopping.store(true, Ordering::Relaxed); @@ -584,9 +774,14 @@ impl OmniVox { .drain() .map(|(_, c)| c) .collect(); - drop(clients); - let server = self.inner.server_writer.lock().await.take(); - drop(server); + for c in clients { + if let Some(mut state) = c.writer.lock().await.take() { + let _ = state.half.shutdown().await; + } + } + if let Some(mut s) = self.inner.server_writer.lock().await.take() { + let _ = s.half.shutdown().await; + } info!("[OMNIVOX] OmniVox stopped"); } @@ -605,14 +800,60 @@ impl OmniVox { } } -fn inner_key(inner: &Inner) -> [u8; KEY_LEN] { - inner.key.lock().unwrap().unwrap_or([0u8; KEY_LEN]) +/// Seals one frame: `u32 len ‖ AeadPacket` with the sequence-number nonce. +/// `None` only if the payload encoding were to fail (it cannot today). +fn seal_frame(key: &[u8; KEY_LEN], counter: u64, data: &T) -> Option> { + let mut plain = Vec::new(); + data.encode(&mut plain); + let packet = encrypt_with_nonce(key, &frame_nonce(counter), &plain); + let mut bytes = Vec::new(); + packet.encode(&mut bytes); + let mut frame = Vec::with_capacity(4 + bytes.len()); + frame.extend_from_slice(&(bytes.len() as u32).to_le_bytes()); + frame.extend_from_slice(&bytes); + Some(frame) } #[derive(Debug, thiserror::Error)] -enum ProcessError { +enum FrameError { + #[error("replayed or out-of-order frame (seen {seen}, expected {expected})")] + Replay { seen: u64, expected: u64 }, #[error(transparent)] Codec(#[from] CodecError), #[error(transparent)] Crypto(#[from] CryptoError), } + +/// Opens one frame, enforcing the strictly increasing sequence. +fn open_frame( + key: &[u8; KEY_LEN], + expected: u64, + body: &[u8], +) -> Result<(Vec, u64), FrameError> { + let packet = AeadPacket::decode(body)?; + let seen = nonce_counter(&packet.nonce).ok_or(FrameError::Replay { + seen: u64::MAX, + expected, + })?; + if seen != expected { + return Err(FrameError::Replay { seen, expected }); + } + let plaintext = decrypt(key, &packet)?; + Ok((plaintext, expected + 1)) +} + +fn fire_data(cb: &DataHandler, data: &T, ctx: &ConnectionContext) { + if catch_unwind(AssertUnwindSafe(|| cb(data, ctx))).is_err() { + error!("[OMNIVOX] data callback panicked; connection continues"); + } +} + +fn fire_ctx(cb: &ConnHandler, ctx: &ConnectionContext) { + if catch_unwind(AssertUnwindSafe(|| cb(ctx))).is_err() { + error!("[OMNIVOX] connection callback panicked"); + } +} + +fn fire_decision(cb: &DecisionHandler, ctx: &ConnectionContext) -> Decision { + catch_unwind(AssertUnwindSafe(|| cb(ctx))).unwrap_or(Decision::Deny) +} diff --git a/tests/loopback.rs b/tests/loopback.rs index 43412e9..78e6098 100644 --- a/tests/loopback.rs +++ b/tests/loopback.rs @@ -1,10 +1,9 @@ -//! Loopback behavior tests mirroring the contract's runtime clauses -//! (B1 unreasonable header, B2 tampered packet survives, framing). -//! Raw keys via `set_encrypt_key` keep these fast (no scrypt). +//! Loopback behavior tests for the v2 transport: framing discipline, +//! timeouts, junk handling, and the send-path contracts. -use omnivox::{ - AeadPacket, Cereal, ConnectionContext, Decision, MAX_MESSAGE_SIZE, OmniVox, encrypt, -}; +mod support; + +use omnivox::{Config, ConnectionContext, Decision, OmniVox}; use std::{ sync::{ Arc, @@ -12,38 +11,24 @@ use std::{ }, time::Duration, }; -use tokio::{ - io::{AsyncReadExt, AsyncWriteExt}, - net::TcpStream, -}; +use support::{Msg, TEST_PSK, client_session, frame}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; -#[derive(Debug, Default)] -struct Msg(Vec); - -impl Cereal for Msg { - fn encode(&self, out: &mut Vec) { - omnivox::write_byte_vec(out, &self.0); - } - fn decode(bytes: &[u8]) -> Result { - Ok(Self(omnivox::Reader::new(bytes).byte_vec()?)) +fn test_config() -> Config { + Config { + idle_timeout_ms: 1500, + body_stall_ms: 800, + send_timeout_ms: 1000, + handshake_timeout_ms: 5000, + max_message_size: 8192, + max_connections: 16, + allow_insecure_defaults: true, } } -fn frame(key: &[u8; 32], msg: &[u8]) -> Vec { - let mut plain = Vec::new(); - Msg(msg.to_vec()).encode(&mut plain); - let packet = encrypt(key, &plain); - let mut bytes = Vec::new(); - packet.encode(&mut bytes); - let mut out = Vec::new(); - out.extend_from_slice(&(bytes.len() as u32).to_le_bytes()); - out.extend_from_slice(&bytes); - out -} - -async fn start_server(port: u16, rx: Arc) -> Arc> { +async fn start_server(port: u16, rx: Arc, config: Config) -> Arc> { let server = Arc::new(OmniVox::::new()); - server.set_encrypt_key([42u8; 32]); + server.set_encrypt_key(TEST_PSK); let data = Arc::new(move |msg: &Msg, _ctx: &ConnectionContext| { if msg.0.starts_with(b"ok") { rx.fetch_add(1, Ordering::SeqCst); @@ -52,8 +37,7 @@ async fn start_server(port: u16, rx: Arc) -> Arc> { server .start_server( port, - 1, - 2000, + config, data, Arc::new(|_ctx| {}), Arc::new(|_ctx| Decision::Allow), @@ -63,151 +47,146 @@ async fn start_server(port: u16, rx: Arc) -> Arc> { server } -fn free_port(hint: u16) -> u16 { - // bind-and-drop probe; tests run sequentially per port - hint +async fn read_expect_eof(r: &mut tokio::net::tcp::OwnedReadHalf) { + let mut buf = [0u8; 1]; + let n = tokio::time::timeout(Duration::from_secs(4), r.read(&mut buf)) + .await + .expect("connection must close, not hang") + .unwrap(); + assert_eq!(n, 0, "expected EOF"); } #[tokio::test] -async fn tampered_message_is_dropped_but_connection_survives() { - let port = free_port(18101); +async fn tampered_body_with_valid_counter_is_skipped_connection_survives() { + let port = 18201; let rx = Arc::new(AtomicU32::new(0)); - let server = start_server(port, Arc::clone(&rx)).await; + let server = start_server(port, Arc::clone(&rx), test_config()).await; - let client = OmniVox::::new(); - client.set_encrypt_key([42u8; 32]); - let got = Arc::new(AtomicU32::new(0)); - let g = Arc::clone(&got); - client - .connect_to_server( - "127.0.0.1", - port, - Some(Arc::new(move |_d, _c| { - g.fetch_add(1, Ordering::SeqCst); - })), - Arc::new(|_c| {}), - ) - .await - .unwrap(); + let (r, mut w, s) = client_session(port, &TEST_PSK).await; - // Valid through the API. - assert!(client.send_to_server(&Msg(b"ok1".to_vec())).await); - // Tampered raw frame: flip a ciphertext byte. - let mut bad = frame(&[42u8; 32], b"ok2"); + // Counter 1, tampered ciphertext: auth fails, frame dropped, alive. + let mut bad = frame(&s.send_key, 1, b"ok-junk"); let last = bad.len() - 1; bad[last] ^= 0x01; - let mut raw = TcpStream::connect(("127.0.0.1", port)).await.unwrap(); - raw.write_all(&bad).await.unwrap(); - // Valid again on the tampered connection's server slot (new conn). - assert!(client.send_to_server(&Msg(b"ok3".to_vec())).await); + w.write_all(&bad).await.unwrap(); + + // Counter 1 valid (the tampered frame consumed nothing). + w.write_all(&frame(&s.send_key, 1, b"ok1")).await.unwrap(); + w.write_all(&frame(&s.send_key, 2, b"ok2")).await.unwrap(); tokio::time::sleep(Duration::from_millis(300)).await; - assert_eq!( - rx.load(Ordering::SeqCst), - 2, - "tampered frame must not decode" - ); - // Contract B2: the tampering connection is NOT dropped — server keeps - // its header wait armed (rx-timeout is 2 s, so a 500 ms read must time - // out without EOF). - let r = tokio::time::timeout(Duration::from_millis(500), raw.read(&mut [0u8; 1])).await; - assert!(r.is_err(), "tampering connection must stay open (B2)"); - assert_eq!( - got.load(Ordering::SeqCst), - 0, - "server does not echo in this test" - ); + assert_eq!(rx.load(Ordering::SeqCst), 2); + // still alive: another frame gets through + w.write_all(&frame(&s.send_key, 3, b"ok3")).await.unwrap(); + tokio::time::sleep(Duration::from_millis(300)).await; + assert_eq!(rx.load(Ordering::SeqCst), 3); server.stop().await; - client.stop().await; + drop(r); + drop(w); } #[tokio::test] -async fn zero_size_header_disconnects_like_cpp_b1() { - let port = free_port(18102); +async fn zero_size_header_disconnects_immediately() { + // v2 retires the v1 no-return quirk: bad header closes, no follow-up. + let port = 18202; let rx = Arc::new(AtomicU32::new(0)); - let server = start_server(port, Arc::clone(&rx)).await; + let server = start_server(port, Arc::clone(&rx), test_config()).await; - let mut raw = TcpStream::connect(("127.0.0.1", port)).await.unwrap(); - raw.write_all(&0u32.to_le_bytes()).await.unwrap(); - // C++ B1: server logs, disconnects, then the 0-byte read "succeeds" and - // the empty packet fails to decode (logged). The socket must close. - let mut buf = [0u8; 1]; - let n = tokio::time::timeout(Duration::from_secs(2), raw.read(&mut buf)) - .await - .expect("server must close, not hang") - .unwrap(); - assert_eq!(n, 0, "EOF after unreasonable header"); + let (mut r, mut w, _s) = client_session(port, &TEST_PSK).await; + w.write_all(&0u32.to_le_bytes()).await.unwrap(); + read_expect_eof(&mut r).await; server.stop().await; } #[tokio::test] async fn oversized_header_disconnects() { - let port = free_port(18103); + let port = 18203; let rx = Arc::new(AtomicU32::new(0)); - let server = start_server(port, Arc::clone(&rx)).await; + let server = start_server(port, Arc::clone(&rx), test_config()).await; - let mut raw = TcpStream::connect(("127.0.0.1", port)).await.unwrap(); - raw.write_all(&(MAX_MESSAGE_SIZE + 1).to_le_bytes()) - .await - .unwrap(); - let mut buf = [0u8; 1]; - let n = tokio::time::timeout(Duration::from_secs(2), raw.read(&mut buf)) - .await - .expect("server must close, not hang") - .unwrap(); - assert_eq!(n, 0); + let (mut r, mut w, _s) = client_session(port, &TEST_PSK).await; + w.write_all(&8193u32.to_le_bytes()).await.unwrap(); + read_expect_eof(&mut r).await; server.stop().await; } #[tokio::test] -async fn header_stall_times_out_server_side() { - let port = free_port(18104); +async fn header_stall_times_out() { + let port = 18204; + let rx = Arc::new(AtomicU32::new(0)); + let server = start_server(port, Arc::clone(&rx), test_config()).await; + + let (mut r, mut w, _s) = client_session(port, &TEST_PSK).await; + w.write_all(&[0x10, 0x00]).await.unwrap(); // partial header, then silence + read_expect_eof(&mut r).await; + server.stop().await; +} + +#[tokio::test] +async fn send_to_unknown_client_is_false() { + let server = OmniVox::::new(); + server.set_encrypt_key(TEST_PSK); + assert!(!server.send_to_client(&Msg(b"x".to_vec()), 999).await); + assert!(!server.send_to_server(&Msg(b"x".to_vec())).await); +} + +#[tokio::test] +async fn valid_handshake_then_api_roundtrip() { + let port = 18205; let rx = Arc::new(AtomicU32::new(0)); - // start_server timeout_ms = 300 for this one let server = Arc::new(OmniVox::::new()); - server.set_encrypt_key([42u8; 32]); + server.set_encrypt_key(TEST_PSK); + let data = Arc::new({ + let rx = Arc::clone(&rx); + move |msg: &Msg, ctx: &ConnectionContext| { + if msg.0.starts_with(b"ping") { + rx.fetch_add(1, Ordering::SeqCst); + let server = Arc::clone(SERVER_FOR_ECHO.get().unwrap()); + let id = ctx.id; + let reply = Msg(b"pong".to_vec()); + tokio::spawn(async move { + server.send_to_client(&reply, id).await; + }); + } + } + }); + let _ = SERVER_FOR_ECHO.set(server.clone()); server .start_server( port, - 1, - 300, - Arc::new(move |_: &Msg, _: &ConnectionContext| { - rx.fetch_add(1, Ordering::SeqCst); - }), + test_config(), + data, Arc::new(|_c| {}), Arc::new(|_c| Decision::Allow), ) .await .unwrap(); - let mut raw = TcpStream::connect(("127.0.0.1", port)).await.unwrap(); - raw.write_all(&[0x10, 0x00]).await.unwrap(); // partial header, then silence - let mut buf = [0u8; 1]; - let n = tokio::time::timeout(Duration::from_secs(3), raw.read(&mut buf)) + let client = OmniVox::::new(); + client.set_encrypt_key(TEST_PSK); + let got = Arc::new(AtomicU32::new(0)); + let g = Arc::clone(&got); + client + .connect_to_server( + "127.0.0.1", + port, + test_config(), + Some(Arc::new(move |msg: &Msg, _c| { + if msg.0 == b"pong" { + g.fetch_add(1, Ordering::SeqCst); + } + })), + Arc::new(|_c| {}), + ) .await - .expect("stall timeout must fire") .unwrap(); - assert_eq!(n, 0); + + assert!(client.send_to_server(&Msg(b"ping".to_vec())).await); + tokio::time::sleep(Duration::from_millis(400)).await; + assert_eq!(rx.load(Ordering::SeqCst), 1); + assert_eq!(got.load(Ordering::SeqCst), 1, "encrypted echo back"); + client.stop().await; server.stop().await; } -#[tokio::test] -async fn send_to_unknown_client_is_false_like_cpp() { - let server = OmniVox::::new(); - server.set_encrypt_key([0u8; 32]); - assert!(!server.send_to_client(&Msg(b"x".to_vec()), 999).await); - assert!(!server.send_to_server(&Msg(b"x".to_vec())).await); -} - -#[test] -fn aead_packet_roundtrip_matches_wire_shape() { - let pkt = AeadPacket { - nonce: [9; 12], - tag: [8; 16], - payload: vec![7; 3], - }; - let mut b = Vec::new(); - pkt.encode(&mut b); - assert_eq!(b.len(), 12 + 16 + 8 + 3); - assert_eq!(AeadPacket::decode(&b).unwrap(), pkt); -} +static SERVER_FOR_ECHO: std::sync::OnceLock>> = std::sync::OnceLock::new(); diff --git a/tests/protocol_v2.rs b/tests/protocol_v2.rs new file mode 100644 index 0000000..d9952cb --- /dev/null +++ b/tests/protocol_v2.rs @@ -0,0 +1,259 @@ +//! Attack regressions: every v1 vulnerability class from the red-team +//! review must now fail closed, provably, with zero callbacks fired. + +mod support; + +use omnivox::{Config, ConnectionContext, ConnectionStatus, Decision, OmniVox, handshake::perform}; +use std::{ + sync::{ + Arc, + atomic::{AtomicU8, AtomicU32, Ordering}, + }, + time::Duration, +}; +use support::{Msg, TEST_PSK, client_session, frame}; +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::TcpStream, +}; + +fn test_config() -> Config { + Config { + idle_timeout_ms: 1500, + body_stall_ms: 800, + send_timeout_ms: 1000, + handshake_timeout_ms: 5000, + max_message_size: 8192, + max_connections: 16, + allow_insecure_defaults: true, + } +} + +async fn counting_server_with(port: u16, rx: Arc, config: Config) -> Arc> { + let server = Arc::new(OmniVox::::new()); + server.set_encrypt_key(TEST_PSK); + let data = Arc::new(move |msg: &Msg, _ctx: &ConnectionContext| { + if msg.0.starts_with(b"ok") { + rx.fetch_add(1, Ordering::SeqCst); + } + }); + server + .start_server( + port, + config, + data, + Arc::new(|_ctx| {}), + Arc::new(|_ctx| Decision::Allow), + ) + .await + .unwrap(); + server +} + +async fn counting_server(port: u16, rx: Arc) -> Arc> { + counting_server_with(port, rx, test_config()).await +} + +async fn expect_eof(r: &mut tokio::net::tcp::OwnedReadHalf) { + let mut buf = [0u8; 1]; + let n = tokio::time::timeout(Duration::from_secs(4), r.read(&mut buf)) + .await + .expect("connection must close") + .unwrap(); + assert_eq!(n, 0); +} + +#[tokio::test] +async fn replayed_frame_is_dropped_and_the_connection_dies() { + let port = 18301; + let rx = Arc::new(AtomicU32::new(0)); + let server = counting_server(port, Arc::clone(&rx)).await; + + let (mut r, mut w, s) = client_session(port, &TEST_PSK).await; + let good = frame(&s.send_key, 1, b"ok-transfer-1000"); + w.write_all(&good).await.unwrap(); + tokio::time::sleep(Duration::from_millis(200)).await; + assert_eq!(rx.load(Ordering::SeqCst), 1, "first delivery"); + + // Reinject the verbatim captured frame. + w.write_all(&good).await.unwrap(); + expect_eof(&mut r).await; // server must drop the connection outright + tokio::time::sleep(Duration::from_millis(200)).await; + assert_eq!( + rx.load(Ordering::SeqCst), + 1, + "replay must never reach the app" + ); + server.stop().await; +} + +#[tokio::test] +async fn frames_from_another_session_are_undecryptable() { + // Frame portability: keys are session-bound by the handshake nonces, + // so a frame valid on connection A is junk on connection B. + let port = 18302; + let rx = Arc::new(AtomicU32::new(0)); + let server = counting_server(port, Arc::clone(&rx)).await; + + let (_r1, _w1, s1) = client_session(port, &TEST_PSK).await; + let (_r2, mut w2, s2) = client_session(port, &TEST_PSK).await; + + let portable = frame(&s1.send_key, 1, b"ok-portable"); + w2.write_all(&portable).await.unwrap(); + // Auth failure with a matching counter would continue (junk policy), but + // the counter check passes while decryption fails → logged, skipped; + // the connection survives and the payload never fires. + tokio::time::sleep(Duration::from_millis(200)).await; + assert_eq!( + rx.load(Ordering::SeqCst), + 0, + "cross-session frame must not decrypt" + ); + // connection B is still alive with its own keys: + w2.write_all(&frame(&s2.send_key, 1, b"ok-native")) + .await + .unwrap(); + tokio::time::sleep(Duration::from_millis(200)).await; + assert_eq!( + rx.load(Ordering::SeqCst), + 1, + "native frame on B must arrive" + ); + server.stop().await; +} + +#[tokio::test] +async fn wrong_version_hello_is_closed() { + let port = 18303; + let rx = Arc::new(AtomicU32::new(0)); + let server = counting_server(port, rx).await; + + let raw = TcpStream::connect(("127.0.0.1", port)).await.unwrap(); + let (mut r, mut w) = raw.into_split(); + let mut hello = [0u8; 69]; + hello[..4].copy_from_slice(b"XXXX"); + hello[4] = 99; + w.write_all(&hello).await.unwrap(); + expect_eof(&mut r).await; + server.stop().await; +} + +#[tokio::test] +async fn v1_traffic_is_rejected_at_the_handshake() { + // A v1 OmniVox client speaks u32-length-prefixed AeadPackets. The v2 + // server must refuse before any allocation path. + let port = 18304; + let rx = Arc::new(AtomicU32::new(0)); + let server = counting_server_with( + port, + rx, + Config { + handshake_timeout_ms: 700, + ..test_config() + }, + ) + .await; + + let raw = TcpStream::connect(("127.0.0.1", port)).await.unwrap(); + let (mut r, mut w) = raw.into_split(); + w.write_all(&36u32.to_le_bytes()).await.unwrap(); // v1 header + w.write_all(&[0u8; 36]).await.unwrap(); + expect_eof(&mut r).await; + server.stop().await; +} + +#[tokio::test] +async fn panicking_callback_does_not_kill_the_connection() { + let port = 18305; + let rx = Arc::new(AtomicU32::new(0)); + let server = Arc::new(OmniVox::::new()); + server.set_encrypt_key(TEST_PSK); + let ok_rx = Arc::clone(&rx); + let data = Arc::new(move |msg: &Msg, _ctx: &ConnectionContext| { + if msg.0 == b"boom" { + let _v: [u8; 8] = msg.0[..8].try_into().unwrap(); // explodes (len 4) + } + if msg.0.starts_with(b"ok") { + ok_rx.fetch_add(1, Ordering::SeqCst); + } + }); + server + .start_server( + port, + test_config(), + data, + Arc::new(|_c| {}), + Arc::new(|_c| Decision::Allow), + ) + .await + .unwrap(); + + let (mut r, mut w, s) = client_session(port, &TEST_PSK).await; + w.write_all(&frame(&s.send_key, 1, b"boom")).await.unwrap(); + w.write_all(&frame(&s.send_key, 2, b"ok-after-panic")) + .await + .unwrap(); + tokio::time::sleep(Duration::from_millis(300)).await; + assert_eq!( + rx.load(Ordering::SeqCst), + 1, + "survives a panicking callback" + ); + let mut buf = [0u8; 1]; + assert!( + tokio::time::timeout(Duration::from_millis(300), r.read(&mut buf)) + .await + .is_err(), + "connection must still be open" + ); + server.stop().await; +} + +#[tokio::test] +async fn malicious_server_silence_times_out_the_client() { + // v2 retires B5: the client side now has the same idle/stall discipline. + let port = 18306; + let listener = tokio::net::TcpListener::bind(("127.0.0.1", port)) + .await + .unwrap(); + + let accepted = tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + let (mut r, mut w) = stream.into_split(); + // Complete the handshake as the server, then go silent forever. + perform(&mut r, &mut w, &TEST_PSK, true).await.unwrap(); + tokio::time::sleep(Duration::from_secs(60)).await; + }); + + let client = OmniVox::::new(); + client.set_encrypt_key(TEST_PSK); + let dropped = Arc::new(AtomicU8::new(0)); + let d = Arc::clone(&dropped); + let cfg = Config { + idle_timeout_ms: 400, + ..test_config() + }; + client + .connect_to_server( + "127.0.0.1", + port, + cfg, + Some(Arc::new(|_m: &Msg, _c: &ConnectionContext| {})), + Arc::new(move |ctx: &ConnectionContext| { + if ctx.connected == ConnectionStatus::NotConnected { + d.store(1, Ordering::SeqCst); + } + }), + ) + .await + .unwrap(); + + tokio::time::sleep(Duration::from_millis(1500)).await; + assert_eq!( + dropped.load(Ordering::SeqCst), + 1, + "client must time out a silent server and report NotConnected" + ); + accepted.abort(); + client.stop().await; +} diff --git a/tests/support/mod.rs b/tests/support/mod.rs new file mode 100644 index 0000000..806e7ad --- /dev/null +++ b/tests/support/mod.rs @@ -0,0 +1,42 @@ +//! Shared helpers for the integration tests: v2 client sessions against a +//! live server, and raw frame construction. + +use omnivox::{Cereal, CodecError, Session, encrypt_with_nonce, frame_nonce, handshake::perform}; +use tokio::net::TcpStream; +use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf}; + +#[derive(Debug, Default)] +pub struct Msg(pub Vec); + +impl Cereal for Msg { + fn encode(&self, out: &mut Vec) { + omnivox::write_byte_vec(out, &self.0); + } + fn decode(bytes: &[u8]) -> Result { + Ok(Self(omnivox::Reader::new(bytes).byte_vec()?)) + } +} + +pub const TEST_PSK: [u8; 32] = [42u8; 32]; + +/// Builds one framed message with the given sequence number. +pub fn frame(key: &[u8; 32], counter: u64, msg: &[u8]) -> Vec { + let mut plain = Vec::new(); + Msg(msg.to_vec()).encode(&mut plain); + let pkt = encrypt_with_nonce(key, &frame_nonce(counter), &plain); + let mut bytes = Vec::new(); + pkt.encode(&mut bytes); + let mut out = Vec::new(); + out.extend_from_slice(&(bytes.len() as u32).to_le_bytes()); + out.extend_from_slice(&bytes); + out +} + +/// Connects and completes the v2 handshake as the client role, returning +/// the raw halves plus the session keys for manual framing. +pub async fn client_session(port: u16, psk: &[u8; 32]) -> (OwnedReadHalf, OwnedWriteHalf, Session) { + let stream = TcpStream::connect(("127.0.0.1", port)).await.unwrap(); + let (mut r, mut w) = stream.into_split(); + let session = perform(&mut r, &mut w, psk, false).await.unwrap(); + (r, w, session) +}