use super::ProtocolError;
use crate::NodeAddr;
use crate::tree::TreeCoordinate;
use std::fmt;
mod control_messages;
pub use control_messages::{
COORDS_REQUIRED_SIZE, CoordsRequired, MTU_EXCEEDED_SIZE, MtuExceeded,
PATH_MTU_NOTIFICATION_SIZE, PathBroken, PathMtuNotification,
};
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[repr(u8)]
pub enum SessionMessageType {
DataPacket = 0x10,
SenderReport = 0x11,
ReceiverReport = 0x12,
PathMtuNotification = 0x13,
CoordsWarmup = 0x14,
EndpointData = 0x15,
TraversalOffer = 0x16,
TraversalAnswer = 0x17,
CoordsRequired = 0x20,
PathBroken = 0x21,
MtuExceeded = 0x22,
}
impl SessionMessageType {
pub fn from_byte(b: u8) -> Option<Self> {
match b {
0x10 => Some(SessionMessageType::DataPacket),
0x11 => Some(SessionMessageType::SenderReport),
0x12 => Some(SessionMessageType::ReceiverReport),
0x13 => Some(SessionMessageType::PathMtuNotification),
0x14 => Some(SessionMessageType::CoordsWarmup),
0x15 => Some(SessionMessageType::EndpointData),
0x16 => Some(SessionMessageType::TraversalOffer),
0x17 => Some(SessionMessageType::TraversalAnswer),
0x20 => Some(SessionMessageType::CoordsRequired),
0x21 => Some(SessionMessageType::PathBroken),
0x22 => Some(SessionMessageType::MtuExceeded),
_ => None,
}
}
pub fn to_byte(self) -> u8 {
self as u8
}
}
impl fmt::Display for SessionMessageType {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let name = match self {
SessionMessageType::DataPacket => "DataPacket",
SessionMessageType::SenderReport => "SenderReport",
SessionMessageType::ReceiverReport => "ReceiverReport",
SessionMessageType::PathMtuNotification => "PathMtuNotification",
SessionMessageType::CoordsWarmup => "CoordsWarmup",
SessionMessageType::EndpointData => "EndpointData",
SessionMessageType::TraversalOffer => "TraversalOffer",
SessionMessageType::TraversalAnswer => "TraversalAnswer",
SessionMessageType::CoordsRequired => "CoordsRequired",
SessionMessageType::PathBroken => "PathBroken",
SessionMessageType::MtuExceeded => "MtuExceeded",
};
write!(f, "{}", name)
}
}
pub(crate) fn coords_wire_size(coords: &TreeCoordinate) -> usize {
2 + coords.entries().len() * 16
}
pub(crate) fn encode_coords(coords: &TreeCoordinate, buf: &mut Vec<u8>) {
let addrs: Vec<&NodeAddr> = coords.node_addrs().collect();
let count = addrs.len() as u16;
buf.extend_from_slice(&count.to_le_bytes());
for addr in addrs {
buf.extend_from_slice(addr.as_bytes());
}
}
pub(crate) fn decode_coords(data: &[u8]) -> Result<(TreeCoordinate, usize), ProtocolError> {
if data.len() < 2 {
return Err(ProtocolError::MessageTooShort {
expected: 2,
got: data.len(),
});
}
let count = u16::from_le_bytes([data[0], data[1]]) as usize;
let needed = 2 + count * 16;
if data.len() < needed {
return Err(ProtocolError::MessageTooShort {
expected: needed,
got: data.len(),
});
}
if count == 0 {
return Err(ProtocolError::Malformed(
"coordinate with zero entries".into(),
));
}
let mut addrs = Vec::with_capacity(count);
for i in 0..count {
let offset = 2 + i * 16;
let mut bytes = [0u8; 16];
bytes.copy_from_slice(&data[offset..offset + 16]);
addrs.push(NodeAddr::from_bytes(bytes));
}
let coord =
TreeCoordinate::from_addrs(addrs).map_err(|e| ProtocolError::Malformed(e.to_string()))?;
Ok((coord, needed))
}
pub(crate) fn decode_optional_coords(
data: &[u8],
) -> Result<(Option<TreeCoordinate>, usize), ProtocolError> {
if data.len() < 2 {
return Err(ProtocolError::MessageTooShort {
expected: 2,
got: data.len(),
});
}
let count = u16::from_le_bytes([data[0], data[1]]) as usize;
let needed = 2 + count * 16;
if data.len() < needed {
return Err(ProtocolError::MessageTooShort {
expected: needed,
got: data.len(),
});
}
if count == 0 {
return Ok((None, 2));
}
let mut addrs = Vec::with_capacity(count);
for i in 0..count {
let offset = 2 + i * 16;
let mut bytes = [0u8; 16];
bytes.copy_from_slice(&data[offset..offset + 16]);
addrs.push(NodeAddr::from_bytes(bytes));
}
let coord =
TreeCoordinate::from_addrs(addrs).map_err(|e| ProtocolError::Malformed(e.to_string()))?;
Ok((Some(coord), needed))
}
fn encode_empty_coords(buf: &mut Vec<u8>) {
buf.extend_from_slice(&0u16.to_le_bytes());
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct SessionFlags {
pub request_ack: bool,
pub bidirectional: bool,
}
impl SessionFlags {
pub fn new() -> Self {
Self::default()
}
pub fn with_ack(mut self) -> Self {
self.request_ack = true;
self
}
pub fn bidirectional(mut self) -> Self {
self.bidirectional = true;
self
}
pub fn to_byte(&self) -> u8 {
let mut flags = 0u8;
if self.request_ack {
flags |= 0x01;
}
if self.bidirectional {
flags |= 0x02;
}
flags
}
pub fn from_byte(byte: u8) -> Self {
Self {
request_ack: byte & 0x01 != 0,
bidirectional: byte & 0x02 != 0,
}
}
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct FspFlags {
pub coords_present: bool,
pub key_epoch: bool,
pub unencrypted: bool,
}
impl FspFlags {
pub fn new() -> Self {
Self::default()
}
pub fn to_byte(&self) -> u8 {
let mut flags = 0u8;
if self.coords_present {
flags |= 0x01;
}
if self.key_epoch {
flags |= 0x02;
}
if self.unencrypted {
flags |= 0x04;
}
flags
}
pub fn from_byte(byte: u8) -> Self {
Self {
coords_present: byte & 0x01 != 0,
key_epoch: byte & 0x02 != 0,
unencrypted: byte & 0x04 != 0,
}
}
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct FspInnerFlags {
pub spin_bit: bool,
}
impl FspInnerFlags {
pub fn new() -> Self {
Self::default()
}
pub fn to_byte(&self) -> u8 {
if self.spin_bit { 0x01 } else { 0x00 }
}
pub fn from_byte(byte: u8) -> Self {
Self {
spin_bit: byte & 0x01 != 0,
}
}
}
#[derive(Clone, Debug)]
pub struct SessionSetup {
pub src_coords: TreeCoordinate,
pub dest_coords: TreeCoordinate,
pub flags: SessionFlags,
pub handshake_payload: Vec<u8>,
}
impl SessionSetup {
pub fn new(src_coords: TreeCoordinate, dest_coords: TreeCoordinate) -> Self {
Self {
src_coords,
dest_coords,
flags: SessionFlags::new(),
handshake_payload: Vec::new(),
}
}
pub fn with_flags(mut self, flags: SessionFlags) -> Self {
self.flags = flags;
self
}
pub fn with_handshake(mut self, payload: Vec<u8>) -> Self {
self.handshake_payload = payload;
self
}
pub fn encode(&self) -> Vec<u8> {
let mut body = Vec::new();
body.push(self.flags.to_byte());
encode_coords(&self.src_coords, &mut body);
encode_coords(&self.dest_coords, &mut body);
let hs_len = self.handshake_payload.len() as u16;
body.extend_from_slice(&hs_len.to_le_bytes());
body.extend_from_slice(&self.handshake_payload);
let payload_len = body.len() as u16;
let mut buf = Vec::with_capacity(4 + body.len());
buf.push(0x01); buf.push(0x00); buf.extend_from_slice(&payload_len.to_le_bytes());
buf.extend_from_slice(&body);
buf
}
pub fn decode(payload: &[u8]) -> Result<Self, ProtocolError> {
if payload.is_empty() {
return Err(ProtocolError::MessageTooShort {
expected: 1,
got: 0,
});
}
let flags = SessionFlags::from_byte(payload[0]);
let mut offset = 1;
let (src_coords, consumed) = decode_coords(&payload[offset..])?;
offset += consumed;
let (dest_coords, consumed) = decode_coords(&payload[offset..])?;
offset += consumed;
if payload.len() < offset + 2 {
return Err(ProtocolError::MessageTooShort {
expected: offset + 2,
got: payload.len(),
});
}
let hs_len = u16::from_le_bytes([payload[offset], payload[offset + 1]]) as usize;
offset += 2;
if payload.len() < offset + hs_len {
return Err(ProtocolError::MessageTooShort {
expected: offset + hs_len,
got: payload.len(),
});
}
let handshake_payload = payload[offset..offset + hs_len].to_vec();
Ok(Self {
src_coords,
dest_coords,
flags,
handshake_payload,
})
}
}
#[derive(Clone, Debug)]
pub struct SessionAck {
pub src_coords: TreeCoordinate,
pub dest_coords: TreeCoordinate,
pub flags: u8,
pub handshake_payload: Vec<u8>,
}
impl SessionAck {
pub fn new(src_coords: TreeCoordinate, dest_coords: TreeCoordinate) -> Self {
Self {
src_coords,
dest_coords,
flags: 0,
handshake_payload: Vec::new(),
}
}
pub fn with_handshake(mut self, payload: Vec<u8>) -> Self {
self.handshake_payload = payload;
self
}
pub fn encode(&self) -> Vec<u8> {
let mut body = Vec::new();
body.push(self.flags);
encode_coords(&self.src_coords, &mut body);
encode_coords(&self.dest_coords, &mut body);
let hs_len = self.handshake_payload.len() as u16;
body.extend_from_slice(&hs_len.to_le_bytes());
body.extend_from_slice(&self.handshake_payload);
let payload_len = body.len() as u16;
let mut buf = Vec::with_capacity(4 + body.len());
buf.push(0x02); buf.push(0x00); buf.extend_from_slice(&payload_len.to_le_bytes());
buf.extend_from_slice(&body);
buf
}
pub fn decode(payload: &[u8]) -> Result<Self, ProtocolError> {
if payload.is_empty() {
return Err(ProtocolError::MessageTooShort {
expected: 1,
got: 0,
});
}
let flags = payload[0];
let mut offset = 1;
let (src_coords, consumed) = decode_coords(&payload[offset..])?;
offset += consumed;
let (dest_coords, consumed) = decode_coords(&payload[offset..])?;
offset += consumed;
if payload.len() < offset + 2 {
return Err(ProtocolError::MessageTooShort {
expected: offset + 2,
got: payload.len(),
});
}
let hs_len = u16::from_le_bytes([payload[offset], payload[offset + 1]]) as usize;
offset += 2;
if payload.len() < offset + hs_len {
return Err(ProtocolError::MessageTooShort {
expected: offset + hs_len,
got: payload.len(),
});
}
let handshake_payload = payload[offset..offset + hs_len].to_vec();
Ok(Self {
src_coords,
dest_coords,
flags,
handshake_payload,
})
}
}
#[derive(Clone, Debug)]
pub struct SessionMsg3 {
pub flags: u8,
pub handshake_payload: Vec<u8>,
}
impl SessionMsg3 {
pub fn new(handshake_payload: Vec<u8>) -> Self {
Self {
flags: 0,
handshake_payload,
}
}
pub fn encode(&self) -> Vec<u8> {
let mut body = Vec::new();
body.push(self.flags);
let hs_len = self.handshake_payload.len() as u16;
body.extend_from_slice(&hs_len.to_le_bytes());
body.extend_from_slice(&self.handshake_payload);
let payload_len = body.len() as u16;
let mut buf = Vec::with_capacity(4 + body.len());
buf.push(0x03); buf.push(0x00); buf.extend_from_slice(&payload_len.to_le_bytes());
buf.extend_from_slice(&body);
buf
}
pub fn decode(payload: &[u8]) -> Result<Self, ProtocolError> {
if payload.is_empty() {
return Err(ProtocolError::MessageTooShort {
expected: 1,
got: 0,
});
}
let flags = payload[0];
let mut offset = 1;
if payload.len() < offset + 2 {
return Err(ProtocolError::MessageTooShort {
expected: offset + 2,
got: payload.len(),
});
}
let hs_len = u16::from_le_bytes([payload[offset], payload[offset + 1]]) as usize;
offset += 2;
if payload.len() < offset + hs_len {
return Err(ProtocolError::MessageTooShort {
expected: offset + hs_len,
got: payload.len(),
});
}
let handshake_payload = payload[offset..offset + hs_len].to_vec();
Ok(Self {
flags,
handshake_payload,
})
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SessionSenderReport {
pub interval_start_counter: u64,
pub interval_end_counter: u64,
pub interval_start_timestamp: u32,
pub interval_end_timestamp: u32,
pub interval_bytes_sent: u32,
pub cumulative_packets_sent: u64,
pub cumulative_bytes_sent: u64,
}
pub const SESSION_SENDER_REPORT_SIZE: usize = 46;
impl SessionSenderReport {
pub fn encode(&self) -> Vec<u8> {
let mut buf = Vec::with_capacity(SESSION_SENDER_REPORT_SIZE);
buf.extend_from_slice(&[0u8; 2]); buf.extend_from_slice(&self.interval_start_counter.to_le_bytes());
buf.extend_from_slice(&self.interval_end_counter.to_le_bytes());
buf.extend_from_slice(&self.interval_start_timestamp.to_le_bytes());
buf.extend_from_slice(&self.interval_end_timestamp.to_le_bytes());
buf.extend_from_slice(&self.interval_bytes_sent.to_le_bytes());
buf.extend_from_slice(&self.cumulative_packets_sent.to_le_bytes());
buf.extend_from_slice(&self.cumulative_bytes_sent.to_le_bytes());
buf
}
pub fn decode(body: &[u8]) -> Result<Self, ProtocolError> {
if body.len() < SESSION_SENDER_REPORT_SIZE {
return Err(ProtocolError::MessageTooShort {
expected: SESSION_SENDER_REPORT_SIZE,
got: body.len(),
});
}
let p = &body[2..];
Ok(Self {
interval_start_counter: u64::from_le_bytes(p[0..8].try_into().unwrap()),
interval_end_counter: u64::from_le_bytes(p[8..16].try_into().unwrap()),
interval_start_timestamp: u32::from_le_bytes(p[16..20].try_into().unwrap()),
interval_end_timestamp: u32::from_le_bytes(p[20..24].try_into().unwrap()),
interval_bytes_sent: u32::from_le_bytes(p[24..28].try_into().unwrap()),
cumulative_packets_sent: u64::from_le_bytes(p[28..36].try_into().unwrap()),
cumulative_bytes_sent: u64::from_le_bytes(p[36..44].try_into().unwrap()),
})
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SessionReceiverReport {
pub highest_counter: u64,
pub cumulative_packets_recv: u64,
pub cumulative_bytes_recv: u64,
pub timestamp_echo: u32,
pub dwell_time: u16,
pub max_burst_loss: u16,
pub mean_burst_loss: u16,
pub jitter: u32,
pub ecn_ce_count: u32,
pub owd_trend: i32,
pub burst_loss_count: u32,
pub cumulative_reorder_count: u32,
pub interval_packets_recv: u32,
pub interval_bytes_recv: u32,
}
pub const SESSION_RECEIVER_REPORT_SIZE: usize = 66;
impl SessionReceiverReport {
pub fn encode(&self) -> Vec<u8> {
let mut buf = Vec::with_capacity(SESSION_RECEIVER_REPORT_SIZE);
buf.extend_from_slice(&[0u8; 2]); buf.extend_from_slice(&self.highest_counter.to_le_bytes());
buf.extend_from_slice(&self.cumulative_packets_recv.to_le_bytes());
buf.extend_from_slice(&self.cumulative_bytes_recv.to_le_bytes());
buf.extend_from_slice(&self.timestamp_echo.to_le_bytes());
buf.extend_from_slice(&self.dwell_time.to_le_bytes());
buf.extend_from_slice(&self.max_burst_loss.to_le_bytes());
buf.extend_from_slice(&self.mean_burst_loss.to_le_bytes());
buf.extend_from_slice(&[0u8; 2]); buf.extend_from_slice(&self.jitter.to_le_bytes());
buf.extend_from_slice(&self.ecn_ce_count.to_le_bytes());
buf.extend_from_slice(&self.owd_trend.to_le_bytes());
buf.extend_from_slice(&self.burst_loss_count.to_le_bytes());
buf.extend_from_slice(&self.cumulative_reorder_count.to_le_bytes());
buf.extend_from_slice(&self.interval_packets_recv.to_le_bytes());
buf.extend_from_slice(&self.interval_bytes_recv.to_le_bytes());
buf
}
pub fn decode(body: &[u8]) -> Result<Self, ProtocolError> {
if body.len() < SESSION_RECEIVER_REPORT_SIZE {
return Err(ProtocolError::MessageTooShort {
expected: SESSION_RECEIVER_REPORT_SIZE,
got: body.len(),
});
}
let p = &body[2..];
Ok(Self {
highest_counter: u64::from_le_bytes(p[0..8].try_into().unwrap()),
cumulative_packets_recv: u64::from_le_bytes(p[8..16].try_into().unwrap()),
cumulative_bytes_recv: u64::from_le_bytes(p[16..24].try_into().unwrap()),
timestamp_echo: u32::from_le_bytes(p[24..28].try_into().unwrap()),
dwell_time: u16::from_le_bytes(p[28..30].try_into().unwrap()),
max_burst_loss: u16::from_le_bytes(p[30..32].try_into().unwrap()),
mean_burst_loss: u16::from_le_bytes(p[32..34].try_into().unwrap()),
jitter: u32::from_le_bytes(p[36..40].try_into().unwrap()),
ecn_ce_count: u32::from_le_bytes(p[40..44].try_into().unwrap()),
owd_trend: i32::from_le_bytes(p[44..48].try_into().unwrap()),
burst_loss_count: u32::from_le_bytes(p[48..52].try_into().unwrap()),
cumulative_reorder_count: u32::from_le_bytes(p[52..56].try_into().unwrap()),
interval_packets_recv: u32::from_le_bytes(p[56..60].try_into().unwrap()),
interval_bytes_recv: u32::from_le_bytes(p[60..64].try_into().unwrap()),
})
}
}
#[cfg(test)]
mod tests;