#![allow(clippy::items_after_statements)]
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::num::NonZeroU64;
use std::ops::RangeInclusive;
use std::time::{Duration, Instant};
use super::*;
use crate::config::Config;
use crate::constants::{
AEAD_TAG_LEN, APPLICATION_ERROR_BASE, CLOSE_LINGER, CLOSE_REASON_MAX, CLOSE_REPLY_MIN_INTERVAL,
DATA_HEADER_LEN, DEAD_TIMEOUT, FRAME_ACK, FRAME_CLOSE, FRAME_DATAGRAM, FRAME_DATAGRAM_LEN,
FRAME_MAX_DATA, FRAME_MAX_STREAM_DATA, FRAME_MAX_STREAMS_BIDI, FRAME_MAX_STREAMS_UNI,
FRAME_PADDING, FRAME_PING, FRAME_RESET_STREAM, FRAME_STOP_SENDING_RESERVED, FRAME_STREAM_BASE,
FRAME_STREAM_MAX, MAX_ACK_RANGES, MAX_DATAGRAM, MAX_PLAINTEXT, PKT_DATA, PROLOGUE,
REKEY_EPOCH_MSGS, REPLAY_WINDOW, VERSION,
};
use crate::core::{EstablishedSession, Install, Role, ToEndpoint, Transmit};
use crate::error::ConnectionLost;
use crate::identity::Identity;
use crate::packet::{Handshake, ReferenceSuite};
use crate::testutil::CountingIdentity;
use crate::varint::{self, VarInt};
type Suite = ReferenceSuite;
type Id = CountingIdentity<Suite>;
#[derive(Debug, Default)]
struct Drained {
outs: Vec<ConnOutput>,
deadline: Option<Instant>,
}
impl Drained {
fn transmits(&self) -> Vec<Transmit> {
self.outs
.iter()
.filter_map(|o| match o {
ConnOutput::Transmit(t) => Some(t.clone()),
_ => None,
})
.collect()
}
fn events(&self) -> Vec<&ConnEvent> {
self.outs
.iter()
.filter_map(|o| match o {
ConnOutput::Event(e) => Some(e),
_ => None,
})
.collect()
}
fn closed(&self) -> Vec<ConnectionLost> {
self.outs
.iter()
.filter_map(|o| match o {
ConnOutput::Event(ConnEvent::Closed(l)) => Some(l.clone()),
_ => None,
})
.collect()
}
fn retired(&self) -> Vec<u32> {
self.outs
.iter()
.filter_map(|o| match o {
ConnOutput::ToEndpoint(ToEndpoint::Retired { our_index }) => Some(*our_index),
_ => None,
})
.collect()
}
fn one_transmit(&self) -> Transmit {
let v = self.transmits();
assert_eq!(v.len(), 1, "expected exactly one Transmit, got {}", v.len());
v[0].clone()
}
fn is_silent(&self) -> bool {
self.outs.is_empty()
}
fn position(&self, f: impl Fn(&ConnOutput) -> bool) -> Option<usize> {
self.outs.iter().position(f)
}
}
struct Peer {
seal: <Suite as Handshake>::Seal,
open: <Suite as Handshake>::Open,
our_index: u32,
peer_index: u32,
addr: SocketAddr,
}
impl Peer {
fn seal(&mut self, plaintext: &[u8]) -> Vec<u8> {
let counter = self.seal.next_counter();
let mut dgram = data_header(self.peer_index, counter);
let mut body = vec![0u8; plaintext.len() + AEAD_TAG_LEN];
let (got, n) = self
.seal
.encrypt_next(&dgram, plaintext, &mut body)
.expect("peer seal");
assert_eq!(got, counter, "next_counter() must predict the seal");
body.truncate(n);
dgram.extend_from_slice(&body);
dgram
}
fn seal_at(&mut self, counter: u64, plaintext: &[u8]) -> Vec<u8> {
assert!(
counter >= self.seal.next_counter(),
"the counter only goes forward (§7.1)"
);
while self.seal.next_counter() < counter {
let _ = self.seal(&padding(1));
}
self.seal(plaintext)
}
fn open(&mut self, dgram: &[u8]) -> Vec<u8> {
let (header, body) = dgram.split_at(DATA_HEADER_LEN);
assert_eq!(header[0], PKT_DATA, "§3.4 packet type");
assert_eq!(header[1], VERSION, "§3.4 version");
assert_eq!(
u32::from_le_bytes(header[2..6].try_into().unwrap()),
self.our_index,
"§3.4 receiver_index routes to the peer's own index"
);
let counter = u64::from_le_bytes(header[6..14].try_into().unwrap());
let mut out = vec![0u8; body.len()];
let n = self
.open
.decrypt_at(counter, header, body, &mut out)
.expect("the packet must open");
out.truncate(n);
out
}
fn counter_of(dgram: &[u8]) -> u64 {
u64::from_le_bytes(dgram[6..14].try_into().expect("data header"))
}
}
fn data_header(receiver_index: u32, counter: u64) -> Vec<u8> {
let mut h = Vec::with_capacity(DATA_HEADER_LEN);
h.push(PKT_DATA);
h.push(VERSION);
h.extend_from_slice(&receiver_index.to_le_bytes());
h.extend_from_slice(&counter.to_le_bytes());
assert_eq!(h.len(), DATA_HEADER_LEN);
h
}
struct Fixture {
conn: Connection<Suite>,
peer: Peer,
our_index: u32,
}
impl Fixture {
fn drain(&mut self) -> Drained {
let mut d = Drained::default();
for _ in 0..100_000 {
match self.conn.poll_output() {
ConnOutput::Timeout(t) => {
d.deadline = t;
return d;
}
other => d.outs.push(other),
}
}
panic!("poll_output() did not reach the terminal Timeout in 100_000 outputs (§16.4)");
}
fn feed(&mut self, now: Instant, dgram: &[u8]) -> Drained {
let src = self.peer.addr;
self.conn.handle_datagram(now, src, dgram);
self.drain()
}
fn feed_from(&mut self, now: Instant, src: SocketAddr, dgram: &[u8]) -> Drained {
self.conn.handle_datagram(now, src, dgram);
self.drain()
}
fn deliver(&mut self, now: Instant, frames: &[u8]) -> Drained {
let dgram = self.peer.seal(frames);
self.feed(now, &dgram)
}
fn timeout(&mut self, now: Instant) -> Drained {
self.conn.handle_timeout(now);
self.drain()
}
fn close(&mut self, now: Instant, code: u64, reason: &[u8]) -> Drained {
self.conn.close(now, code, reason);
self.drain()
}
fn next_counter(&self) -> u64 {
self.conn
.session()
.expect("established")
.seal
.next_counter()
}
}
const PEER_ADDR_OCTET: u8 = 2;
fn v4(a: u8, port: u16) -> SocketAddr {
SocketAddr::new(IpAddr::V4(Ipv4Addr::new(10, 0, 0, a)), port)
}
fn t0() -> Instant {
Instant::now()
}
fn handshake(epoch: NonZeroU64) -> (EstablishedSession<Suite>, Peer) {
const A_INDEX: u32 = 0x1111_1111; const B_INDEX: u32 = 0x2222_2222;
let a: Id = CountingIdentity::seeded([7u8; 32]);
let b: Id = CountingIdentity::seeded([9u8; 32]);
let a_pub = *a.public_static();
let b_pub = *b.public_static();
let (ap, ask) = a.open().expect("identity opens");
let (bp, bsk) = b.open().expect("identity opens");
let init = <Suite as Handshake>::initiator(ap, PROLOGUE, b_pub);
let (msg1, sent) =
<Suite as Handshake>::write_msg1(init, ask, &[0u8; crate::constants::MSG1_PAYLOAD_LEN])
.expect("msg1");
let resp = <Suite as Handshake>::responder(bp, PROLOGUE, bsk).expect("responder");
let (claimed, mid) = <Suite as Handshake>::read_msg1_intro(resp, &msg1).expect("msg1 intro");
assert_eq!(
claimed.as_ref(),
a_pub.as_ref(),
"the claimed static is the initiator's"
);
let (_payload, read) = <Suite as Handshake>::complete(mid).expect("complete");
let (msg2, b_transport) = <Suite as Handshake>::write_msg2(read).expect("msg2");
let a_transport = <Suite as Handshake>::read_msg2(sent, &msg2).expect("read msg2");
let (a_seal, a_open) = <Suite as Handshake>::into_datagram(a_transport, epoch);
let (b_seal, b_open) = <Suite as Handshake>::into_datagram(b_transport, epoch);
let peer_addr = v4(PEER_ADDR_OCTET, 2);
(
EstablishedSession {
seal: a_seal,
open: a_open,
our_index: A_INDEX,
peer_index: B_INDEX,
anchor: peer_addr,
},
Peer {
seal: b_seal,
open: b_open,
our_index: B_INDEX,
peer_index: A_INDEX,
addr: peer_addr,
},
)
}
fn default_epoch() -> NonZeroU64 {
NonZeroU64::new(REKEY_EPOCH_MSGS).expect("REKEY_EPOCH_MSGS is nonzero")
}
fn established_at(now: Instant) -> Fixture {
established_at_with_epoch(now, default_epoch())
}
fn established_at_with_epoch(now: Instant, epoch: NonZeroU64) -> Fixture {
let (session, peer) = handshake(epoch);
let our_index = session.our_index;
let mut conn = Connection::connecting([0x5au8; 32]);
conn.handle_endpoint_event(
now,
Install {
session,
role: Role::Initiator,
anchor_from_msg1: false,
},
);
let mut f = Fixture {
conn,
peer,
our_index,
};
let d = f.drain();
assert!(
d.events()
.iter()
.any(|e| matches!(e, ConnEvent::Established)),
"§16.4: the install emits Established"
);
assert!(
f.conn.is_established(),
"§16.4: the install establishes the connection"
);
f
}
fn accepted_at(now: Instant) -> Fixture {
let (session, peer) = handshake(default_epoch());
let our_index = session.our_index;
let conn = Connection::established(now, [0x5au8; 32], session, Role::Responder);
let mut f = Fixture {
conn,
peer,
our_index,
};
let _ = f.drain();
f
}
fn connecting() -> Connection<Suite> {
Connection::connecting([0x5au8; 32])
}
fn drain_bare(conn: &mut Connection<Suite>) -> Drained {
let mut d = Drained::default();
for _ in 0..100_000 {
match conn.poll_output() {
ConnOutput::Timeout(t) => {
d.deadline = t;
return d;
}
other => d.outs.push(other),
}
}
panic!("poll_output() did not reach the terminal Timeout in 100_000 outputs (§16.4)");
}
const NS: Duration = Duration::from_nanos(1);
fn vi(v: u64) -> Vec<u8> {
let mut out = Vec::new();
varint::encode(VarInt::new(v).expect("fits the 62-bit space"), &mut out);
out
}
fn vi_padded(v: u64, len: usize) -> Vec<u8> {
let (prefix, bytes) = match len {
2 => (0b01u8, 2usize),
4 => (0b10u8, 4),
8 => (0b11u8, 8),
_ => panic!("non-minimal widths are 2, 4 and 8"),
};
assert!(
v < (1u64 << (bytes * 8 - 2)),
"value does not fit the width"
);
let mut out = v.to_be_bytes()[8 - bytes..].to_vec();
out[0] |= prefix << 6;
out
}
fn padding(n: usize) -> Vec<u8> {
vec![u8::try_from(FRAME_PADDING).unwrap(); n]
}
fn ping() -> Vec<u8> {
vi(FRAME_PING)
}
fn ack(largest: u64, delay: u64, first_range: u64, pairs: &[(u64, u64)]) -> Vec<u8> {
let mut f = vi(FRAME_ACK);
f.extend(vi(largest));
f.extend(vi(delay));
f.extend(vi(pairs.len() as u64));
f.extend(vi(first_range));
for (gap, range) in pairs {
f.extend(vi(*gap));
f.extend(vi(*range));
}
f
}
fn close_frame(code: u64, reason: &[u8]) -> Vec<u8> {
let mut f = vi(FRAME_CLOSE);
f.extend(vi(code));
f.extend(vi(reason.len() as u64));
f.extend_from_slice(reason);
f
}
#[derive(Debug, PartialEq, Eq)]
struct ParsedClose {
code: u64,
reason: Vec<u8>,
}
fn parse_frames(mut p: &[u8]) -> Vec<PeerFrame> {
let mut out = Vec::new();
while !p.is_empty() {
let (ty, n) = varint::decode(p).expect("a frame type");
p = &p[n..];
let ty = u64::from(ty);
if ty == FRAME_PADDING {
out.push(PeerFrame::Padding);
} else if ty == FRAME_PING {
out.push(PeerFrame::Ping);
} else if ty == FRAME_ACK {
let (largest, n) = varint::decode(p).expect("largest");
p = &p[n..];
let (delay, n) = varint::decode(p).expect("ack_delay");
p = &p[n..];
let (count, n) = varint::decode(p).expect("range_count");
p = &p[n..];
let (first, n) = varint::decode(p).expect("first_range");
p = &p[n..];
let mut pairs = Vec::new();
for _ in 0..u64::from(count) {
let (gap, n) = varint::decode(p).expect("gap");
p = &p[n..];
let (range, n) = varint::decode(p).expect("range");
p = &p[n..];
pairs.push((u64::from(gap), u64::from(range)));
}
out.push(PeerFrame::Ack {
largest: u64::from(largest),
delay: u64::from(delay),
first_range: u64::from(first),
pairs,
});
} else if ty == FRAME_CLOSE {
let (code, n) = varint::decode(p).expect("error_code");
p = &p[n..];
let (len, n) = varint::decode(p).expect("reason_len");
p = &p[n..];
let len = usize::try_from(u64::from(len)).expect("reason fits");
assert!(p.len() >= len, "reason overruns the plaintext");
out.push(PeerFrame::Close(ParsedClose {
code: u64::from(code),
reason: p[..len].to_vec(),
}));
p = &p[len..];
} else {
panic!("slither emitted frame type {ty:#x}, which slice 3a has no business sending");
}
}
out
}
#[derive(Debug, PartialEq, Eq)]
enum PeerFrame {
Padding,
Ping,
Ack {
largest: u64,
delay: u64,
first_range: u64,
pairs: Vec<(u64, u64)>,
},
Close(ParsedClose),
}
fn one_close(peer: &mut Peer, dgram: &[u8]) -> ParsedClose {
let pt = peer.open(dgram);
let frames = parse_frames(&pt);
let closes: Vec<_> = frames
.into_iter()
.filter_map(|f| match f {
PeerFrame::Close(c) => Some(c),
_ => None,
})
.collect();
assert_eq!(closes.len(), 1, "expected exactly one CLOSE frame");
closes.into_iter().next().expect("one")
}
mod poll_contract {
use super::*;
#[test]
fn the_drain_always_terminates_in_timeout() {
let t = t0();
let (session, mut peer) = handshake(default_epoch());
let mut conn = Connection::connecting([1u8; 32]);
conn.handle_endpoint_event(
t,
Install {
session,
role: Role::Initiator,
anchor_from_msg1: false,
},
);
let _ = drain_bare(&mut conn);
let dgram = peer.seal(&padding(1));
conn.handle_datagram(t, peer.addr, &dgram);
let _ = drain_bare(&mut conn);
conn.handle_datagram(t, peer.addr, b"not a slither packet at all");
let _ = drain_bare(&mut conn);
conn.handle_timeout(t + Duration::from_secs(1));
let _ = drain_bare(&mut conn);
conn.close(t + Duration::from_secs(2), 0, b"bye");
let _ = drain_bare(&mut conn);
let mut bare = connecting();
let _ = drain_bare(&mut bare);
bare.handle_datagram(t, v4(9, 9), b"nothing");
let _ = drain_bare(&mut bare);
bare.handle_timeout(t);
let _ = drain_bare(&mut bare);
bare.close(t, 0, b"");
let _ = drain_bare(&mut bare);
}
#[test]
fn a_second_drain_with_no_mutating_call_yields_only_timeout() {
let t = t0();
let mut f = established_at(t);
let first = f.close(t, 0, b"bye");
assert!(!first.outs.is_empty(), "close() produces output");
let second = f.drain();
assert!(
second.is_silent(),
"a re-drain must be empty, got {:?}",
second.outs
);
assert_eq!(
second.deadline, first.deadline,
"the announced deadline is stable across drains"
);
}
#[test]
fn the_close_transmit_precedes_the_closed_event_it_caused() {
let t = t0();
let mut f = established_at(t);
let d = f.close(t, 0, b"bye");
let tx = d
.position(|o| matches!(o, ConnOutput::Transmit(_)))
.expect("the CLOSE transmit");
let ev = d
.position(|o| matches!(o, ConnOutput::Event(ConnEvent::Closed(_))))
.expect("the Closed event");
assert!(
tx < ev,
"§16.4: the transmit precedes the event it caused ({tx} vs {ev})"
);
}
#[test]
fn an_install_establishes_once_and_preserves_the_sub_seed() {
let t = t0();
let (session, _peer) = handshake(default_epoch());
let index = session.our_index;
let mut conn = Connection::connecting([0xa5u8; 32]);
assert!(!conn.is_established(), "not established before the install");
assert!(conn.session().is_none(), "no session before the install");
assert_eq!(conn.sub_seed(), &[0xa5u8; 32], "§16.6's sub-seed is held");
conn.handle_endpoint_event(
t,
Install {
session,
role: Role::Initiator,
anchor_from_msg1: false,
},
);
let d = drain_bare(&mut conn);
let established = d
.events()
.iter()
.filter(|e| matches!(e, ConnEvent::Established))
.count();
assert_eq!(established, 1, "§16.4: exactly one install, one event");
assert!(conn.is_established());
assert_eq!(
conn.session().expect("a session").our_index,
index,
"the installed session is the one that was handed over"
);
assert_eq!(
conn.sub_seed(),
&[0xa5u8; 32],
"§16.6: the sub-seed survives the install — it is the connection's, not the session's"
);
}
#[test]
fn a_datagram_before_the_install_is_silent() {
let t = t0();
let mut conn = connecting();
conn.handle_datagram(t, v4(9, 9), &data_header(1, 0));
let d = drain_bare(&mut conn);
assert!(d.is_silent(), "silent drop, got {:?}", d.outs);
assert_eq!(d.deadline, None, "nothing is armed on a bare connection");
assert!(!conn.is_established());
}
#[test]
fn a_runt_or_mistyped_datagram_is_a_silent_drop() {
let t = t0();
let mut f = established_at(t);
for bad in [
vec![],
vec![PKT_DATA],
vec![PKT_DATA, VERSION],
data_header(f.our_index, 0)[..DATA_HEADER_LEN - 1].to_vec(),
data_header(f.our_index, 0), ] {
let d = f.feed(t, &bad);
assert!(
d.is_silent(),
"a malformed datagram must drop silently, got {:?}",
d.outs
);
}
assert!(f.conn.is_established(), "and must not kill the connection");
}
}
mod counter {
use super::*;
#[test]
fn the_first_seal_of_a_connection_is_counter_zero() {
let t = t0();
let mut f = established_at(t);
assert_eq!(f.next_counter(), 0, "§7.1: the space starts at 0");
let d = f.close(t, 0, b"bye");
let tx = d.one_transmit();
assert_eq!(
Peer::counter_of(&tx.data),
0,
"§7.1: the first seal burns counter 0"
);
}
#[test]
fn every_seal_burns_the_next_counter_and_never_repeats() {
let t = t0();
let mut f = established_at(t);
let mut counters = vec![];
counters.push(Peer::counter_of(&f.close(t, 0, b"bye").one_transmit().data));
for k in 1..=2u32 {
let at = t + CLOSE_REPLY_MIN_INTERVAL * k + Duration::from_millis(1);
let dgram = f.peer.seal(&padding(1));
let d = f.feed(at, &dgram);
counters.push(Peer::counter_of(&d.one_transmit().data));
}
assert_eq!(
counters,
vec![0, 1, 2],
"§7.1: monotonic, contiguous, never reused"
);
assert_eq!(f.next_counter(), 3, "three seals, three counters");
}
#[test]
fn the_close_packet_routes_and_opens_under_the_peers_key() {
let t = t0();
let mut f = established_at(t);
let tx = f.close(t, 0x11, b"bye").one_transmit();
assert_eq!(tx.to, f.peer.addr, "§15.2: to the session's address");
assert!(
tx.data.len() <= MAX_DATAGRAM,
"§8.6/§3.5: never above the MTU"
);
assert_eq!(tx.data[0], PKT_DATA);
assert_eq!(tx.data[1], VERSION);
assert_eq!(
u32::from_le_bytes(tx.data[2..6].try_into().unwrap()),
f.peer.our_index,
"§3.4: receiver_index is the *peer's* index, little-endian"
);
let close = one_close(&mut f.peer, &tx.data);
assert_eq!(close.code, 0x11);
assert_eq!(close.reason, b"bye");
}
}
mod replay {
use super::*;
fn probe(code: u64) -> Vec<u8> {
close_frame(code, b"probe")
}
fn was_delivered(d: &Drained, code: u64) -> bool {
d.closed()
.iter()
.any(|l| matches!(l, ConnectionLost::PeerClosed { code: c, .. } if *c == code))
}
#[test]
fn a_duplicate_counter_is_dropped_without_delivery() {
let t = t0();
let mut f = established_at(t);
let dgram = f.peer.seal(&padding(1));
let first = f.feed(t, &dgram);
assert!(first.is_silent(), "a PADDING packet is silent");
let again = f.feed(t + Duration::from_secs(1), &dgram);
assert!(again.is_silent(), "the duplicate produced {:?}", again.outs);
assert!(f.conn.is_established(), "and did not kill the connection");
}
#[test]
fn a_counter_exactly_the_window_behind_the_greatest_is_delivered() {
let t = t0();
let mut f = established_at(t);
let window = REPLAY_WINDOW as u64;
let edge = f.peer.seal_at(1, &probe(0x21));
let greatest = f.peer.seal_at(1 + window, &padding(1));
assert!(f.feed(t, &greatest).is_silent(), "greatest lands silently");
let d = f.feed(t, &edge);
assert!(
was_delivered(&d, 0x21),
"greatest − {window} must be delivered (§7.2), got {:?}",
d.outs
);
}
#[test]
fn a_counter_one_past_the_window_is_dropped() {
let t = t0();
let mut f = established_at(t);
let window = REPLAY_WINDOW as u64;
let past = f.peer.seal_at(1, &probe(0x22));
let greatest = f.peer.seal_at(2 + window, &padding(1));
assert!(f.feed(t, &greatest).is_silent());
let d = f.feed(t, &past);
assert!(
!was_delivered(&d, 0x22),
"greatest − {} must be dropped (§7.2)",
window + 1
);
assert!(d.is_silent(), "and dropped silently, got {:?}", d.outs);
assert!(f.conn.is_established(), "a stale packet is not a violation");
}
#[test]
fn the_far_edge_of_the_window_is_marked_not_merely_admitted() {
let t = t0();
let mut f = established_at(t);
let window = REPLAY_WINDOW as u64;
let edge = f.peer.seal_at(1, &padding(1));
let edge_replay_probe = edge.clone();
let greatest = f.peer.seal_at(1 + window, &padding(1));
assert!(f.feed(t, &greatest).is_silent());
assert!(f.feed(t, &edge).is_silent(), "the edge is accepted");
let d0 = f.close(t, 0, b"bye");
assert_eq!(d0.transmits().len(), 1, "the local CLOSE");
let at = t + CLOSE_REPLY_MIN_INTERVAL + Duration::from_millis(1);
let d = f.feed(at, &edge_replay_probe);
assert!(
d.transmits().is_empty(),
"a replayed far-edge packet is not window-fresh and owes no reply"
);
}
#[test]
fn a_failed_decryption_never_burns_its_counter() {
let t = t0();
let mut f = established_at(t);
let genuine = f.peer.seal(&probe(0x23));
let mut corrupt = genuine.clone();
let last = corrupt.len() - 1;
corrupt[last] ^= 0xff;
let d = f.feed(t, &corrupt);
assert!(
d.is_silent(),
"a forgery is a silent drop, got {:?}",
d.outs
);
assert!(f.conn.is_established(), "and never a protocol violation");
let d = f.feed(t + Duration::from_millis(1), &genuine);
assert!(
was_delivered(&d, 0x23),
"the genuine packet at the same counter must still be delivered (§7.2)"
);
}
#[test]
fn reordering_inside_the_window_is_delivered() {
let t = t0();
let mut f = established_at(t);
let older = f.peer.seal_at(10, &padding(1));
let decisive = f.peer.seal_at(11, &probe(0x24));
let newest = f.peer.seal_at(12, &padding(1));
assert!(f.feed(t, &newest).is_silent());
assert!(
f.feed(t, &older).is_silent(),
"an older counter still lands"
);
let d = f.feed(t, &decisive);
assert!(
was_delivered(&d, 0x24),
"a counter below the greatest and inside the window is delivered (§7.2)"
);
}
}
mod liveness {
use super::*;
use crate::constants::KEEPALIVE_TIMEOUT;
#[test]
fn a_new_session_arms_the_death_clock_at_install() {
let t = t0();
let mut f = established_at(t);
let d = f.drain();
assert_eq!(
d.deadline,
Some(t + DEAD_TIMEOUT),
"§7.4: armed at install, anchored at the install instant"
);
}
#[test]
fn a_half_open_session_is_reaped_in_silence_and_not_one_nanosecond_early() {
let t = t0();
let mut f = established_at(t);
let early = f.timeout(t + DEAD_TIMEOUT - NS);
assert!(
early.closed().is_empty(),
"not dead before DEAD_TIMEOUT, got {:?}",
early.outs
);
assert!(
early.transmits().is_empty(),
"§15.4: liveness transmits nothing"
);
let d = f.timeout(t + DEAD_TIMEOUT + NS);
assert_eq!(
d.closed(),
vec![ConnectionLost::TimedOut],
"§15.4: the liveness row surfaces TimedOut"
);
assert!(
d.transmits().is_empty(),
"§15.4: liveness transmits *nothing* — no CLOSE, no probe"
);
}
#[test]
fn a_liveness_death_retires_in_the_same_drain() {
let t = t0();
let mut f = established_at(t);
let d = f.timeout(t + DEAD_TIMEOUT + NS);
assert_eq!(d.retired(), vec![f.our_index], "§16.4: Retired is a MUST");
let closed = d
.position(|o| matches!(o, ConnOutput::Event(ConnEvent::Closed(_))))
.expect("Closed");
let retired = d
.position(|o| matches!(o, ConnOutput::ToEndpoint(ToEndpoint::Retired { .. })))
.expect("Retired");
assert!(
closed < retired,
"Closed precedes Retired ({closed} vs {retired})"
);
assert_eq!(d.deadline, None, "nothing is armed after the death");
}
#[test]
fn an_unauthenticated_datagram_never_refreshes_liveness() {
let t = t0();
let mut f = established_at(t);
let mut forged = f.peer.seal(&padding(1));
let last = forged.len() - 1;
forged[last] ^= 0xff;
let mid = t + DEAD_TIMEOUT / 2;
assert!(f.feed(mid, &forged).is_silent(), "a forgery is silent");
let d = f.drain();
assert_eq!(
d.deadline,
Some(t + DEAD_TIMEOUT),
"the anchor must not have moved to {mid:?}"
);
let d = f.timeout(t + DEAD_TIMEOUT + NS);
assert_eq!(
d.closed(),
vec![ConnectionLost::TimedOut],
"the forgery bought no time at all"
);
}
#[test]
fn a_fresh_receive_re_anchors_the_clock_and_the_keepalive_then_arms_it() {
let t = t0();
let mut f = established_at(t);
let r = t + Duration::from_secs(10);
let dgram = f.peer.seal(&padding(1));
assert!(f.feed(r, &dgram).is_silent());
let d = f.timeout(t + DEAD_TIMEOUT + NS);
assert!(
d.closed().is_empty(),
"the receive re-anchored the clock, got {:?}",
d.outs
);
let d = f.timeout(r + DEAD_TIMEOUT + NS);
assert_eq!(
d.closed(),
vec![ConnectionLost::TimedOut],
"§7.4 + §7.5: the keepalive is the arming send, and death lands at the new anchor"
);
}
#[test]
fn a_fresh_receive_disarms_the_liveness_deadline() {
let t = t0();
let mut f = established_at(t);
let r = t + Duration::from_secs(10);
let dgram = f.peer.seal(&padding(1));
let d = f.feed(r, &dgram);
assert_ne!(
d.deadline,
Some(r + DEAD_TIMEOUT),
"§16.5: a receive disarms Liveness; it does not re-arm it"
);
assert_eq!(
d.deadline,
Some(t + KEEPALIVE_TIMEOUT),
"slice 7 arms exactly one thing here: §7.5's passive keepalive"
);
}
}
mod ratchet {
use super::*;
#[test]
fn the_default_epoch_size_is_rekey_epoch_msgs() {
assert_eq!(
Config::new().epoch_size(),
NonZeroU64::new(REKEY_EPOCH_MSGS).expect("nonzero"),
"§7.7: `REKEY_EPOCH_MSGS` otherwise (ruling 82)"
);
assert_eq!(REKEY_EPOCH_MSGS, 65_536, "§7.7: 2¹⁶ messages per epoch");
}
#[test]
fn a_packet_two_epochs_back_is_refused_and_one_epoch_back_still_opens() {
let t = t0();
let epoch = NonZeroU64::new(8).expect("nonzero");
let mut f = established_at_with_epoch(t, epoch);
let e0 = f.peer.seal_at(1, &close_frame(0x40, b"e0"));
let e1 = f.peer.seal_at(8, &close_frame(0x41, b"e1"));
let e2 = f.peer.seal_at(16, &padding(1));
assert!(f.feed(t, &e2).is_silent(), "epoch 2 opens and is silent");
let d = f.feed(t, &e0);
assert!(
d.closed().is_empty(),
"a packet two epochs back must not open (§7.7), got {:?}",
d.outs
);
assert!(d.is_silent(), "and must be a silent drop, got {:?}", d.outs);
assert!(f.conn.is_established(), "and must not kill the connection");
let d = f.feed(t, &e1);
assert!(
d.closed()
.iter()
.any(|l| matches!(l, ConnectionLost::PeerClosed { code: 0x41, .. })),
"§7.7: one epoch back is retained (straggler tolerance), got {:?}",
d.outs
);
}
#[test]
fn crossing_an_epoch_boundary_is_invisible() {
let t = t0();
let epoch = NonZeroU64::new(4).expect("nonzero");
let mut f = established_at_with_epoch(t, epoch);
for c in 0..=12u64 {
let dgram = f.peer.seal(&padding(1));
assert_eq!(Peer::counter_of(&dgram), c);
let d = f.feed(t, &dgram);
assert!(
d.is_silent(),
"counter {c} (epoch {}) produced {:?}",
c / 4,
d.outs
);
}
assert!(
f.conn.is_established(),
"three epoch boundaries crossed, connection untouched"
);
assert_eq!(f.conn.session().expect("session").our_index, f.our_index);
}
}
mod one_session {
use super::*;
#[test]
fn the_session_and_its_counter_space_are_the_connections_own() {
let t = t0();
let mut f = established_at(t);
let index = f.conn.session().expect("session").our_index;
for _ in 0..4 {
let dgram = f.peer.seal(&padding(1));
assert!(f.feed(t, &dgram).is_silent());
}
assert_eq!(f.next_counter(), 0, "receives burn no send counters");
let tx = f.close(t, 0, b"bye").one_transmit();
assert_eq!(Peer::counter_of(&tx.data), 0, "§7.1: still starting at 0");
assert_eq!(
f.conn.session().expect("session").our_index,
index,
"§7.8: the same session throughout"
);
}
}
mod exhaustion {
use super::*;
#[test]
fn a_seal_failure_kills_the_connection_and_moves_nothing() {
let t = t0();
let mut f = established_at(t);
let before = f.next_counter();
f.conn.fail_next_seal();
let d = f.close(t, 0, b"bye");
assert_eq!(
d.closed(),
vec![ConnectionLost::NonceExhausted],
"§7.9: the seal failure is never silent"
);
assert!(
d.transmits().is_empty(),
"§16.7: nothing was sealed, so nothing may be transmitted"
);
assert_ne!(
f.conn.session().map(|s| s.seal.next_counter()),
Some(before + 1),
"§16.7: on seal failure nothing moved — the counter must not have advanced"
);
assert_eq!(
d.retired(),
vec![f.our_index],
"ruling 81: no linger, so Retired lands in the same drain"
);
assert_eq!(d.deadline, None, "and no linger deadline is armed");
}
}
mod codec {
use super::*;
const PROTOCOL_VIOLATION: u64 = 0x01;
fn assert_structural_failure(frames: &[u8], what: &str) {
let t = t0();
let mut f = established_at(t);
let d = f.deliver(t, frames);
assert_eq!(
d.closed(),
vec![ConnectionLost::ProtocolViolation {
code: PROTOCOL_VIOLATION
}],
"§8.2/§15.2: {what} is a signalled death, got {:?}",
d.outs
);
let tx = d.one_transmit();
let close = one_close(&mut f.peer, &tx.data);
assert_eq!(
close.code, PROTOCOL_VIOLATION,
"§8.2: CLOSE carries PROTOCOL_VIOLATION (0x01) for {what}"
);
assert_eq!(
d.deadline,
Some(t + CLOSE_LINGER),
"§15.2: a violation lingers as for a local close"
);
assert!(
d.retired().is_empty(),
"ruling 81: the linger has not expired yet"
);
}
fn assert_harmless(frames: &[u8], what: &str) {
let t = t0();
let mut f = established_at(t);
let d = f.deliver(t, frames);
assert!(
d.closed().is_empty(),
"§8: {what} must not kill the connection, got {:?}",
d.outs
);
for tr in d.transmits() {
let pt = f.peer.open(&tr.data);
for frame in parse_frames(&pt) {
assert!(
matches!(frame, PeerFrame::Ack { .. } | PeerFrame::Padding),
"§8: {what} owes nothing but §12's ACK, got {frame:?}"
);
}
}
assert!(f.conn.is_established());
}
#[test]
fn padding_alone_is_harmless() {
assert_harmless(&padding(1), "one PADDING byte");
assert_harmless(&padding(64), "sixty-four PADDING bytes");
}
#[test]
fn padding_may_appear_anywhere_around_other_frames() {
let mut s = padding(3);
s.extend(ping());
s.extend(padding(2));
s.extend(ack(0, 0, 0, &[]));
s.extend(padding(5));
assert_harmless(&s, "PADDING wrapped around PING and ACK");
}
#[test]
fn a_ping_is_accepted_and_answered_with_nothing_in_slice_3a() {
assert_harmless(&ping(), "a PING");
}
#[test]
fn a_well_formed_ack_is_decoded_and_not_acted_on() {
assert_harmless(&ack(0, 0, 0, &[]), "an ACK covering counter 0");
assert_harmless(
&ack(100, 1_234, 4, &[(0, 0), (3, 2)]),
"an ACK with two extra ranges",
);
}
#[test]
fn an_ack_largest_above_anything_sealed_is_ignored_not_fatal() {
assert_harmless(&ack(1_000_000, 0, 0, &[]), "an ACK from the future");
}
#[test]
fn the_max_ack_ranges_boundary_is_tested_from_both_sides() {
let at_cap: Vec<(u64, u64)> = vec![(0, 0); MAX_ACK_RANGES];
assert_harmless(
&ack(10_000, 0, 0, &at_cap),
"an ACK at exactly MAX_ACK_RANGES",
);
let over_cap: Vec<(u64, u64)> = vec![(0, 0); MAX_ACK_RANGES + 1];
assert_structural_failure(
&ack(10_000, 0, 0, &over_cap),
"an ACK with range_count = MAX_ACK_RANGES + 1",
);
}
#[test]
fn an_ack_range_descending_below_zero_is_a_structural_failure() {
assert_structural_failure(&ack(5, 0, 10, &[]), "an ACK range below counter zero");
}
#[test]
fn a_structural_ack_error_beats_the_semantic_no_op() {
let over_cap: Vec<(u64, u64)> = vec![(0, 0); MAX_ACK_RANGES + 1];
assert_structural_failure(
&ack(u64::from(u32::MAX), 0, 0, &over_cap),
"an over-long ACK whose largest is also unsealed",
);
}
#[test]
fn an_unknown_frame_type_is_a_structural_failure() {
assert_structural_failure(&vi(0x3f), "a 1-byte unknown type");
assert_structural_failure(&vi(0x7f), "a 2-byte unknown type");
}
#[test]
fn the_reserved_type_is_a_structural_failure() {
assert_structural_failure(
&vi(FRAME_STOP_SENDING_RESERVED),
"the reserved STOP_SENDING type",
);
}
#[test]
fn a_truncated_frame_is_a_structural_failure() {
let mut s = vi(FRAME_CLOSE);
s.extend(vi(0));
s.extend(vi(5));
s.extend_from_slice(b"ab");
assert_structural_failure(&s, "a CLOSE whose reason is short");
}
#[test]
fn a_varint_overrunning_the_plaintext_is_a_structural_failure() {
assert_structural_failure(&[0x80, 0x00], "a 4-byte varint with 2 bytes left");
assert_structural_failure(&[0xc0], "an 8-byte varint with 1 byte left");
}
#[test]
fn a_length_field_overrunning_the_plaintext_is_a_structural_failure() {
let mut s = vi(FRAME_CLOSE);
s.extend(vi(0));
s.extend(vi(200));
s.extend_from_slice(b"short");
assert_structural_failure(&s, "a CLOSE whose reason_len overruns");
}
#[test]
fn the_close_reason_max_boundary_is_tested_from_both_sides() {
let t = t0();
let mut f = established_at(t);
let reason = vec![b'r'; CLOSE_REASON_MAX];
let d = f.deliver(t, &close_frame(9, &reason));
assert_eq!(
d.closed(),
vec![ConnectionLost::PeerClosed {
code: 9,
reason: reason.clone()
}],
"§8.4: exactly CLOSE_REASON_MAX is valid"
);
let over = vec![b'r'; CLOSE_REASON_MAX + 1];
assert_structural_failure(
&close_frame(9, &over),
"a CLOSE reason one byte over CLOSE_REASON_MAX",
);
}
#[test]
fn parse_then_apply_a_valid_close_followed_by_garbage_is_a_violation() {
let mut s = close_frame(0x42, b"applied too early");
s.extend(vi(0x7f));
let t = t0();
let mut f = established_at(t);
let d = f.deliver(t, &s);
assert_eq!(
d.closed(),
vec![ConnectionLost::ProtocolViolation {
code: PROTOCOL_VIOLATION
}],
"§8.2: nothing from the packet is applied — not even a valid CLOSE"
);
let close = one_close(&mut f.peer, &d.one_transmit().data);
assert_eq!(
close.code, PROTOCOL_VIOLATION,
"and the CLOSE we send is ours, not an echo of theirs"
);
}
#[test]
fn garbage_before_a_valid_close_is_also_a_violation() {
let mut s = vi(0x7f);
s.extend(close_frame(0x42, b"never applied"));
assert_structural_failure(&s, "an unknown type ahead of a valid CLOSE");
}
#[test]
fn non_minimal_varints_are_accepted() {
let mut s = vi_padded(FRAME_CLOSE, 8);
s.extend(vi_padded(7, 4));
s.extend(vi_padded(2, 2));
s.extend_from_slice(b"hi");
let t = t0();
let mut f = established_at(t);
let d = f.deliver(t, &s);
assert_eq!(
d.closed(),
vec![ConnectionLost::PeerClosed {
code: 7,
reason: b"hi".to_vec()
}],
"§8.1: a non-minimal encoding decodes to the same value"
);
}
#[test]
fn a_close_code_at_the_varint_maximum_round_trips() {
let max = (1u64 << 62) - 1;
let t = t0();
let mut f = established_at(t);
let d = f.deliver(t, &close_frame(max, b""));
assert_eq!(
d.closed(),
vec![ConnectionLost::PeerClosed {
code: max,
reason: vec![]
}],
"§8.1: varints cap at 2⁶² − 1 and carry it exactly"
);
}
#[test]
fn the_varint_width_boundaries_round_trip_through_a_frame() {
for code in [63u64, 64, 16_383, 16_384, (1 << 30) - 1, 1 << 30] {
let t = t0();
let mut f = established_at(t);
let d = f.deliver(t, &close_frame(code, b""));
assert_eq!(
d.closed(),
vec![ConnectionLost::PeerClosed {
code,
reason: vec![]
}],
"§8.1: code {code} must survive its encoding width"
);
}
}
#[test]
fn an_empty_plaintext_is_not_a_frame_error() {
assert_harmless(&[], "an empty plaintext (the keepalive)");
}
#[test]
fn a_full_size_plaintext_is_accepted() {
assert_harmless(&padding(MAX_PLAINTEXT), "MAX_PLAINTEXT bytes of PADDING");
}
}
mod teardown {
use super::*;
const NO_ERROR: u64 = 0x00;
const PROTOCOL_VIOLATION: u64 = 0x01;
#[test]
fn a_local_close_emits_one_close_and_surfaces_locally_closed() {
let t = t0();
let mut f = established_at(t);
let d = f.close(t, NO_ERROR, b"bye");
let tx = d.one_transmit();
assert_eq!(tx.to, f.peer.addr, "§15.2: to the session's address");
assert_eq!(tx.data[0], PKT_DATA, "§15.1: authenticated, in-seal only");
let close = one_close(&mut f.peer, &tx.data);
assert_eq!(close.code, NO_ERROR);
assert_eq!(close.reason, b"bye");
assert_eq!(d.closed(), vec![ConnectionLost::LocallyClosed]);
assert_eq!(
d.deadline,
Some(t + CLOSE_LINGER),
"§15.2: closing for CLOSE_LINGER"
);
}
#[test]
fn retired_waits_for_the_linger_expiry_while_the_reply_rule_still_works() {
let t = t0();
let mut f = established_at(t);
let d = f.close(t, NO_ERROR, b"bye");
assert!(d.retired().is_empty(), "not at the death (ruling 81)");
let at = t + CLOSE_REPLY_MIN_INTERVAL + Duration::from_millis(1);
let dgram = f.peer.seal(&padding(1));
let d = f.feed(at, &dgram);
assert_eq!(
d.transmits().len(),
1,
"§15.2: the linger still owes a reply — this is what Retired would delete"
);
assert!(d.retired().is_empty(), "and still no Retired");
let d = f.timeout(t + CLOSE_LINGER);
assert_eq!(
d.retired(),
vec![f.our_index],
"ruling 81: Retired at the CloseLinger expiry, carrying our index"
);
assert!(
d.closed().is_empty(),
"§16.4/Q3: Closed is emitted exactly once, and it already was"
);
assert_eq!(d.deadline, None, "all state is dropped");
}
#[test]
fn the_linger_expires_at_exactly_close_linger_and_not_before() {
let t = t0();
let mut f = established_at(t);
let d = f.close(t, NO_ERROR, b"");
assert_eq!(d.deadline, Some(t + CLOSE_LINGER));
let early = f.timeout(t + CLOSE_LINGER - NS);
assert!(
early.retired().is_empty(),
"not one nanosecond early, got {:?}",
early.outs
);
assert_eq!(
early.deadline,
Some(t + CLOSE_LINGER),
"and the deadline is unmoved by a spurious wake"
);
let d = f.timeout(t + CLOSE_LINGER);
assert_eq!(d.retired(), vec![f.our_index]);
}
#[test]
fn handle_timeout_twice_at_the_expiry_retires_once() {
let t = t0();
let mut f = established_at(t);
let _ = f.close(t, NO_ERROR, b"");
f.conn.handle_timeout(t + CLOSE_LINGER);
f.conn.handle_timeout(t + CLOSE_LINGER);
let d = f.drain();
assert_eq!(
d.retired(),
vec![f.our_index],
"§16.5: exactly one Retired for two due calls"
);
let after = f.timeout(t + CLOSE_LINGER + Duration::from_secs(1));
assert!(
after.is_silent(),
"and nothing afterwards, got {:?}",
after.outs
);
}
#[test]
fn a_received_close_surfaces_peer_closed_transmits_nothing_and_drains() {
let t = t0();
let mut f = established_at(t);
let d = f.deliver(t, &close_frame(0x99, b"peer's reason"));
assert_eq!(
d.closed(),
vec![ConnectionLost::PeerClosed {
code: 0x99,
reason: b"peer's reason".to_vec()
}]
);
assert!(d.transmits().is_empty(), "§15.2: emit **nothing**");
assert_eq!(d.deadline, Some(t + CLOSE_LINGER), "the drain is armed");
assert!(d.retired().is_empty(), "ruling 81: not at the death");
let dgram = f.peer.seal(&padding(1));
let late = f.feed(t + Duration::from_secs(2), &dgram);
assert!(
late.transmits().is_empty(),
"§15.2: draining never replies, got {:?}",
late.transmits()
);
let d = f.timeout(t + CLOSE_LINGER);
assert_eq!(d.retired(), vec![f.our_index], "then drop all state");
}
#[test]
fn the_linger_replies_at_most_once_per_second_under_a_flood() {
let t = t0();
let mut f = established_at(t);
let _ = f.close(t, NO_ERROR, b"");
let base = t + Duration::from_secs(2);
let mut replies = 0;
for k in 0..10u32 {
let dgram = f.peer.seal(&padding(1));
replies += f
.feed(base + Duration::from_millis(u64::from(k) * 10), &dgram)
.transmits()
.len();
}
assert_eq!(
replies, 1,
"§15.2: ten fresh packets in 90 ms owe exactly one reply"
);
}
#[test]
fn the_linger_replies_again_after_the_rate_interval() {
let t = t0();
let mut f = established_at(t);
let _ = f.close(t, NO_ERROR, b"");
let first_at = t + Duration::from_secs(2);
let dgram = f.peer.seal(&padding(1));
assert_eq!(f.feed(first_at, &dgram).transmits().len(), 1);
let dgram = f.peer.seal(&padding(1));
let too_soon = f.feed(first_at + CLOSE_REPLY_MIN_INTERVAL - NS, &dgram);
assert!(
too_soon.transmits().is_empty(),
"one nanosecond inside the interval is still capped"
);
let dgram = f.peer.seal(&padding(1));
let d = f.feed(first_at + CLOSE_REPLY_MIN_INTERVAL, &dgram);
assert_eq!(
d.transmits().len(),
1,
"§15.2: ≤ 1 per second is a rate — the next second owes another"
);
let close = one_close(&mut f.peer, &d.one_transmit().data);
assert_eq!(close.code, NO_ERROR, "and it repeats our code and reason");
}
#[test]
fn the_linger_reply_goes_to_the_session_address_not_the_packets_source() {
let t = t0();
let mut f = established_at(t);
let _ = f.close(t, NO_ERROR, b"");
let elsewhere = v4(9, 9999);
assert_ne!(elsewhere, f.peer.addr);
let dgram = f.peer.seal(&padding(1));
let d = f.feed_from(t + Duration::from_secs(2), elsewhere, &dgram);
let tx = d.one_transmit();
assert_eq!(tx.to, f.peer.addr, "§15.2: the closing state does not roam");
}
#[test]
fn a_packet_that_fails_the_aead_gets_no_linger_reply() {
let t = t0();
let mut f = established_at(t);
let _ = f.close(t, NO_ERROR, b"");
let mut forged = f.peer.seal(&padding(1));
let last = forged.len() - 1;
forged[last] ^= 0xff;
let d = f.feed(t + Duration::from_secs(2), &forged);
assert!(
d.transmits().is_empty(),
"§15.2: an off-path forger draws nothing, got {:?}",
d.transmits()
);
let genuine = f.peer.seal(&padding(1));
let d = f.feed(t + Duration::from_millis(2_100), &genuine);
assert_eq!(
d.transmits().len(),
1,
"and the reply rule is demonstrably still live"
);
}
#[test]
fn a_replayed_packet_gets_no_linger_reply() {
let t = t0();
let mut f = established_at(t);
let _ = f.close(t, NO_ERROR, b"");
let dgram = f.peer.seal(&padding(1));
let first_at = t + Duration::from_secs(2);
assert_eq!(f.feed(first_at, &dgram).transmits().len(), 1);
let replay_at = first_at + CLOSE_REPLY_MIN_INTERVAL + Duration::from_millis(1);
let d = f.feed(replay_at, &dgram);
assert!(
d.transmits().is_empty(),
"a replay is not window-fresh, got {:?}",
d.transmits()
);
let fresh = f.peer.seal(&padding(1));
let d = f.feed(replay_at + Duration::from_millis(1), &fresh);
assert_eq!(
d.transmits().len(),
1,
"and a genuinely fresh packet at the same moment does draw one"
);
}
#[test]
fn a_close_received_while_closing_stops_the_replies() {
let t = t0();
let mut f = established_at(t);
let _ = f.close(t, NO_ERROR, b"");
let at = t + Duration::from_secs(2);
let d = f.deliver(at, &close_frame(3, b"you too"));
assert!(
d.transmits().is_empty(),
"§15.2: no reply to the peer's CLOSE, got {:?}",
d.transmits()
);
let dgram = f.peer.seal(&padding(1));
let d = f.feed(
at + CLOSE_REPLY_MIN_INTERVAL + Duration::from_millis(1),
&dgram,
);
assert!(
d.transmits().is_empty(),
"§15.2: reply-free from then on, got {:?}",
d.transmits()
);
}
#[test]
fn a_close_received_while_closing_does_not_restart_the_linger() {
let t = t0();
let mut f = established_at(t);
let _ = f.close(t, NO_ERROR, b"");
let at = t + Duration::from_secs(4);
let d = f.deliver(at, &close_frame(3, b"you too"));
assert_eq!(
d.deadline,
Some(t + CLOSE_LINGER),
"Q2: the original deadline stands"
);
let d = f.timeout(t + CLOSE_LINGER);
assert_eq!(
d.retired(),
vec![f.our_index],
"and the state is dropped on the original schedule"
);
}
#[test]
fn closed_is_emitted_exactly_once_across_the_whole_life() {
let t = t0();
let mut f = established_at(t);
let mut seen = f.close(t, NO_ERROR, b"").closed();
seen.extend(
f.deliver(t + Duration::from_secs(1), &close_frame(3, b"too"))
.closed(),
);
seen.extend(f.timeout(t + CLOSE_LINGER).closed());
seen.extend(
f.timeout(t + CLOSE_LINGER + Duration::from_secs(30))
.closed(),
);
assert_eq!(
seen,
vec![ConnectionLost::LocallyClosed],
"Q3: one death, one event, and it is the *first* cause"
);
}
#[test]
fn entering_closing_disarms_the_liveness_timer() {
let t = t0();
let mut f = established_at(t);
let close_at = t + DEAD_TIMEOUT - Duration::from_secs(1);
let d = f.close(close_at, NO_ERROR, b"");
assert_eq!(
d.deadline,
Some(close_at + CLOSE_LINGER),
"the only armed timer is CloseLinger"
);
let d = f.timeout(t + DEAD_TIMEOUT + NS);
assert!(
d.closed().is_empty(),
"no second Closed from a surviving Liveness, got {:?}",
d.outs
);
assert!(d.retired().is_empty(), "and no early Retired");
let d = f.timeout(close_at + CLOSE_LINGER);
assert_eq!(d.retired(), vec![f.our_index], "the linger still expires");
}
#[test]
fn a_violation_surfaces_protocol_violation_not_locally_closed() {
let t = t0();
let mut f = established_at(t);
let d = f.deliver(t, &vi(0x7f));
assert_eq!(
d.closed(),
vec![ConnectionLost::ProtocolViolation {
code: PROTOCOL_VIOLATION
}]
);
assert!(
!d.closed().contains(&ConnectionLost::LocallyClosed),
"§15.2: not LocallyClosed"
);
}
#[test]
fn an_application_code_and_reason_ride_the_close_frame_verbatim() {
for code in [
APPLICATION_ERROR_BASE,
APPLICATION_ERROR_BASE + 7,
(1u64 << 62) - 1,
] {
let t = t0();
let mut f = established_at(t);
let tx = f.close(t, code, b"application reason").one_transmit();
let close = one_close(&mut f.peer, &tx.data);
assert_eq!(close.code, code, "§15.3: the application's code, exactly");
assert_eq!(close.reason, b"application reason");
}
}
#[test]
fn close_never_produces_an_over_length_reason() {
let t = t0();
let mut f = established_at(t);
let exact = vec![b'x'; CLOSE_REASON_MAX];
let tx = f.close(t, NO_ERROR, &exact).one_transmit();
assert_eq!(
one_close(&mut f.peer, &tx.data).reason,
exact,
"exactly CLOSE_REASON_MAX is emitted whole"
);
let t = t0();
let mut f = established_at(t);
let mut over = vec![b'h'; CLOSE_REASON_MAX];
over.push(b'!');
let tx = f.close(t, NO_ERROR, &over).one_transmit();
let close = one_close(&mut f.peer, &tx.data);
assert_eq!(close.reason.len(), CLOSE_REASON_MAX, "§8.4: truncated");
assert_eq!(
close.reason,
over[..CLOSE_REASON_MAX],
"§8.4: truncated at the tail, not the head"
);
}
#[test]
fn a_close_before_the_session_exists_transmits_nothing_and_does_not_linger() {
let t = t0();
let mut conn = connecting();
conn.close(t, NO_ERROR, b"never mind");
let d = drain_bare(&mut conn);
assert_eq!(
d.closed(),
vec![ConnectionLost::LocallyClosed],
"§15.4: the local surface is LocallyClosed"
);
assert!(
d.transmits().is_empty(),
"there is no seal capability, so there is no CLOSE"
);
assert_eq!(
d.deadline, None,
"ruling 81: no linger — there is nothing to linger for"
);
if let (Some(c), Some(r)) = (
d.position(|o| matches!(o, ConnOutput::Event(ConnEvent::Closed(_)))),
d.position(|o| matches!(o, ConnOutput::ToEndpoint(ToEndpoint::Retired { .. }))),
) {
assert!(c < r, "ruling 81: Closed then Retired, in the same drain");
}
}
#[test]
fn an_accepted_connection_closes_the_same_way() {
let t = t0();
let mut f = accepted_at(t);
let d = f.close(t, NO_ERROR, b"bye");
let tx = d.one_transmit();
assert_eq!(one_close(&mut f.peer, &tx.data).reason, b"bye");
assert_eq!(d.closed(), vec![ConnectionLost::LocallyClosed]);
assert_eq!(d.deadline, Some(t + CLOSE_LINGER));
assert!(d.retired().is_empty());
assert_eq!(f.timeout(t + CLOSE_LINGER).retired(), vec![f.our_index]);
}
#[test]
fn a_violation_while_closing_neither_re_reports_nor_extends_the_linger() {
let t = t0();
let mut f = established_at(t);
let _ = f.close(t, NO_ERROR, b"");
let d = f.deliver(t + Duration::from_secs(2), &vi(0x7f));
assert!(
d.closed().is_empty(),
"Q3: Closed was already emitted, got {:?}",
d.outs
);
assert_eq!(
d.deadline,
Some(t + CLOSE_LINGER),
"the linger deadline is unmoved"
);
}
}
mod seal_order {
use super::*;
#[test]
fn close_seals_before_any_poll_output() {
let t = t0();
let mut f = established_at(t);
assert_eq!(f.next_counter(), 0);
f.conn.close(t, 0, b"bye");
assert_eq!(
f.next_counter(),
1,
"§16.7: the CLOSE is sealed inside close(), not inside poll_output()"
);
let d = f.drain();
assert_eq!(d.transmits().len(), 1);
assert_eq!(f.next_counter(), 1, "the drain seals nothing");
}
#[test]
fn a_linger_reply_seals_inside_handle_datagram() {
let t = t0();
let mut f = established_at(t);
let _ = f.close(t, 0, b"");
assert_eq!(f.next_counter(), 1);
let dgram = f.peer.seal(&padding(1));
let addr = f.peer.addr;
f.conn
.handle_datagram(t + Duration::from_secs(2), addr, &dgram);
assert_eq!(
f.next_counter(),
2,
"§16.7: the reply is sealed inside handle_datagram"
);
let d = f.drain();
assert_eq!(d.transmits().len(), 1);
assert_eq!(f.next_counter(), 2);
}
#[test]
fn a_call_that_sends_nothing_burns_no_counter() {
let t = t0();
let mut f = established_at(t);
let dgram = f.peer.seal(&padding(1));
let _ = f.feed(t, &dgram);
assert_eq!(f.next_counter(), 0, "a receive alone seals nothing");
let _ = f.close(t, 0, b"");
assert_eq!(f.next_counter(), 1);
let dgram = f.peer.seal(&padding(1));
let d = f.feed(t + Duration::from_millis(1), &dgram);
assert_eq!(d.transmits().len(), 1, "the first reply is not capped");
assert_eq!(f.next_counter(), 2, "and it burned exactly one counter");
let dgram = f.peer.seal(&padding(1));
let d = f.feed(t + Duration::from_millis(2), &dgram);
assert!(d.transmits().is_empty(), "capped by §15.2's 1 Hz rule");
assert_eq!(f.next_counter(), 2, "… and nothing is sealed");
}
}
mod ack_eliciting_classifier {
use super::*;
use crate::core::connection::frame::is_ack_eliciting;
#[test]
fn every_row_of_the_frame_table_is_classified() {
let expected: &[(u64, bool)] = &[
(FRAME_PADDING, false),
(FRAME_PING, true),
(FRAME_ACK, false),
(FRAME_RESET_STREAM, true),
(FRAME_MAX_DATA, true),
(FRAME_MAX_STREAM_DATA, true),
(FRAME_MAX_STREAMS_BIDI, true),
(FRAME_MAX_STREAMS_UNI, true),
(FRAME_CLOSE, false),
(FRAME_DATAGRAM, true),
(FRAME_DATAGRAM_LEN, true),
];
for (ty, want) in expected {
assert_eq!(
is_ack_eliciting(*ty),
*want,
"§8.3: type {ty:#x} is ack-eliciting = {want}"
);
}
for ty in FRAME_STREAM_BASE..=FRAME_STREAM_MAX {
assert!(
is_ack_eliciting(ty),
"§8.3: every STREAM type {ty:#x} is ack-eliciting"
);
}
}
#[test]
fn ack_eliciting_is_not_the_complement_of_the_quiet_set() {
assert!(is_ack_eliciting(FRAME_MAX_DATA));
assert!(is_ack_eliciting(FRAME_RESET_STREAM));
assert!(!is_ack_eliciting(FRAME_ACK));
assert!(!is_ack_eliciting(FRAME_CLOSE));
}
}
mod timer_table {
use super::*;
use crate::core::connection::timers::{TimerKind, Timers};
fn clear(t: &mut Timers, k: TimerKind) {
t.disarm(k);
}
fn armed(pairs: &[(TimerKind, Instant)]) -> Timers {
let mut t = Timers::default();
for (k, at) in pairs {
t.arm(*k, *at);
}
t
}
fn first_due(t: &Timers, now: Instant) -> Option<TimerKind> {
t.due(now).iter().next()
}
#[test]
fn next_is_the_minimum_over_the_armed_timers_and_none_when_bare() {
let t = t0();
let mut timers = Timers::default();
assert_eq!(timers.next(), None, "nothing armed, no deadline");
assert_eq!(first_due(&timers, t), None);
timers.set(TimerKind::CloseLinger, Some(t + Duration::from_secs(5)));
timers.set(TimerKind::Liveness, Some(t + Duration::from_secs(25)));
assert_eq!(
timers.next(),
Some(t + Duration::from_secs(5)),
"the minimum, not the first field"
);
timers.set(TimerKind::Loss, Some(t + Duration::from_secs(1)));
assert_eq!(timers.next(), Some(t + Duration::from_secs(1)));
assert_eq!(
first_due(&timers, t),
None,
"nothing is due before its deadline"
);
assert_eq!(
first_due(&timers, t + Duration::from_secs(1)),
Some(TimerKind::Loss)
);
}
#[test]
fn the_ratified_equal_deadline_order_is_a_total_order() {
let t = t0();
let mut timers = armed(&[
(TimerKind::Liveness, t),
(TimerKind::CloseLinger, t),
(TimerKind::Contested, t),
(TimerKind::Loss, t),
(TimerKind::AckDelay, t),
(TimerKind::Keepalive, t),
(TimerKind::PersistentKeepalive, t),
]);
let mut order = vec![];
while let Some(k) = first_due(&timers, t) {
order.push(k);
clear(&mut timers, k);
}
assert_eq!(
order,
vec![
TimerKind::Liveness,
TimerKind::CloseLinger,
TimerKind::Contested,
TimerKind::Loss,
TimerKind::AckDelay,
TimerKind::Keepalive,
TimerKind::PersistentKeepalive,
],
"§16.5 (ruling 76): a terminal outcome precedes a routine one"
);
}
#[test]
fn loss_beats_pto_at_the_same_instant() {
let t = t0();
let timers = armed(&[(TimerKind::Loss, t), (TimerKind::Pto, t)]);
assert_eq!(first_due(&timers, t), Some(TimerKind::Loss));
}
#[test]
fn liveness_beats_contested_and_close_linger() {
let t = t0();
let timers = armed(&[
(TimerKind::Liveness, t),
(TimerKind::Contested, t),
(TimerKind::CloseLinger, t),
]);
assert_eq!(first_due(&timers, t), Some(TimerKind::Liveness));
}
#[test]
fn teardown_collection_precedes_keepalive_evaluation() {
let t = t0();
let timers = armed(&[
(TimerKind::CloseLinger, t),
(TimerKind::Keepalive, t),
(TimerKind::PersistentKeepalive, t),
]);
assert_eq!(first_due(&timers, t), Some(TimerKind::CloseLinger));
}
#[test]
fn ack_delay_fires_after_the_loss_evaluation() {
let t = t0();
let timers = armed(&[(TimerKind::AckDelay, t), (TimerKind::Loss, t)]);
assert_eq!(first_due(&timers, t), Some(TimerKind::Loss));
let timers = armed(&[(TimerKind::AckDelay, t), (TimerKind::Pto, t)]);
assert_eq!(first_due(&timers, t), Some(TimerKind::Pto));
}
#[test]
fn priority_breaks_ties_and_does_not_reorder_distinct_deadlines() {
let t = t0();
let timers = armed(&[
(TimerKind::Liveness, t + Duration::from_secs(10)),
(TimerKind::PersistentKeepalive, t),
]);
assert_eq!(timers.next(), Some(t));
assert_eq!(
first_due(&timers, t),
Some(TimerKind::PersistentKeepalive),
"the lowest-priority timer is still the only due one"
);
}
}
mod replay_window_api {
use super::*;
use crate::core::connection::session::ReplayWindow;
#[test]
fn ranges_are_descending_newest_first_and_merged() {
let mut w = ReplayWindow::default();
assert_eq!(w.greatest(), None, "a fresh window has seen nothing");
for c in [0u64, 1, 2, 5, 9] {
assert!(w.check_and_mark(c), "counter {c} is fresh");
}
assert_eq!(w.greatest(), Some(9));
let ranges: Vec<RangeInclusive<u64>> = w.ranges_desc().collect();
assert_eq!(
ranges,
vec![9..=9, 5..=5, 0..=2],
"§12.2: newest-first, descending, runs merged"
);
}
#[test]
fn a_counter_is_fresh_exactly_once() {
let mut w = ReplayWindow::default();
assert!(w.check_and_mark(7));
assert!(!w.check_and_mark(7), "§7.2: a duplicate is not fresh");
assert!(w.check_and_mark(3), "an older, unseen counter is fresh");
assert_eq!(w.greatest(), Some(7), "and does not move `greatest` back");
}
#[test]
fn the_lower_edge_is_the_window_width_exactly() {
let window = REPLAY_WINDOW as u64;
let mut w = ReplayWindow::default();
let greatest = 1_000_000u64;
assert!(w.check_and_mark(greatest));
assert!(
w.check_and_mark(greatest - window),
"§7.2: exactly {window} behind is not *more than* {window} behind"
);
assert!(
!w.check_and_mark(greatest - window - 1),
"§7.2: one further back is dropped"
);
}
}
#[allow(dead_code)]
mod owed {
pub const ESTABLISHED_HAS_NO_INSTANT: () = ();
pub const LIVENESS_EXACT_INSTANT: () = ();
pub const CLOSE_RATE_CLOCK_ORIGIN: () = ();
pub const RETIRED_WITHOUT_A_SESSION: () = ();
pub const VIOLATION_WHILE_CLOSING: () = ();
pub const SEAL_VERSUS_SEAL_QUIET: () = ();
pub const PACKING_ORDER: () = ();
pub const OVERSIZE_DATA_PACKET: () = ();
}