mod loss_compression;
mod srt;
pub use srt::*;
use std::{
array::IntoIter,
convert::TryFrom,
convert::TryInto,
fmt::{self, Debug, Display, Formatter},
iter::FromIterator,
mem::size_of,
net::{IpAddr, Ipv4Addr, Ipv6Addr},
ops::{Add, Range, RangeInclusive, Sub},
};
use bitflags::bitflags;
use bytes::{Buf, BufMut};
use log::warn;
use crate::{
options::{PacketCount, PacketSize},
protocol::time::Rtt,
};
use super::*;
use loss_compression::{compress_loss_list, decompress_loss_list};
#[derive(Clone, PartialEq, Eq)]
pub struct ControlPacket {
pub timestamp: TimeStamp,
pub dest_sockid: SocketId,
pub control_type: ControlTypes,
}
impl ControlPacket {
pub const HEADER_SIZE: usize = super::Packet::HEADER_SIZE.0 as usize;
}
#[derive(Clone, PartialEq, Eq)]
#[allow(clippy::large_enum_variant)]
pub enum ControlTypes {
Handshake(HandshakeControlInfo),
KeepAlive,
Ack(Acknowledgement),
Nak(CompressedLossList),
CongestionWarning,
Shutdown,
Ack2(FullAckSeqNumber),
DropRequest {
msg_to_drop: MsgNumber,
range: RangeInclusive<SeqNumber>,
},
PeerError(u32),
Srt(SrtControlPacket),
}
bitflags! {
struct ExtFlags: u16 {
const HS = 0b1;
const KM = 0b10;
const CONFIG = 0b100;
}
}
#[derive(Clone, PartialEq, Eq, Debug, Default)]
pub struct HsV5Info {
pub crypto_size: u8,
pub ext_hs: Option<SrtControlPacket>,
pub ext_km: Option<SrtControlPacket>,
pub ext_group: Option<SrtControlPacket>,
pub sid: Option<String>,
}
#[derive(Clone, PartialEq, Eq)]
#[allow(clippy::large_enum_variant)]
pub enum HandshakeVsInfo {
V4(SocketType),
V5(HsV5Info),
}
#[derive(Clone, PartialEq, Eq)]
pub struct HandshakeControlInfo {
pub init_seq_num: SeqNumber,
pub max_packet_size: PacketSize,
pub max_flow_size: PacketCount,
pub shake_type: ShakeType,
pub socket_id: SocketId,
pub syn_cookie: i32,
pub peer_addr: IpAddr,
pub info: HandshakeVsInfo,
}
#[derive(Clone, PartialEq, Eq, Debug)]
pub struct AckStatistics {
pub rtt: Rtt,
pub buffer_available: u32,
pub packet_receive_rate: Option<u32>,
pub estimated_link_capacity: Option<u32>,
pub data_receive_rate: Option<u32>,
}
#[derive(Clone, PartialEq, Eq, Debug, Copy, Ord, PartialOrd)]
pub struct FullAckSeqNumber(u32);
#[derive(Clone, PartialEq, Eq, Debug)]
pub enum Acknowledgement {
Lite(SeqNumber),
Small(SeqNumber, AckStatistics),
Full(SeqNumber, AckStatistics, FullAckSeqNumber),
}
#[derive(Clone, Eq, PartialEq)]
pub struct CompressedLossList(Vec<u32>);
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SocketType {
Stream = 1,
Datagram = 2,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ShakeType {
Induction,
Waveahand,
Conclusion,
Agreement,
Rejection(RejectReason),
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum CoreRejectReason {
System = 1001,
Peer = 1002,
Resource = 1003,
Rogue = 1004,
Backlog = 1005,
Ipe = 1006,
Close = 1007,
Version = 1008,
RdvCookie = 1009,
BadSecret = 1010,
Unsecure = 1011,
MessageApi = 1012,
Congestion = 1013,
Filter = 1014,
Group = 1015,
Timeout = 1016,
}
#[non_exhaustive]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ServerRejectReason {
Fallback = 2000,
KeyNotSup = 2001,
Filepath = 2002,
HostNotFound = 2003,
BadRequest = 2400,
Unauthorized = 2401,
Overload = 2402,
Forbidden = 2403,
Notfound = 2404,
BadMode = 2405,
Unacceptable = 2406,
Conflict = 2409,
NotSupMedia = 2415,
Locked = 2423,
FailedDepend = 2424,
InternalServerError = 2500,
Unimplemented = 2501,
Gateway = 2502,
Down = 2503,
Version = 2505,
NoRoom = 2507,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RejectReason {
Core(CoreRejectReason),
CoreUnrecognized(i32),
Server(ServerRejectReason),
ServerUnrecognized(i32),
User(i32),
}
impl HandshakeVsInfo {
fn type_flags(&self, shake_type: ShakeType) -> u32 {
match self {
HandshakeVsInfo::V4(ty) => *ty as u32,
HandshakeVsInfo::V5(hs) => {
if shake_type == ShakeType::Induction
&& (hs.ext_hs.is_some() || hs.ext_km.is_some() || hs.sid.is_some())
{
panic!("Handshake is both induction and has SRT extensions, not valid");
}
let mut flags = ExtFlags::empty();
if hs.ext_hs.is_some() {
flags |= ExtFlags::HS;
}
if hs.ext_km.is_some() {
flags |= ExtFlags::KM;
}
if hs.sid.is_some() {
flags |= ExtFlags::CONFIG;
}
(u32::from(hs.crypto_size) >> 3 << 16)
| if shake_type == ShakeType::Induction {
u32::from(SRT_MAGIC_CODE)
} else {
u32::from(flags.bits())
}
}
}
}
pub fn version(&self) -> u32 {
match self {
HandshakeVsInfo::V4(_) => 4,
HandshakeVsInfo::V5 { .. } => 5,
}
}
}
impl SocketType {
pub fn from_u16(num: u16) -> Result<SocketType, u16> {
match num {
1 => Ok(SocketType::Stream),
2 => Ok(SocketType::Datagram),
i => Err(i),
}
}
}
impl ControlPacket {
pub fn parse(buf: &mut impl Buf, is_ipv6: bool) -> Result<ControlPacket, PacketParseError> {
let control_type = buf.get_u16() << 1 >> 1;
let reserved = buf.get_u16();
let add_info = buf.get_u32();
let timestamp = TimeStamp::from_micros(buf.get_u32());
let dest_sockid = buf.get_u32();
Ok(ControlPacket {
timestamp,
dest_sockid: SocketId(dest_sockid),
control_type: ControlTypes::deserialize(
control_type,
reserved,
add_info,
buf,
is_ipv6,
)?,
})
}
pub fn serialize<T: BufMut>(&self, into: &mut T) {
into.put_u16(self.control_type.id_byte() | (0b1 << 15));
into.put_u16(self.control_type.reserved());
into.put_u32(self.control_type.additional_info());
into.put_u32(self.timestamp.as_micros());
into.put_u32(self.dest_sockid.0);
self.control_type.serialize(into);
}
pub fn handshake(&self) -> Option<&HandshakeControlInfo> {
if let ControlTypes::Handshake(hs) = &self.control_type {
Some(hs)
} else {
None
}
}
pub fn wire_size(&self) -> usize {
Self::HEADER_SIZE
+ match &self.control_type {
ControlTypes::Handshake(hs) => hs.serialized_size(),
ControlTypes::Ack(ack) => ack.serialized_size(),
ControlTypes::Nak(nak) => nak.0.len() * size_of::<u32>(),
ControlTypes::DropRequest { .. } => 2 * size_of::<u32>(),
ControlTypes::Srt(srt) => usize::from(srt.size_words()) * size_of::<u32>(),
ControlTypes::CongestionWarning
| ControlTypes::Ack2(_)
| ControlTypes::Shutdown
| ControlTypes::KeepAlive
| ControlTypes::PeerError(_) => 4,
}
}
}
impl Debug for ControlPacket {
fn fmt(&self, f: &mut Formatter) -> Result<(), fmt::Error> {
write!(
f,
"{{{:?} ts={:?} dst={:?}}}",
self.control_type, self.timestamp, self.dest_sockid,
)
}
}
const SRT_MAGIC_CODE: u16 = 0x4A17;
impl ControlTypes {
pub fn new_drop_request(msg_to_drop: MsgNumber, drop_range: Range<SeqNumber>) -> Self {
assert!(!drop_range.is_empty());
Self::DropRequest {
msg_to_drop,
range: RangeInclusive::new(drop_range.start, drop_range.end - 1),
}
}
pub fn new_key_refresh_request(key_material: KeyingMaterialMessage) -> ControlTypes {
ControlTypes::Srt(SrtControlPacket::KeyRefreshRequest(key_material))
}
fn deserialize<T: Buf>(
packet_type: u16,
reserved: u16,
extra_info: u32,
mut buf: T,
is_ipv6: bool,
) -> Result<ControlTypes, PacketParseError> {
match packet_type {
0x0 => {
if buf.remaining() < 8 * 4 + 16 {
return Err(PacketParseError::NotEnoughData);
}
let udt_version = buf.get_i32();
if udt_version != 4 && udt_version != 5 {
return Err(PacketParseError::BadUdtVersion(udt_version));
}
let crypto_size = buf.get_u16() << 3;
let type_ext_socket_type = buf.get_u16();
let init_seq_num = SeqNumber::new_truncate(buf.get_u32()); let max_packet_size = PacketSize(buf.get_u32() as u64);
let max_flow_size = PacketCount(buf.get_u32() as u64);
let shake_type = match ShakeType::try_from(buf.get_i32()) {
Ok(ct) => ct,
Err(err_ct) => return Err(PacketParseError::BadConnectionType(err_ct)),
};
let socket_id = SocketId(buf.get_u32());
let syn_cookie = buf.get_i32();
let peer_addr = if !is_ipv6 {
let ip = buf.get_u32_le();
buf.get_u32();
buf.get_u32();
buf.get_u32();
IpAddr::from(Ipv4Addr::from(ip))
} else {
let mut ip_buf = [0u8; 16];
buf.copy_to_slice(&mut ip_buf);
IpAddr::from(Ipv6Addr::from(ip_buf))
};
let info = match udt_version {
4 => HandshakeVsInfo::V4(match SocketType::from_u16(type_ext_socket_type) {
Ok(t) => t,
Err(e) => return Err(PacketParseError::BadSocketType(e)),
}),
5 => {
let crypto_size = match crypto_size {
0 | 16 | 24 | 32 => crypto_size as u8,
c => {
warn!(
"Unrecognized crypto key length: {}, disabling encryption. Should be 0, 16, 24, or 32 bytes.",
c
);
0
}
};
if shake_type == ShakeType::Induction {
if type_ext_socket_type != SRT_MAGIC_CODE {
warn!("HSv5 induction response did not have SRT_MAGIC_CODE, which is suspicious")
}
HandshakeVsInfo::V5(HsV5Info::default())
} else {
let extensions = match ExtFlags::from_bits(type_ext_socket_type) {
Some(i) => i,
None => {
warn!(
"Unnecessary bits in extensions flags: {:b}",
type_ext_socket_type
);
ExtFlags::from_bits_truncate(type_ext_socket_type)
}
};
let mut sid = None;
let mut ext_hs = None;
let mut ext_km = None;
while buf.remaining() > 4 {
let pack_type = buf.get_u16();
let pack_size_words = buf.get_u16();
let pack_size = usize::from(pack_size_words) * 4;
if buf.remaining() < pack_size {
return Err(PacketParseError::NotEnoughData);
}
let mut buffer = buf.take(pack_size);
match pack_type {
1 | 2 => {
if !extensions.contains(ExtFlags::HS) {
warn!("Handshake contains handshake extension type {} without HSREQ flag!", pack_type);
}
if ext_hs != None {
warn!("Handshake contains multiple handshake extensions, only the last will be applied!");
}
ext_hs =
Some(SrtControlPacket::parse(pack_type, &mut buffer)?);
}
3 | 4 => {
if !extensions.contains(ExtFlags::KM) {
warn!("Handshake contains key material extension type {} without KMREQ flag!", pack_type);
}
if ext_km != None {
warn!("Handshake contains multiple key material extensions, only the last will be applied!");
}
ext_km =
Some(SrtControlPacket::parse(pack_type, &mut buffer)?);
}
_ => {
if !extensions.contains(ExtFlags::CONFIG) {
warn!("Handshake contains config extension type {} without CONFIG flag!", pack_type);
}
match SrtControlPacket::parse(pack_type, &mut buffer)? {
SrtControlPacket::StreamId(stream_id) => {
sid = Some(stream_id)
}
_ => unimplemented!("Implement other kinds"),
}
}
}
buf = buffer.into_inner();
}
if buf.remaining() != 0 {
warn!("Handshake has data left, but not enough for an extension!");
}
if ext_hs.is_none() && extensions.contains(ExtFlags::HS) {
warn!("Handshake has HSREQ flag, but contains no handshake extensions!");
}
if ext_km.is_none() && extensions.contains(ExtFlags::KM) {
warn!("Handshake has KMREQ flag, but contains no key material extensions!");
}
HandshakeVsInfo::V5(HsV5Info {
crypto_size,
ext_hs,
ext_km,
ext_group: None,
sid,
})
}
}
_ => unreachable!(), };
Ok(ControlTypes::Handshake(HandshakeControlInfo {
init_seq_num,
max_packet_size,
max_flow_size,
shake_type,
socket_id,
syn_cookie,
peer_addr,
info,
}))
}
0x1 => {
if buf.remaining() >= 4 {
buf.get_u32();
}
Ok(ControlTypes::KeepAlive)
}
0x2 => {
if buf.remaining() < 4 {
return Err(PacketParseError::NotEnoughData);
}
let ack_number = SeqNumber::new_truncate(buf.get_u32());
let full_ack_seq_number = FullAckSeqNumber::new(extra_info);
let ack = if buf.remaining() < 3 * size_of::<u32>() {
Acknowledgement::Lite(ack_number)
} else {
let rtt_mean = TimeSpan::from_micros(buf.get_i32());
let rtt_variance = TimeSpan::from_micros(buf.get_i32());
let buffer_available = buf.get_u32();
let packet_receive_rate =
(buf.remaining() >= size_of::<u32>()).then(|| buf.get_u32());
let estimated_link_capacity =
(buf.remaining() >= size_of::<u32>()).then(|| buf.get_u32());
let data_receive_rate =
(buf.remaining() >= size_of::<u32>()).then(|| buf.get_u32());
let stats = AckStatistics {
rtt: Rtt::new(rtt_mean, rtt_variance),
buffer_available,
packet_receive_rate,
estimated_link_capacity,
data_receive_rate,
};
match full_ack_seq_number {
None => Acknowledgement::Small(ack_number, stats),
Some(full_ack) => Acknowledgement::Full(ack_number, stats, full_ack),
}
};
Ok(ControlTypes::Ack(ack))
}
0x3 => {
let mut loss_info = Vec::new();
while buf.remaining() >= 4 {
loss_info.push(buf.get_u32());
}
Ok(ControlTypes::Nak(CompressedLossList(loss_info)))
}
0x4 => {
if buf.remaining() >= 4 {
buf.get_u32(); }
Ok(ControlTypes::CongestionWarning)
}
0x5 => {
if buf.remaining() >= 4 {
buf.get_u32(); }
Ok(ControlTypes::Shutdown)
}
0x6 => {
if buf.remaining() >= 4 {
buf.get_u32(); }
if let Some(ack_seq_no) = FullAckSeqNumber::new(extra_info) {
Ok(ControlTypes::Ack2(ack_seq_no))
} else {
Err(PacketParseError::ZeroAckSequenceNumber)
}
}
0x7 => {
if buf.remaining() < 2 * 4 {
return Err(PacketParseError::NotEnoughData);
}
let start = SeqNumber::new_truncate(buf.get_u32());
let end = SeqNumber::new_truncate(buf.get_u32());
Ok(ControlTypes::DropRequest {
msg_to_drop: MsgNumber::new_truncate(extra_info as u32), range: RangeInclusive::new(start, end),
})
}
0x8 => {
if buf.remaining() >= 4 {
buf.get_u32(); }
Ok(ControlTypes::PeerError(extra_info))
}
0x7FFF => {
Ok(ControlTypes::Srt(SrtControlPacket::parse(
reserved, &mut buf,
)?))
}
x => Err(PacketParseError::BadControlType(x)),
}
}
fn id_byte(&self) -> u16 {
match *self {
ControlTypes::Handshake(_) => 0x0,
ControlTypes::KeepAlive => 0x1,
ControlTypes::Ack { .. } => 0x2,
ControlTypes::Nak(_) => 0x3,
ControlTypes::CongestionWarning => 0x4,
ControlTypes::Shutdown => 0x5,
ControlTypes::Ack2(_) => 0x6,
ControlTypes::DropRequest { .. } => 0x7,
ControlTypes::PeerError(_) => 0x8,
ControlTypes::Srt(_) => 0x7FFF,
}
}
fn additional_info(&self) -> u32 {
match self {
ControlTypes::DropRequest { msg_to_drop, .. } => msg_to_drop.as_raw(),
ControlTypes::Ack2(a) | ControlTypes::Ack(Acknowledgement::Full(_, _, a)) => {
(*a).into()
}
ControlTypes::PeerError(err) => *err,
ControlTypes::Ack(_)
| ControlTypes::Handshake(_)
| ControlTypes::KeepAlive
| ControlTypes::Nak(_)
| ControlTypes::CongestionWarning
| ControlTypes::Shutdown
| ControlTypes::Srt(_) => 0,
}
}
fn reserved(&self) -> u16 {
match self {
ControlTypes::Srt(srt) => srt.type_id(),
_ => 0,
}
}
fn serialize<T: BufMut>(&self, into: &mut T) {
match self {
ControlTypes::Handshake(ref c) => c.serialize(into),
ControlTypes::Ack(ack) => ack.serialize(into),
ControlTypes::Nak(ref n) => {
for loss in n.iter_compressed() {
into.put_u32(loss);
}
}
ControlTypes::DropRequest {
msg_to_drop: _,
range,
} => {
into.put_u32(range.start().as_raw());
into.put_u32(range.end().as_raw());
}
ControlTypes::CongestionWarning
| ControlTypes::Ack2(_)
| ControlTypes::Shutdown
| ControlTypes::KeepAlive
| ControlTypes::PeerError(_) => {
into.put_u32(0x0);
}
ControlTypes::Srt(srt) => {
srt.serialize(into);
}
};
}
}
impl Debug for ControlTypes {
fn fmt(&self, f: &mut Formatter) -> Result<(), fmt::Error> {
match self {
ControlTypes::Handshake(hs) => write!(f, "{:?}", hs),
ControlTypes::KeepAlive => write!(f, "KeepAlive"),
ControlTypes::Ack(aci) => write!(f, "{:?}", aci),
ControlTypes::Nak(nak) => {
write!(f, "Nak({:?})", nak) }
ControlTypes::CongestionWarning => write!(f, "CongestionWarning"),
ControlTypes::Shutdown => write!(f, "Shutdown"),
ControlTypes::Ack2(ackno) => write!(f, "Ack2({})", ackno.0),
ControlTypes::DropRequest { msg_to_drop, range } => {
write!(f, "DropReq(msg={} {:?})", msg_to_drop, range)
}
ControlTypes::PeerError(e) => write!(f, "PeerError({})", e),
ControlTypes::Srt(srt) => write!(f, "{:?}", srt),
}
}
}
impl CompressedLossList {
pub fn try_from_iter(iter: impl Iterator<Item = SeqNumber>) -> Option<CompressedLossList> {
let loss_list = compress_loss_list(iter).collect::<Vec<_>>();
if loss_list.is_empty() {
None
} else {
Some(CompressedLossList(loss_list))
}
}
pub fn try_from_range(r: Range<SeqNumber>) -> Option<CompressedLossList> {
if r.is_empty() {
None
} else if r.start + 1 == r.end {
Some(CompressedLossList(vec![r.start.as_raw()]))
} else {
Some(CompressedLossList(vec![
(1 << 31) | r.start.as_raw(),
(r.end - 1).as_raw(),
]))
}
}
pub fn iter_compressed(&self) -> impl Iterator<Item = u32> + '_ {
self.0.iter().copied()
}
pub fn iter_decompressed(&self) -> impl Iterator<Item = SeqNumber> + '_ {
decompress_loss_list(self.iter_compressed())
}
pub fn into_iter_decompressed(self) -> impl Iterator<Item = SeqNumber> {
decompress_loss_list(self.0.into_iter())
}
}
impl FromIterator<SeqNumber> for CompressedLossList {
fn from_iter<T: IntoIterator<Item = SeqNumber>>(iter: T) -> Self {
Self::try_from_iter(iter.into_iter()).unwrap()
}
}
impl<'a> FromIterator<&'a SeqNumber> for CompressedLossList {
fn from_iter<T: IntoIterator<Item = &'a SeqNumber>>(iter: T) -> Self {
Self::try_from_iter(iter.into_iter().copied()).unwrap()
}
}
impl From<Range<SeqNumber>> for CompressedLossList {
fn from(range: Range<SeqNumber>) -> Self {
Self::try_from_range(range).unwrap()
}
}
impl Debug for CompressedLossList {
fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
let mut iter = self.0.iter();
while let Some(a) = iter.next() {
if a & 0x80000000 != 0 {
let b = iter.next().expect("Unterminated list");
write!(f, "{}..={},", a & 0x7fffffff, b)?;
} else {
write!(f, "{},", a)?;
}
}
Ok(())
}
}
impl FullAckSeqNumber {
pub const INITIAL: FullAckSeqNumber = FullAckSeqNumber(1);
pub fn new(raw: u32) -> Option<FullAckSeqNumber> {
if raw == 0 {
None
} else {
Some(FullAckSeqNumber(raw))
}
}
pub fn is_full(&self) -> bool {
self.0 != 0
}
}
impl From<FullAckSeqNumber> for u32 {
fn from(u: FullAckSeqNumber) -> Self {
u.0
}
}
impl Add<u32> for FullAckSeqNumber {
type Output = FullAckSeqNumber;
fn add(self, rhs: u32) -> Self::Output {
FullAckSeqNumber(self.0 + rhs)
}
}
impl Sub<FullAckSeqNumber> for FullAckSeqNumber {
type Output = usize;
fn sub(self, rhs: FullAckSeqNumber) -> Self::Output {
(self.0 - rhs.0).try_into().unwrap()
}
}
impl Acknowledgement {
pub fn ack_number(&self) -> SeqNumber {
match self {
Acknowledgement::Lite(seq_number)
| Acknowledgement::Small(seq_number, _)
| Acknowledgement::Full(seq_number, _, _) => *seq_number,
}
}
pub fn full_ack_seq_number(&self) -> Option<FullAckSeqNumber> {
match self {
Acknowledgement::Full(_, _, full_ack_seq_number) => Some(*full_ack_seq_number),
Acknowledgement::Lite(_) | Acknowledgement::Small(_, _) => None,
}
}
pub fn rtt(&self) -> Option<Rtt> {
match self {
Acknowledgement::Full(_, stats, _) | Acknowledgement::Small(_, stats) => {
Some(stats.rtt)
}
Acknowledgement::Lite(_) => None,
}
}
pub fn statistics(&self) -> Option<&AckStatistics> {
match self {
Acknowledgement::Small(_, stats) | Acknowledgement::Full(_, stats, _) => Some(stats),
Acknowledgement::Lite(_) => None,
}
}
pub fn serialize(&self, into: &mut impl BufMut) {
into.put_u32(self.ack_number().as_raw());
if let Some(stats) = self.statistics() {
into.put_i32(stats.rtt.mean().as_micros());
into.put_i32(stats.rtt.variance().as_micros());
into.put_u32(stats.buffer_available);
if let Some(prr) = stats.packet_receive_rate {
into.put_u32(prr);
if let Some(elc) = stats.estimated_link_capacity {
into.put_u32(elc);
if let Some(drr) = stats.data_receive_rate {
into.put_u32(drr);
}
}
}
}
}
fn serialized_size(&self) -> usize {
size_of::<u32>() + self.statistics().map(AckStatistics::serialized_size).unwrap_or(0)
}
}
impl Debug for HandshakeControlInfo {
fn fmt(&self, f: &mut Formatter) -> Result<(), fmt::Error> {
write!(
f,
"HS {:?} from={:?} {:?}",
self.shake_type, self.socket_id, self.info
)
}
}
impl Debug for HandshakeVsInfo {
fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
match self {
HandshakeVsInfo::V4(stype) => write!(f, "UDT: {:?}", stype),
HandshakeVsInfo::V5(hs) => {
write!(f, "SRT: crypto={:?}", hs.crypto_size)?;
if let Some(pack) = &hs.ext_hs {
write!(f, " hs={:?}", pack)?;
}
if let Some(pack) = &hs.ext_km {
write!(f, " km={:?}", pack)?;
}
if let Some(sid) = &hs.sid {
write!(f, " sid={:?}", sid)?;
}
Ok(())
}
}
}
}
impl TryFrom<i32> for ShakeType {
type Error = i32;
fn try_from(value: i32) -> Result<Self, Self::Error> {
match value {
1 => Ok(ShakeType::Induction),
0 => Ok(ShakeType::Waveahand),
-1 => Ok(ShakeType::Conclusion),
-2 => Ok(ShakeType::Agreement),
i if i < 1000 => Err(i), i => Ok(ShakeType::Rejection(RejectReason::try_from(i).unwrap())), }
}
}
impl From<ShakeType> for i32 {
fn from(st: ShakeType) -> i32 {
match st {
ShakeType::Induction => 1,
ShakeType::Waveahand => 0,
ShakeType::Conclusion => -1,
ShakeType::Agreement => -2,
ShakeType::Rejection(rej) => rej.into(),
}
}
}
impl TryFrom<i32> for RejectReason {
type Error = i32;
fn try_from(value: i32) -> Result<Self, Self::Error> {
match value {
v if v < 1000 => Err(v),
v if v < 2000 => Ok(match CoreRejectReason::try_from(v) {
Ok(rr) => RejectReason::Core(rr),
Err(rr) => RejectReason::CoreUnrecognized(rr),
}),
v if v < 3000 => Ok(match ServerRejectReason::try_from(v) {
Ok(rr) => RejectReason::Server(rr),
Err(rr) => RejectReason::ServerUnrecognized(rr),
}),
v => Ok(RejectReason::User(v)),
}
}
}
impl From<RejectReason> for i32 {
fn from(rr: RejectReason) -> i32 {
match rr {
RejectReason::Core(c) => c.into(),
RejectReason::CoreUnrecognized(c) => c,
RejectReason::Server(s) => s.into(),
RejectReason::ServerUnrecognized(s) => s,
RejectReason::User(u) => u,
}
}
}
impl Display for RejectReason {
fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
match self {
RejectReason::Core(c) => write!(f, "{}", c),
RejectReason::CoreUnrecognized(c) => write!(f, "Unrecognized core error: {}", c),
RejectReason::Server(s) => write!(f, "{}", s),
RejectReason::ServerUnrecognized(s) => write!(f, "Unrecognized server error: {}", s),
RejectReason::User(u) => write!(f, "User error: {}", u),
}
}
}
impl From<CoreRejectReason> for RejectReason {
fn from(rr: CoreRejectReason) -> RejectReason {
RejectReason::Core(rr)
}
}
impl TryFrom<i32> for CoreRejectReason {
type Error = i32;
fn try_from(value: i32) -> Result<Self, Self::Error> {
use CoreRejectReason::*;
Ok(match value {
1001 => System,
1002 => Peer,
1003 => Resource,
1004 => Rogue,
1005 => Backlog,
1006 => Ipe,
1007 => Close,
1008 => Version,
1009 => RdvCookie,
1010 => BadSecret,
1011 => Unsecure,
1012 => MessageApi,
1013 => Congestion,
1014 => Filter,
1015 => Group,
1016 => Timeout,
other => return Err(other),
})
}
}
impl From<CoreRejectReason> for i32 {
fn from(rr: CoreRejectReason) -> i32 {
rr as i32
}
}
impl Display for CoreRejectReason {
fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
match self {
CoreRejectReason::System => write!(f, "broken due to system function error"),
CoreRejectReason::Peer => write!(f, "connection was rejected by peer"),
CoreRejectReason::Resource => write!(f, "internal problem with resource allocation"),
CoreRejectReason::Rogue => write!(f, "incorrect data in handshake messages"),
CoreRejectReason::Backlog => write!(f, "listener's backlog exceeded"),
CoreRejectReason::Ipe => write!(f, "internal program error"),
CoreRejectReason::Close => write!(f, "socket is closing"),
CoreRejectReason::Version => {
write!(f, "peer is older version than agent's minimum set")
}
CoreRejectReason::RdvCookie => write!(f, "rendezvous cookie collision"),
CoreRejectReason::BadSecret => write!(f, "wrong password"),
CoreRejectReason::Unsecure => write!(f, "password required or unexpected"),
CoreRejectReason::MessageApi => write!(f, "streamapi/messageapi collision"),
CoreRejectReason::Congestion => write!(f, "incompatible congestion-controller type"),
CoreRejectReason::Filter => write!(f, "incompatible packet filter"),
CoreRejectReason::Group => write!(f, "incompatible group"),
CoreRejectReason::Timeout => write!(f, "connection timeout"),
}
}
}
impl From<ServerRejectReason> for RejectReason {
fn from(rr: ServerRejectReason) -> RejectReason {
RejectReason::Server(rr)
}
}
impl TryFrom<i32> for ServerRejectReason {
type Error = i32;
fn try_from(value: i32) -> Result<Self, Self::Error> {
Ok(match value {
2000 => ServerRejectReason::Fallback,
2001 => ServerRejectReason::KeyNotSup,
2002 => ServerRejectReason::Filepath,
2003 => ServerRejectReason::HostNotFound,
2400 => ServerRejectReason::BadRequest,
2401 => ServerRejectReason::Unauthorized,
2402 => ServerRejectReason::Overload,
2403 => ServerRejectReason::Forbidden,
2404 => ServerRejectReason::Notfound,
2405 => ServerRejectReason::BadMode,
2406 => ServerRejectReason::Unacceptable,
2409 => ServerRejectReason::Conflict,
2415 => ServerRejectReason::NotSupMedia,
2423 => ServerRejectReason::Locked,
2424 => ServerRejectReason::FailedDepend,
2500 => ServerRejectReason::InternalServerError,
2501 => ServerRejectReason::Unimplemented,
2502 => ServerRejectReason::Gateway,
2503 => ServerRejectReason::Down,
2505 => ServerRejectReason::Version,
2507 => ServerRejectReason::NoRoom,
unrecog => return Err(unrecog),
})
}
}
impl From<ServerRejectReason> for i32 {
fn from(rr: ServerRejectReason) -> i32 {
rr as i32
}
}
impl Display for ServerRejectReason {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
ServerRejectReason::Fallback =>
write!(f, "the application wants to report some problem, but can't precisely specify it"),
ServerRejectReason::KeyNotSup =>
write!(f, "The key used in the StreamID keyed string is not supported by the service"),
ServerRejectReason::Filepath =>write!(f, "The resource type designates a file and the path is either wrong syntax or not found"),
ServerRejectReason::HostNotFound => write!(f, "The `h` host specification was not recognized by the service"),
ServerRejectReason::BadRequest => write!(f, "General syntax error in the SocketID specification (also a fallback code for undefined cases)"),
ServerRejectReason::Unauthorized => write!(f, "Authentication failed, provided that the user was correctly identified and access to the required resource would be granted"),
ServerRejectReason::Overload => write!(f, "The server is too heavily loaded, or you have exceeded credits for accessing the service and the resource"),
ServerRejectReason::Forbidden => write!(f, "Access denied to the resource by any kind of reason"),
ServerRejectReason::Notfound => write!(f, "Resource not found at this time"),
ServerRejectReason::BadMode => write!(f, "The mode specified in `m` key in StreamID is not supported for this request"),
ServerRejectReason::Unacceptable => write!(f, "The requested parameters specified in SocketID cannot be satisfied for the requested resource. Also when m=publish and the data format is not acceptable"),
ServerRejectReason::Conflict => write!(f, "The resource being accessed is already locked for modification. This is in case of m=publish and the specified resource is currently read-only"),
ServerRejectReason::NotSupMedia => write!(f, "The media type is not supported by the application. This is the `t` key that specifies the media type as stream, file and auth, possibly extended by the application"),
ServerRejectReason::Locked => write!(f, "The resource being accessed is locked for any access"),
ServerRejectReason::FailedDepend => write!(f, "The request failed because it specified a dependent session ID that has been disconnected"),
ServerRejectReason::InternalServerError => write!(f, "Unexpected internal server error"),
ServerRejectReason::Unimplemented => write!(f, "The request was recognized, but the current version doesn't support it (unimplemented)"),
ServerRejectReason::Gateway => write!(f, "The server acts as a gateway and the target endpoint rejected the connection"),
ServerRejectReason::Down => write!(f, "The service has been temporarily taken over by a stub reporting this error. The real service can be down for maintenance or crashed"),
ServerRejectReason::Version => write!(f, "SRT version not supported. This might be either unsupported backward compatibility, or an upper value of a version"),
ServerRejectReason::NoRoom => write!(f, "The data stream cannot be archived due to lacking storage space. This is in case when the request type was to send a file or the live stream to be archived"),
}
}
}
impl HandshakeControlInfo {
fn serialized_size(&self) -> usize {
8 * size_of::<u32>() + size_of::<u128>() + match &self.info {
HandshakeVsInfo::V4(_) => 0,
HandshakeVsInfo::V5(info) => {
info.ext_hs.as_ref().map(|hs| 2 * size_of::<u16>() + usize::from(hs.size_words()) * size_of::<u32>()).unwrap_or(0)
+
info.ext_km.as_ref().map(|hs| 2 * size_of::<u16>() + usize::from(hs.size_words()) * size_of::<u32>()).unwrap_or(0)
+
info.sid.as_ref().map(|sid| 2 * size_of::<u16>() + ((sid.len() + 3) / 4 * 4)).unwrap_or(0)
}
}
}
fn serialize(&self, into: &mut impl BufMut) {
into.put_u32(self.info.version());
into.put_u32(self.info.type_flags(self.shake_type));
into.put_u32(self.init_seq_num.as_raw());
into.put_u32(self.max_packet_size.0 as u32);
into.put_u32(self.max_flow_size.0 as u32);
into.put_i32(self.shake_type.into());
into.put_u32(self.socket_id.0);
into.put_i32(self.syn_cookie);
match self.peer_addr {
IpAddr::V4(four) => {
let v = u32::from(four);
into.put_u32_le(v);
into.put(&[0; 12][..]);
}
IpAddr::V6(six) => {
let v = u128::from(six);
into.put_u128(v);
}
}
if let HandshakeVsInfo::V5(hs) = &self.info {
for ext in IntoIter::new([
&hs.ext_hs,
&hs.ext_km,
&hs.sid.clone().map(SrtControlPacket::StreamId),
])
.filter_map(|s| s.as_ref())
{
into.put_u16(ext.type_id());
into.put_u16(ext.size_words());
ext.serialize(into);
}
}
}
}
impl AckStatistics {
fn serialized_size(&self) -> usize {
size_of::<u32>()
* (3 + match self {
Self {
packet_receive_rate: None,
..
} => 0,
Self {
estimated_link_capacity: None,
..
} => 1,
Self {
data_receive_rate: None,
..
} => 2,
_ => 3,
})
}
}
#[cfg(test)]
mod test {
use super::*;
use std::{array::IntoIter, time::Duration};
use std::{convert::TryInto, io::Cursor};
use crate::options::*;
fn ser_des_test(pack: ControlPacket) -> Vec<u8> {
let mut buf = vec![];
pack.serialize(&mut buf);
let mut cursor = Cursor::new(&buf);
let des = ControlPacket::parse(&mut cursor, false).unwrap();
assert_eq!(cursor.remaining(), 0);
assert_eq!(pack, des);
assert_eq!(
pack.wire_size(),
buf.len() + 28,
"Packet {:?} wrong wire size",
pack
);
buf
}
#[test]
fn lite_ack_ser_des_test() {
ser_des_test(ControlPacket {
timestamp: TimeStamp::from_micros(1234),
dest_sockid: SocketId(0),
control_type: ControlTypes::Ack(Acknowledgement::Lite(SeqNumber::new_truncate(1234))),
});
}
#[test]
fn handshake_ser_des_test() {
ser_des_test(ControlPacket {
timestamp: TimeStamp::from_micros(0),
dest_sockid: SocketId(0),
control_type: ControlTypes::Handshake(HandshakeControlInfo {
init_seq_num: SeqNumber::new_truncate(1_827_131),
max_packet_size: PacketSize(1500),
max_flow_size: PacketCount(25600),
shake_type: ShakeType::Conclusion,
socket_id: SocketId(1231),
syn_cookie: 0,
peer_addr: "127.0.0.1".parse().unwrap(),
info: HandshakeVsInfo::V5(HsV5Info {
crypto_size: 0, ext_hs: Some(SrtControlPacket::HandshakeResponse(SrtHandshake {
version: SrtVersion::CURRENT,
flags: SrtShakeFlags::NAKREPORT | SrtShakeFlags::TSBPDSND,
send_latency: Duration::from_millis(3000),
recv_latency: Duration::from_millis(12345),
})),
ext_km: None,
ext_group: None,
sid: None,
}),
}),
});
}
#[test]
fn ack_ser_des_test() {
ser_des_test(ControlPacket {
timestamp: TimeStamp::from_micros(113_703),
dest_sockid: SocketId(2_453_706_529),
control_type: ControlTypes::Ack(Acknowledgement::Full(
SeqNumber::new_truncate(282_049_186),
AckStatistics {
rtt: Rtt::new(TimeSpan::from_micros(10_002), TimeSpan::from_micros(1000)),
buffer_available: 1314,
packet_receive_rate: Some(0),
estimated_link_capacity: Some(0),
data_receive_rate: Some(0),
},
FullAckSeqNumber::new(1).unwrap(),
)),
});
}
#[test]
fn ack2_ser_des_test() {
let buf = ser_des_test(ControlPacket {
timestamp: TimeStamp::from_micros(125_812),
dest_sockid: SocketId(8313),
control_type: ControlTypes::Ack2(FullAckSeqNumber::new(831).unwrap()),
});
assert_eq!((u32::from(buf[6]) << 8) + u32::from(buf[7]), 831);
}
#[test]
fn enc_size_ser_des_test() {
ser_des_test(ControlPacket {
timestamp: TimeStamp::from_micros(0),
dest_sockid: SocketId(0),
control_type: ControlTypes::Handshake(HandshakeControlInfo {
init_seq_num: SeqNumber(0),
max_packet_size: PacketSize(1816),
max_flow_size: PacketCount(0),
shake_type: ShakeType::Conclusion,
socket_id: SocketId(0),
syn_cookie: 0,
peer_addr: [127, 0, 0, 1].into(),
info: HandshakeVsInfo::V5(HsV5Info {
crypto_size: 16,
ext_km: None,
ext_hs: None,
ext_group: None,
sid: None,
}),
}),
});
}
#[test]
fn sid_ser_des_test() {
ser_des_test(ControlPacket {
timestamp: TimeStamp::from_micros(0),
dest_sockid: SocketId(0),
control_type: ControlTypes::Handshake(HandshakeControlInfo {
init_seq_num: SeqNumber(0),
max_packet_size: PacketSize(1816),
max_flow_size: PacketCount(0),
shake_type: ShakeType::Conclusion,
socket_id: SocketId(0),
syn_cookie: 0,
peer_addr: [127, 0, 0, 1].into(),
info: HandshakeVsInfo::V5(HsV5Info {
crypto_size: 0,
ext_km: None,
ext_hs: None,
ext_group: None,
sid: Some("Hello hello".into()),
}),
}),
});
}
#[test]
fn keepalive_ser_des_test() {
ser_des_test(ControlPacket {
timestamp: TimeStamp::from_micros(0),
dest_sockid: SocketId(0),
control_type: ControlTypes::KeepAlive,
});
}
#[test]
fn congestion_warning_ser_des_test() {
ser_des_test(ControlPacket {
timestamp: TimeStamp::from_micros(100),
dest_sockid: rand::random(),
control_type: ControlTypes::CongestionWarning,
});
}
#[test]
fn peer_error_ser_des_test() {
ser_des_test(ControlPacket {
timestamp: TimeStamp::from_micros(100),
dest_sockid: rand::random(),
control_type: ControlTypes::PeerError(1234),
});
}
#[test]
fn congestion_ser_des_test() {
ser_des_test(ControlPacket {
timestamp: TimeStamp::from_micros(100),
dest_sockid: rand::random(),
control_type: ControlTypes::Srt(SrtControlPacket::Congestion("live".to_string())),
});
}
#[test]
fn group_ser_des_test() {
ser_des_test(ControlPacket {
timestamp: TimeStamp::from_micros(100),
dest_sockid: rand::random(),
control_type: ControlTypes::Srt(SrtControlPacket::Group {
ty: GroupType::MainBackup,
flags: GroupFlags::MSG_SYNC,
weight: 123,
}),
});
}
#[test]
fn drop_request_ser_des_test() {
ser_des_test(ControlPacket {
timestamp: TimeStamp::from_micros(100),
dest_sockid: rand::random(),
control_type: ControlTypes::DropRequest {
msg_to_drop: MsgNumber(123),
range: SeqNumber(13)..=SeqNumber(100),
},
});
}
#[test]
fn filter_ser_des_test() {
ser_des_test(ControlPacket {
timestamp: TimeStamp::from_micros(100),
dest_sockid: rand::random(),
control_type: ControlTypes::Srt(SrtControlPacket::Filter(FilterSpec(
IntoIter::new([("hi".to_string(), "bye".to_string())]).collect(),
))),
});
}
#[test]
fn raw_srt_packet_test() {
let packet_data =
hex::decode("FFFF000000000000000189702BFFEFF2000103010000001E00000078").unwrap();
let packet = ControlPacket::parse(&mut Cursor::new(packet_data), false).unwrap();
assert_eq!(
packet,
ControlPacket {
timestamp: TimeStamp::from_micros(100_720),
dest_sockid: SocketId(738_193_394),
control_type: ControlTypes::Srt(SrtControlPacket::Reject)
}
)
}
#[test]
fn raw_handshake_ipv6() {
let packet_data = hex::decode("8000000000000000000002b00000000000000004000000023c3b0296000005dc00002000000000010669ead20000000000000000000000000000000001000000").unwrap();
let packet = ControlPacket::parse(&mut Cursor::new(&packet_data[..]), true).unwrap();
let r = ControlPacket {
timestamp: TimeStamp::from_micros(688),
dest_sockid: SocketId(0),
control_type: ControlTypes::Handshake(HandshakeControlInfo {
init_seq_num: SeqNumber(1010500246),
max_packet_size: PacketSize(1500),
max_flow_size: PacketCount(8192),
shake_type: ShakeType::Induction,
socket_id: SocketId(0x0669EAD2),
syn_cookie: 0,
peer_addr: "::1.0.0.0".parse().unwrap(),
info: HandshakeVsInfo::V4(SocketType::Datagram),
}),
};
assert_eq!(packet, r);
let mut buf = vec![];
packet.serialize(&mut buf);
assert_eq!(&buf[..], &packet_data[..]);
}
#[test]
fn raw_handshake_srt() {
let packet_data = hex::decode("8000000000000000000F9EC400000000000000050000000144BEA60D000005DC00002000FFFFFFFF3D6936B6E3E405DD0100007F00000000000000000000000000010003000103010000002F00780000").unwrap();
let packet = ControlPacket::parse(&mut Cursor::new(&packet_data[..]), false).unwrap();
assert_eq!(
packet,
ControlPacket {
timestamp: TimeStamp::from_micros(1_023_684),
dest_sockid: SocketId(0),
control_type: ControlTypes::Handshake(HandshakeControlInfo {
init_seq_num: SeqNumber(1_153_345_037),
max_packet_size: PacketSize(1500),
max_flow_size: PacketCount(8192),
shake_type: ShakeType::Conclusion,
socket_id: SocketId(1_030_305_462),
syn_cookie: -471_595_555,
peer_addr: "127.0.0.1".parse().unwrap(),
info: HandshakeVsInfo::V5(HsV5Info {
crypto_size: 0,
ext_hs: Some(SrtControlPacket::HandshakeRequest(SrtHandshake {
version: SrtVersion::new(1, 3, 1),
flags: SrtShakeFlags::TSBPDSND
| SrtShakeFlags::TSBPDRCV
| SrtShakeFlags::HAICRYPT
| SrtShakeFlags::TLPKTDROP
| SrtShakeFlags::REXMITFLG,
send_latency: Duration::from_millis(120),
recv_latency: Duration::new(0, 0)
})),
ext_km: None,
ext_group: None,
sid: None,
})
})
}
);
let mut buf = vec![];
packet.serialize(&mut buf);
assert_eq!(&buf[..], &packet_data[..]);
}
#[test]
fn raw_handshake_sid() {
let packet_data = hex::decode("800000000000000000000b1400000000000000050000000563444b2e000005dc00002000ffffffff37eb0ee52154fbd60100007f0000000000000000000000000001000300010401000000bf0014001400050003646362616867666500006a69").unwrap();
let packet = ControlPacket::parse(&mut Cursor::new(&packet_data[..]), false).unwrap();
assert_eq!(
packet,
ControlPacket {
timestamp: TimeStamp::from_micros(2836),
dest_sockid: SocketId(0),
control_type: ControlTypes::Handshake(HandshakeControlInfo {
init_seq_num: SeqNumber(1_665_420_078),
max_packet_size: PacketSize(1500),
max_flow_size: PacketCount(8192),
shake_type: ShakeType::Conclusion,
socket_id: SocketId(0x37eb0ee5),
syn_cookie: 559_217_622,
peer_addr: "127.0.0.1".parse().unwrap(),
info: HandshakeVsInfo::V5(HsV5Info {
crypto_size: 0,
ext_hs: Some(SrtControlPacket::HandshakeRequest(SrtHandshake {
version: SrtVersion::new(1, 4, 1),
flags: SrtShakeFlags::TSBPDSND
| SrtShakeFlags::TSBPDRCV
| SrtShakeFlags::HAICRYPT
| SrtShakeFlags::REXMITFLG
| SrtShakeFlags::TLPKTDROP
| SrtShakeFlags::NAKREPORT
| SrtShakeFlags::PACKET_FILTER,
send_latency: Duration::from_millis(20),
recv_latency: Duration::from_millis(20)
})),
ext_km: None,
ext_group: None,
sid: Some(String::from("abcdefghij")),
})
})
}
);
let mut buf = vec![];
packet.serialize(&mut buf);
assert_eq!(&buf[..], &packet_data[..]);
}
#[test]
fn raw_handshake_crypto() {
let packet_data = hex::decode("800000000000000000175E8A0000000000000005000000036FEFB8D8000005DC00002000FFFFFFFF35E790ED5D16CCEA0100007F00000000000000000000000000010003000103010000002F01F401F40003000E122029010000000002000200000004049D75B0AC924C6E4C9EC40FEB4FE973DB1D215D426C18A2871EBF77E2646D9BAB15DBD7689AEF60EC").unwrap();
let packet = ControlPacket::parse(&mut Cursor::new(&packet_data[..]), false).unwrap();
assert_eq!(
packet,
ControlPacket {
timestamp: TimeStamp::from_micros(1_531_530),
dest_sockid: SocketId(0),
control_type: ControlTypes::Handshake(HandshakeControlInfo {
init_seq_num: SeqNumber(1_877_981_400),
max_packet_size: PacketSize(1_500),
max_flow_size: PacketCount(8_192),
shake_type: ShakeType::Conclusion,
socket_id: SocketId(904_368_365),
syn_cookie: 1_561_775_338,
peer_addr: "127.0.0.1".parse().unwrap(),
info: HandshakeVsInfo::V5(HsV5Info {
crypto_size: 0,
ext_hs: Some(SrtControlPacket::HandshakeRequest(SrtHandshake {
version: SrtVersion::new(1, 3, 1),
flags: SrtShakeFlags::TSBPDSND
| SrtShakeFlags::TSBPDRCV
| SrtShakeFlags::HAICRYPT
| SrtShakeFlags::TLPKTDROP
| SrtShakeFlags::REXMITFLG,
send_latency: Duration::from_millis(500),
recv_latency: Duration::from_millis(500)
})),
ext_km: Some(SrtControlPacket::KeyRefreshRequest(KeyingMaterialMessage {
pt: PacketType::KeyingMaterial,
key_flags: KeyFlags::EVEN,
keki: 0,
cipher: CipherType::Ctr,
auth: Auth::None,
salt: hex::decode("9D75B0AC924C6E4C9EC40FEB4FE973DB").unwrap(),
wrapped_keys: hex::decode(
"1D215D426C18A2871EBF77E2646D9BAB15DBD7689AEF60EC"
)
.unwrap()
})),
ext_group: None,
sid: None,
})
})
}
);
let mut buf = vec![];
packet.serialize(&mut buf);
assert_eq!(&buf[..], &packet_data[..])
}
#[test]
fn raw_handshake_crypto_pt2() {
let packet_data = hex::decode("8000000000000000000000000C110D94000000050000000374B7526E000005DC00002000FFFFFFFF18C1CED1F3819B720100007F00000000000000000000000000020003000103010000003F03E803E80004000E12202901000000000200020000000404D3B3D84BE1188A4EBDA4DA16EA65D522D82DE544E1BE06B6ED8128BF15AA4E18EC50EAA95546B101").unwrap();
let _packet = ControlPacket::parse(&mut Cursor::new(&packet_data[..]), false).unwrap();
dbg!(&_packet);
}
#[test]
fn short_ack() {
let packet_data =
hex::decode("800200000000000e000246e5d96d5e1a389c24780000452900007bb000001fa9")
.unwrap();
let _cp = ControlPacket::parse(&mut Cursor::new(packet_data), false).unwrap();
}
#[test]
fn test_reject_reason_deser_ser() {
assert_eq!(
Ok(RejectReason::Server(ServerRejectReason::Unimplemented)),
<i32 as TryInto<RejectReason>>::try_into(
RejectReason::Server(ServerRejectReason::Unimplemented).into()
)
);
}
#[test]
fn test_unordered_hs_extensions() {
let packet_data = hex::decode(concat!(
"80000000000000000000dea800000000",
"000000050004000751dca3b8000005b8",
"00002000ffffffff025c84b8da7ee4e7",
"0100007f000000000000000000000000",
"0001000300010402000000bf003c003c",
"000500033a3a212365683d7500000078",
"00030012122029010000000002000200",
"00000408437937d8c23ce2090754c5a7",
"a9e608c14631aef7ac0b8a46b77b8c0b",
"97d4061e565dcb86e4c5cc3701e1f992",
"a5b2de3651c937c94f3333a6"
))
.unwrap();
let packet = ControlPacket::parse(&mut Cursor::new(packet_data), false).unwrap();
let reference = ControlPacket {
timestamp: TimeStamp::from_micros(57000),
dest_sockid: SocketId(0),
control_type: ControlTypes::Handshake(HandshakeControlInfo {
init_seq_num: SeqNumber(1373414328),
max_packet_size: PacketSize(1464),
max_flow_size: PacketCount(8192),
shake_type: ShakeType::Conclusion,
socket_id: SocketId(0x025C84B8),
syn_cookie: 0xda7ee4e7u32 as i32,
peer_addr: [127, 0, 0, 1].into(),
info: HandshakeVsInfo::V5(HsV5Info {
crypto_size: 32,
ext_hs: Some(SrtControlPacket::HandshakeRequest(SrtHandshake {
version: SrtVersion::new(1, 4, 2),
flags: SrtShakeFlags::TSBPDSND
| SrtShakeFlags::TSBPDRCV
| SrtShakeFlags::HAICRYPT
| SrtShakeFlags::TLPKTDROP
| SrtShakeFlags::NAKREPORT
| SrtShakeFlags::REXMITFLG
| SrtShakeFlags::PACKET_FILTER,
send_latency: Duration::from_millis(60),
recv_latency: Duration::from_millis(60)
})),
ext_km: Some(SrtControlPacket::KeyRefreshRequest(KeyingMaterialMessage {
pt: PacketType::KeyingMaterial,
key_flags: KeyFlags::EVEN,
keki: 0,
cipher: CipherType::Ctr,
auth: Auth::None,
salt: hex::decode("437937d8c23ce2090754c5a7a9e608c1").unwrap(),
wrapped_keys: hex::decode(
"4631aef7ac0b8a46b77b8c0b97d4061e565dcb86e4c5cc3701e1f992a5b2de3651c937c94f3333a6"
)
.unwrap()
})),
ext_group: None,
sid: Some("#!::u=hex".into()),
}),
}),
};
assert_eq!(packet, reference);
}
#[test]
fn drop_request() {}
}