0.15.0: simple builder API, v2.1 framing, stream lane, benches

- omnivox::server::<T>()/client::<T>() builders; plain closures, enum in,
  match in the callback — the original C++ UX with the ceremony removed
- serde + bincode replace the hand-rolled Cereal codec (vectors.rs still
  governs the crypto leaves via the C++ golden oracles)
- v2.1 frames: u32 len | type | ct | tag; nonce derived from the sequence
  counter (no wire nonce), type byte authenticated as AAD; any frame-level
  failure is fatal, authenticated-but-undecodable payloads are skipped
- stream lane: send_stream/on_stream, 1 MiB frames, shared sequence with
  messages (exact sender interleaving preserved); see dump-agent/viewer
- per-connection bounded queues + writer tasks; reply/reply_stream are
  non-blocking try_send lanes for callbacks
- benches: seal/open criterion group + transport harness (spray msg/s,
  stream GiB/s, raw loopback baseline)
- profile.dev.package scrypt/pbkdf2 opt-level 3: golden-vector scrypt at
  N=2^17 was unusable in debug builds
- PROTOCOL-V2.md and README rewritten for v2.1

tests/loopback.rs

@@ -1,9 +1,9 @@
-//! Loopback behavior tests for the v2 transport: framing discipline,
-//! timeouts, junk handling, and the send-path contracts.
+//! Behavior tests for the v2.1 transport via the public builder API:
+//! framing discipline, timeouts, the echo roundtrip, and send-path returns.
 
 mod support;
 
-use omnivox::{Config, ConnectionContext, Decision, OmniVox};
+use omnivox::{Config, Decision, Server};
 use std::{
     sync::{
         Arc,
@@ -11,7 +11,7 @@ use std::{
     },
     time::Duration,
 };
-use support::{Msg, TEST_PSK, client_session, frame};
+use support::{Msg, TEST_PSK, client_session};
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 
 fn test_config() -> Config {
@@ -21,33 +21,28 @@ fn test_config() -> Config {
         send_timeout_ms: 1000,
         handshake_timeout_ms: 5000,
         max_message_size: 8192,
+        max_stream_frame: 8192,
         max_connections: 16,
         allow_insecure_defaults: true,
     }
 }
 
-async fn start_server(port: u16, rx: Arc<AtomicU32>, config: Config) -> Arc<OmniVox<Msg>> {
-    let server = Arc::new(OmniVox::<Msg>::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),
-        )
+async fn counting_server(port: u16, rx: Arc<AtomicU32>) -> Server<Msg> {
+    omnivox::server::<Msg>()
+        .key(TEST_PSK)
+        .config(test_config())
+        .on_data(move |msg: &Msg, _ctx| {
+            if msg.0.starts_with(b"ok") {
+                rx.fetch_add(1, Ordering::SeqCst);
+            }
+        })
+        .on_connect(|_ctx| Decision::Allow)
+        .serve(port)
         .await
-        .unwrap();
-    server
+        .unwrap()
 }
 
-async fn read_expect_eof(r: &mut tokio::net::tcp::OwnedReadHalf) {
+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
@@ -57,136 +52,102 @@ async fn read_expect_eof(r: &mut tokio::net::tcp::OwnedReadHalf) {
 }
 
 #[tokio::test]
-async fn tampered_body_with_valid_counter_is_skipped_connection_survives() {
-    let port = 18201;
+async fn tampered_frame_disconnects_v21_policy() {
+    // v2.1: authentication failure is fatal — no skip-and-continue.
+    let port = 18401;
     let rx = Arc::new(AtomicU32::new(0));
-    let server = start_server(port, Arc::clone(&rx), test_config()).await;
+    let server = counting_server(port, Arc::clone(&rx)).await;
 
-    let (r, mut w, s) = client_session(port, &TEST_PSK).await;
-
-    // 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;
-    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);
-    // 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);
+    let (mut r, mut w, s) = client_session(port, &TEST_PSK).await;
+    let mut good = support::msg_frame(&s.send_key, 1, b"ok1");
+    let last = good.len() - 5; // flip a ciphertext byte
+    good[last] ^= 0x01;
+    w.write_all(&good).await.unwrap();
+    expect_eof(&mut r).await;
+    assert_eq!(rx.load(Ordering::SeqCst), 0);
     server.stop().await;
-    drop(r);
-    drop(w);
 }
 
 #[tokio::test]
-async fn zero_size_header_disconnects_immediately() {
-    // v2 retires the v1 no-return quirk: bad header closes, no follow-up.
-    let port = 18202;
+async fn zero_size_header_disconnects() {
+    let port = 18402;
     let rx = Arc::new(AtomicU32::new(0));
-    let server = start_server(port, Arc::clone(&rx), test_config()).await;
-
+    let server = counting_server(port, rx).await;
     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;
+    expect_eof(&mut r).await;
     server.stop().await;
 }
 
 #[tokio::test]
 async fn oversized_header_disconnects() {
-    let port = 18203;
+    let port = 18403;
     let rx = Arc::new(AtomicU32::new(0));
-    let server = start_server(port, Arc::clone(&rx), test_config()).await;
-
+    let server = counting_server(port, rx).await;
     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;
+    w.write_all(&9000u32.to_le_bytes()).await.unwrap();
+    expect_eof(&mut r).await;
     server.stop().await;
 }
 
 #[tokio::test]
 async fn header_stall_times_out() {
-    let port = 18204;
+    let port = 18404;
     let rx = Arc::new(AtomicU32::new(0));
-    let server = start_server(port, Arc::clone(&rx), test_config()).await;
-
+    let server = counting_server(port, rx).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;
+    expect_eof(&mut r).await;
     server.stop().await;
 }
 
 #[tokio::test]
 async fn send_to_unknown_client_is_false() {
-    let server = OmniVox::<Msg>::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);
+    let rx = Arc::new(AtomicU32::new(0));
+    let server = omnivox::server::<Msg>()
+        .key(TEST_PSK)
+        .config(test_config())
+        .on_data(move |_: &Msg, _| {})
+        .serve(18405)
+        .await
+        .unwrap();
+    assert!(!server.send_to_client(999, &Msg(b"x".to_vec())).await);
+    assert_eq!(server.broadcast(&Msg(b"x".to_vec())).await, 0);
+    server.stop().await;
+    let _ = rx;
 }
 
 #[tokio::test]
-async fn valid_handshake_then_api_roundtrip() {
-    let port = 18205;
-    let rx = Arc::new(AtomicU32::new(0));
-    let server = Arc::new(OmniVox::<Msg>::new());
-    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,
-            test_config(),
-            data,
-            Arc::new(|_c| {}),
-            Arc::new(|_c| Decision::Allow),
-        )
+async fn echo_roundtrip_through_the_public_api() {
+    let port = 18406;
+    let server = omnivox::server::<Msg>()
+        .key(TEST_PSK)
+        .config(test_config())
+        .on_data(|msg, ctx| {
+            // echo via the synchronous callback lane
+            let _ = ctx.reply(msg);
+        })
+        .serve(port)
         .await
         .unwrap();
 
-    let client = OmniVox::<Msg>::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| {}),
-        )
+    let client = omnivox::client::<Msg>()
+        .key(TEST_PSK)
+        .config(test_config())
+        .on_data(move |msg, _ctx| {
+            if msg.0 == b"ping" {
+                g.fetch_add(1, Ordering::SeqCst);
+            }
+        })
+        .connect("127.0.0.1", port)
         .await
         .unwrap();
 
-    assert!(client.send_to_server(&Msg(b"ping".to_vec())).await);
+    assert!(client.send(&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");
+    assert_eq!(got.load(Ordering::SeqCst), 1, "encrypted echo returned");
     client.stop().await;
     server.stop().await;
 }
-
-static SERVER_FOR_ECHO: std::sync::OnceLock<Arc<OmniVox<Msg>>> = std::sync::OnceLock::new();

tests/protocol_v2.rs

@@ -1,9 +1,9 @@
-//! Attack regressions: every v1 vulnerability class from the red-team
-//! review must now fail closed, provably, with zero callbacks fired.
+//! Attack regressions for protocol v2.1: every red-team finding class must
+//! fail closed, provably, with zero callbacks fired on the attack input.
 
 mod support;
 
-use omnivox::{Config, ConnectionContext, ConnectionStatus, Decision, OmniVox, handshake::perform};
+use omnivox::{Config, ConnectionStatus, Decision};
 use std::{
     sync::{
         Arc,
@@ -11,7 +11,7 @@ use std::{
     },
     time::Duration,
 };
-use support::{Msg, TEST_PSK, client_session, frame};
+use support::{Msg, TEST_PSK, client_session, msg_frame, raw_frame, stream_frame};
 use tokio::{
     io::{AsyncReadExt, AsyncWriteExt},
     net::TcpStream,
@@ -24,34 +24,24 @@ fn test_config() -> Config {
         send_timeout_ms: 1000,
         handshake_timeout_ms: 5000,
         max_message_size: 8192,
+        max_stream_frame: 8192,
         max_connections: 16,
         allow_insecure_defaults: true,
     }
 }
 
-async fn counting_server_with(port: u16, rx: Arc<AtomicU32>, config: Config) -> Arc<OmniVox<Msg>> {
-    let server = Arc::new(OmniVox::<Msg>::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),
-        )
+async fn counting_server(port: u16, rx: Arc<AtomicU32>) -> omnivox::Server<Msg> {
+    omnivox::server::<Msg>()
+        .key(TEST_PSK)
+        .config(test_config())
+        .on_data(move |msg: &Msg, _ctx| {
+            if msg.0.starts_with(b"ok") {
+                rx.fetch_add(1, Ordering::SeqCst);
+            }
+        })
+        .serve(port)
         .await
-        .unwrap();
-    server
-}
-
-async fn counting_server(port: u16, rx: Arc<AtomicU32>) -> Arc<OmniVox<Msg>> {
-    counting_server_with(port, rx, test_config()).await
+        .unwrap()
 }
 
 async fn expect_eof(r: &mut tokio::net::tcp::OwnedReadHalf) {
@@ -64,21 +54,21 @@ async fn expect_eof(r: &mut tokio::net::tcp::OwnedReadHalf) {
 }
 
 #[tokio::test]
-async fn replayed_frame_is_dropped_and_the_connection_dies() {
-    let port = 18301;
+async fn replayed_frame_kills_the_connection() {
+    let port = 18501;
     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");
+    let good = msg_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.
+    // Reinject the verbatim captured frame: nonce is derived from our own
+    // counter (now 2), so authentication fails outright.
     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;
+    expect_eof(&mut r).await;
     assert_eq!(
         rx.load(Ordering::SeqCst),
         1,
@@ -88,43 +78,47 @@ async fn replayed_frame_is_dropped_and_the_connection_dies() {
 }
 
 #[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;
+async fn frames_from_another_session_are_rejected() {
+    let port = 18502;
     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 (mut r2, mut w2, _s2) = client_session(port, &TEST_PSK).await;
 
-    let portable = frame(&s1.send_key, 1, b"ok-portable");
+    // A perfectly valid frame from session A, injected into session B.
+    let portable = msg_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;
+    expect_eof(&mut r2).await; // v2.1: auth failure is fatal
     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 frame_type_tamper_is_rejected() {
+    // Message -> stream downgrade attempt: type byte is AAD, so this is an
+    // authentication failure, not a lane switch.
+    let port = 18503;
+    let rx = Arc::new(AtomicU32::new(0));
+    let server = counting_server(port, rx).await;
+
+    let (mut r, mut w, s) = client_session(port, &TEST_PSK).await;
+    let mut frame = msg_frame(&s.send_key, 1, b"ok-msg");
+    let type_idx = frame.len() - 17; // [len|type|ct|tag] -> type at offset 4
+    frame[4] = 1;
+    let _ = type_idx;
+    w.write_all(&frame).await.unwrap();
+    expect_eof(&mut r).await;
     server.stop().await;
 }
 
 #[tokio::test]
 async fn wrong_version_hello_is_closed() {
-    let port = 18303;
+    let port = 18504;
     let rx = Arc::new(AtomicU32::new(0));
     let server = counting_server(port, rx).await;
 
@@ -140,19 +134,20 @@ async fn wrong_version_hello_is_closed() {
 
 #[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 port = 18505;
     let rx = Arc::new(AtomicU32::new(0));
-    let server = counting_server_with(
-        port,
-        rx,
-        Config {
+    let server = omnivox::server::<Msg>()
+        .key(TEST_PSK)
+        .config(Config {
             handshake_timeout_ms: 700,
             ..test_config()
-        },
-    )
-    .await;
+        })
+        .on_data(move |_: &Msg, _| {
+            rx.fetch_add(1, Ordering::SeqCst);
+        })
+        .serve(port)
+        .await
+        .unwrap();
 
     let raw = TcpStream::connect(("127.0.0.1", port)).await.unwrap();
     let (mut r, mut w) = raw.into_split();
@@ -163,41 +158,43 @@ async fn v1_traffic_is_rejected_at_the_handshake() {
 }
 
 #[tokio::test]
-async fn panicking_callback_does_not_kill_the_connection() {
-    let port = 18305;
+async fn panicking_callbacks_never_kill_the_connection() {
+    let port = 18506;
     let rx = Arc::new(AtomicU32::new(0));
-    let server = Arc::new(OmniVox::<Msg>::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),
-        )
+    let server = omnivox::server::<Msg>()
+        .key(TEST_PSK)
+        .config(test_config())
+        .on_data(move |msg: &Msg, _ctx| {
+            if msg.0 == b"boom" {
+                panic!("handler exploded on attacker input");
+            }
+            if msg.0.starts_with(b"ok") {
+                ok_rx.fetch_add(1, Ordering::SeqCst);
+            }
+        })
+        .on_stream(|_b, _ctx| {
+            panic!("stream handler exploded");
+        })
+        .serve(port)
         .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"))
+    w.write_all(&msg_frame(&s.send_key, 1, b"boom"))
+        .await
+        .unwrap();
+    w.write_all(&stream_frame(&s.send_key, 2, b"stream-boom"))
+        .await
+        .unwrap();
+    w.write_all(&msg_frame(&s.send_key, 3, b"ok-after-panics"))
         .await
         .unwrap();
     tokio::time::sleep(Duration::from_millis(300)).await;
     assert_eq!(
         rx.load(Ordering::SeqCst),
         1,
-        "survives a panicking callback"
+        "connection survives two panics"
     );
     let mut buf = [0u8; 1];
     assert!(
@@ -211,8 +208,7 @@ async fn panicking_callback_does_not_kill_the_connection() {
 
 #[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 port = 18507;
     let listener = tokio::net::TcpListener::bind(("127.0.0.1", port))
         .await
         .unwrap();
@@ -221,30 +217,27 @@ async fn malicious_server_silence_times_out_the_client() {
         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();
+        omnivox::handshake::perform(&mut r, &mut w, &TEST_PSK, true)
+            .await
+            .unwrap();
         tokio::time::sleep(Duration::from_secs(60)).await;
     });
 
-    let client = OmniVox::<Msg>::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);
-                }
-            }),
-        )
+    let client = omnivox::client::<Msg>()
+        .key(TEST_PSK)
+        .config(Config {
+            idle_timeout_ms: 400,
+            ..test_config()
+        })
+        .on_data(|_m: &Msg, _c| {})
+        .on_state(move |ctx| {
+            if ctx.status() == ConnectionStatus::NotConnected {
+                d.store(1, Ordering::SeqCst);
+            }
+        })
+        .connect("127.0.0.1", port)
         .await
         .unwrap();
 
@@ -257,3 +250,124 @@ async fn malicious_server_silence_times_out_the_client() {
     accepted.abort();
     client.stop().await;
 }
+
+#[tokio::test]
+async fn messages_and_streams_interleave_in_order() {
+    // One sequence across both lanes: strict per-direction ordering of the
+    // exact interleaving the sender produced.
+    let port = 18508;
+    let tape = Arc::new(std::sync::Mutex::new(Vec::<String>::new()));
+    let t = Arc::clone(&tape);
+    let t2 = Arc::clone(&tape);
+    let server = omnivox::server::<Msg>()
+        .key(TEST_PSK)
+        .config(test_config())
+        .on_data(move |msg: &Msg, _ctx| {
+            t.lock()
+                .unwrap()
+                .push(format!("msg:{}", String::from_utf8_lossy(&msg.0)));
+        })
+        .on_stream(move |bytes, _ctx| {
+            t2.lock()
+                .unwrap()
+                .push(format!("stream:{}", String::from_utf8_lossy(bytes)));
+        })
+        .serve(port)
+        .await
+        .unwrap();
+
+    let (_r, mut w, s) = client_session(port, &TEST_PSK).await;
+    w.write_all(&msg_frame(&s.send_key, 1, b"alpha"))
+        .await
+        .unwrap();
+    w.write_all(&stream_frame(&s.send_key, 2, b"S1"))
+        .await
+        .unwrap();
+    w.write_all(&msg_frame(&s.send_key, 3, b"beta"))
+        .await
+        .unwrap();
+    w.write_all(&stream_frame(&s.send_key, 4, b"S2"))
+        .await
+        .unwrap();
+    tokio::time::sleep(Duration::from_millis(400)).await;
+    assert_eq!(
+        *tape.lock().unwrap(),
+        vec![
+            "msg:alpha".to_string(),
+            "stream:S1".to_string(),
+            "msg:beta".to_string(),
+            "stream:S2".to_string(),
+        ]
+    );
+    server.stop().await;
+}
+
+#[tokio::test]
+async fn lying_decoder_length_is_skipped_connection_survives() {
+    // A frame that authenticates but whose payload claims a huge bincode
+    // length: decode fails, frame is skipped (authenticated junk policy),
+    // connection continues.
+    let port = 18509;
+    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 lying = {
+        // Msg(Vec<u8>) bincode: u64 LE length then bytes. Claim 2^40.
+        let mut plain = Vec::new();
+        plain.extend_from_slice(&(1u64 << 40).to_le_bytes());
+        plain.extend_from_slice(b"ha");
+        raw_frame(&s.send_key, 1, omnivox::FrameType::Message, &plain)
+    };
+    w.write_all(&lying).await.unwrap();
+    w.write_all(&msg_frame(&s.send_key, 2, b"ok-real"))
+        .await
+        .unwrap();
+    tokio::time::sleep(Duration::from_millis(300)).await;
+    assert_eq!(
+        rx.load(Ordering::SeqCst),
+        1,
+        "junk skipped, real frame arrives"
+    );
+    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 decision_deny_rejects_after_handshake() {
+    // A fully authenticated client can still be refused; it must see EOF
+    // and no data path may open.
+    let port = 18510;
+    let seen = Arc::new(AtomicU8::new(0));
+    let s = Arc::clone(&seen);
+    let server = omnivox::server::<Msg>()
+        .key(TEST_PSK)
+        .config(test_config())
+        .on_data(move |_: &Msg, _| {
+            s.store(9, Ordering::SeqCst); // must never run
+        })
+        .on_connect(move |_ctx| Decision::Deny)
+        .serve(port)
+        .await
+        .unwrap();
+
+    let (mut r, mut w, sess) = client_session(port, &TEST_PSK).await;
+    // handshake completed (client_session); the Deny must have closed us.
+    expect_eof(&mut r).await;
+    let _ = w
+        .write_all(&msg_frame(&sess.send_key, 1, b"ok-after-deny"))
+        .await;
+    tokio::time::sleep(Duration::from_millis(200)).await;
+    assert_eq!(
+        seen.load(Ordering::SeqCst),
+        0,
+        "denied client never reaches data path"
+    );
+    server.stop().await;
+}

tests/support/mod.rs

@@ -1,39 +1,55 @@
-//! Shared helpers for the integration tests: v2 client sessions against a
-//! live server, and raw frame construction.
+#![allow(dead_code)] // each test binary uses a subset
 
-use omnivox::{Cereal, CodecError, Session, encrypt_with_nonce, frame_nonce, handshake::perform};
+//! Shared helpers for the integration tests: raw v2.1 client sessions
+//! (handshake completed, manual framing) for the attack regressions.
+
+use omnivox::{
+    FrameType,
+    frame::{FrameType as Ft, seal},
+    handshake::{Session, perform},
+};
+use serde::{Deserialize, Serialize};
 use tokio::net::TcpStream;
 use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
 
-#[derive(Debug, Default)]
+/// Test message type: serde-derives make it a `Message` by blanket impl.
+#[derive(Serialize, Deserialize, Debug, PartialEq)]
 pub struct Msg(pub Vec<u8>);
 
-impl Cereal for Msg {
-    fn encode(&self, out: &mut Vec<u8>) {
-        omnivox::write_byte_vec(out, &self.0);
-    }
-    fn decode(bytes: &[u8]) -> Result<Self, CodecError> {
-        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<u8> {
-    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
+pub fn bincode_cfg() -> bincode::config::Configuration<
+    bincode::config::LittleEndian,
+    bincode::config::Fixint,
+    bincode::config::Limit<16777216>,
+> {
+    bincode::config::standard()
+        .with_fixed_int_encoding()
+        .with_limit::<16777216>()
 }
 
-/// Connects and completes the v2 handshake as the client role, returning
-/// the raw halves plus the session keys for manual framing.
+/// Serialize like the library does.
+pub fn encode(msg: &Msg) -> Vec<u8> {
+    bincode::serde::encode_to_vec(msg, bincode_cfg()).unwrap()
+}
+
+/// One complete framed message (with u32 length prefix) ready to write.
+pub fn msg_frame(key: &[u8; 32], counter: u64, payload: &[u8]) -> Vec<u8> {
+    seal(key, counter, Ft::Message, &encode(&Msg(payload.to_vec())))
+}
+
+/// One complete framed stream chunk.
+pub fn stream_frame(key: &[u8; 32], counter: u64, payload: &[u8]) -> Vec<u8> {
+    seal(key, counter, Ft::Stream, payload)
+}
+
+/// Arbitrary-kind frame for protocol-violation tests.
+pub fn raw_frame(key: &[u8; 32], counter: u64, kind: FrameType, plaintext: &[u8]) -> Vec<u8> {
+    seal(key, counter, kind, plaintext)
+}
+
+/// Connect and complete the v2 handshake as the client role, returning
+/// raw halves and 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();